From a01a3c15709d34eea32d65c4e47b010e91695001 Mon Sep 17 00:00:00 2001 From: lihe Date: Tue, 1 Sep 2026 14:13:15 +0800 Subject: [PATCH 1/7] feat: add CometBFT rewind command Signed-off-by: lihe --- CHANGELOG.md | 1 + cmd/cronosd/cmd/rewind_cometbft.go | 128 ++++++++++++++++++++++++ cmd/cronosd/cmd/rewind_cometbft_test.go | 106 ++++++++++++++++++++ cmd/cronosd/cmd/root.go | 2 +- 4 files changed, 236 insertions(+), 1 deletion(-) create mode 100644 cmd/cronosd/cmd/rewind_cometbft.go create mode 100644 cmd/cronosd/cmd/rewind_cometbft_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d1ef35..2e03e26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ### Improvements +* Add `rewind-cometbft` for removing the CometBFT block and consensus-state suffix above a recovery height. * [#9](https://github.com/Chaintable/cronos/pull/9) feat(blocksync): add an opt-in continuous sync mode for non-validator nodes. * [#2095](https://github.com/crypto-org-chain/cronos/pull/2095) feat: add hardcoded unblockable address list that is filtered out of both the mempool and consensus blocklists. diff --git a/cmd/cronosd/cmd/rewind_cometbft.go b/cmd/cronosd/cmd/rewind_cometbft.go new file mode 100644 index 0000000..c187caf --- /dev/null +++ b/cmd/cronosd/cmd/rewind_cometbft.go @@ -0,0 +1,128 @@ +package cmd + +import ( + "fmt" + "strconv" + + cmtcfg "github.com/cometbft/cometbft/config" + cmtstate "github.com/cometbft/cometbft/state" + cmtstore "github.com/cometbft/cometbft/store" + "github.com/cosmos/cosmos-sdk/server" + "github.com/spf13/cobra" +) + +const rewindProgressInterval int64 = 1000 + +// RewindCometBFTCmd removes the CometBFT block and consensus-state suffix above +// a target height. Application and index databases are intentionally handled by +// separate recovery steps. +func RewindCometBFTCmd() *cobra.Command { + return &cobra.Command{ + Use: "rewind-cometbft [height]", + Short: "Rewind CometBFT blocks and consensus state to a specific height", + Long: `Rewind CometBFT blocks and consensus state to a specific height. + +This command always deletes blocks above the target. It does not modify the +application, transaction-index, or EVM-index databases. Stop cronosd before +running it and rewind those databases as separate recovery steps.`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + target, err := strconv.ParseInt(args[0], 10, 64) + if err != nil { + return fmt.Errorf("parse target height: %w", err) + } + if target <= 0 { + return fmt.Errorf("target height must be positive: %d", target) + } + + ctx := server.GetServerContextFromCmd(cmd) + cfg := ctx.Config + + blockDB, err := cmtcfg.DefaultDBProvider(&cmtcfg.DBContext{ID: "blockstore", Config: cfg}) + if err != nil { + return fmt.Errorf("open blockstore: %w", err) + } + defer blockDB.Close() + blockStore := cmtstore.NewBlockStore(blockDB) + + stateDB, err := cmtcfg.DefaultDBProvider(&cmtcfg.DBContext{ID: "state", Config: cfg}) + if err != nil { + return fmt.Errorf("open state store: %w", err) + } + defer stateDB.Close() + stateStore := cmtstate.NewStore(stateDB, cmtstate.StoreOptions{ + DiscardABCIResponses: cfg.Storage.DiscardABCIResponses, + }) + + initialState, err := stateStore.Load() + if err != nil { + return fmt.Errorf("load CometBFT state: %w", err) + } + cmd.Printf("rewinding CometBFT state=%d blockstore=%d target=%d\n", + initialState.LastBlockHeight, blockStore.Height(), target) + + height, hash, removed, err := rewindCometBFT(blockStore, stateStore, target, func(height, removed int64) { + if removed%rewindProgressInterval == 0 { + cmd.Printf("rewind progress height=%d removed=%d\n", height, removed) + } + }) + if err != nil { + return err + } + + cmd.Printf("rewound CometBFT to height %d app_hash=%X removed=%d\n", height, hash, removed) + return nil + }, + } +} + +func rewindCometBFT( + blockStore *cmtstore.BlockStore, + stateStore cmtstate.Store, + target int64, + progress func(height, removed int64), +) (int64, []byte, int64, error) { + if target <= 0 { + return 0, nil, 0, fmt.Errorf("target height must be positive: %d", target) + } + if base := blockStore.Base(); base > 0 && target < base { + return 0, nil, 0, fmt.Errorf("target %d is below blockstore base %d", target, base) + } + + var removed int64 + for { + currentState, err := stateStore.Load() + if err != nil { + return 0, nil, removed, fmt.Errorf("load CometBFT state: %w", err) + } + if currentState.IsEmpty() { + return 0, nil, removed, fmt.Errorf("no CometBFT state found") + } + + stateHeight := currentState.LastBlockHeight + blockHeight := blockStore.Height() + if stateHeight < target || blockHeight < target { + return 0, nil, removed, fmt.Errorf( + "target %d is above available state=%d or blockstore=%d", target, stateHeight, blockHeight, + ) + } + if stateHeight == target && blockHeight == target { + return target, currentState.AppHash, removed, nil + } + + height, _, err := cmtstate.Rollback(blockStore, stateStore, true) + if err != nil { + return 0, nil, removed, fmt.Errorf( + "rewind CometBFT from state=%d blockstore=%d: %w", stateHeight, blockHeight, err, + ) + } + if height < target { + return 0, nil, removed, fmt.Errorf("rewind crossed target %d and reached %d", target, height) + } + + removed++ + if progress != nil { + progress(height, removed) + } + } +} diff --git a/cmd/cronosd/cmd/rewind_cometbft_test.go b/cmd/cronosd/cmd/rewind_cometbft_test.go new file mode 100644 index 0000000..7e477e1 --- /dev/null +++ b/cmd/cronosd/cmd/rewind_cometbft_test.go @@ -0,0 +1,106 @@ +package cmd + +import ( + "fmt" + "testing" + "time" + + dbm "github.com/cometbft/cometbft-db" + cmtcrypto "github.com/cometbft/cometbft/crypto" + "github.com/cometbft/cometbft/crypto/tmhash" + cmtstateproto "github.com/cometbft/cometbft/proto/tendermint/state" + cmtversion "github.com/cometbft/cometbft/proto/tendermint/version" + cmtstate "github.com/cometbft/cometbft/state" + cmtstore "github.com/cometbft/cometbft/store" + cmttypes "github.com/cometbft/cometbft/types" + "github.com/cometbft/cometbft/version" + "github.com/stretchr/testify/require" +) + +func TestRewindCometBFT(t *testing.T) { + blockStore := cmtstore.NewBlockStore(dbm.NewMemDB()) + stateStore := cmtstate.NewStore(dbm.NewMemDB(), cmtstate.StoreOptions{}) + validatorSet, _ := cmttypes.RandValidatorSet(1, 10) + params := cmttypes.DefaultConsensusParams() + + var lastBlockID cmttypes.BlockID + for height := int64(1); height <= 5; height++ { + block := &cmttypes.Block{ + Header: cmttypes.Header{ + Version: cmtversion.Consensus{Block: version.BlockProtocol, App: 1}, + ChainID: "rewind-test", + Height: height, + Time: time.Unix(height, 0).UTC(), + LastBlockID: lastBlockID, + LastCommitHash: cmtcrypto.CRandBytes(tmhash.Size), + DataHash: cmtcrypto.CRandBytes(tmhash.Size), + ValidatorsHash: validatorSet.Hash(), + NextValidatorsHash: validatorSet.Hash(), + ConsensusHash: params.Hash(), + AppHash: testAppHash(height - 1), + LastResultsHash: tmhash.Sum([]byte(fmt.Sprintf("results-%d", height-1))), + EvidenceHash: cmtcrypto.CRandBytes(tmhash.Size), + ProposerAddress: cmtcrypto.CRandBytes(cmtcrypto.AddressSize), + }, + LastCommit: &cmttypes.Commit{Height: height - 1}, + } + parts, err := block.MakePartSet(cmttypes.BlockPartSizeBytes) + require.NoError(t, err) + blockStore.SaveBlock(block, parts, &cmttypes.Commit{Height: height}) + lastBlockID = blockStore.LoadBlockMeta(height).BlockID + + state := cmtstate.State{ + Version: cmtstateproto.Version{ + Consensus: block.Version, + Software: version.TMCoreSemVer, + }, + ChainID: block.ChainID, + InitialHeight: 1, + LastBlockHeight: height, + LastBlockID: lastBlockID, + LastBlockTime: block.Time, + NextValidators: validatorSet, + Validators: validatorSet, + LastValidators: validatorSet, + LastHeightValidatorsChanged: 2, + ConsensusParams: *params, + LastHeightConsensusParamsChanged: 2, + LastResultsHash: tmhash.Sum([]byte(fmt.Sprintf("results-%d", height))), + AppHash: testAppHash(height), + } + if height == 1 { + require.NoError(t, stateStore.Bootstrap(state)) + } else { + require.NoError(t, stateStore.Save(state)) + } + } + + var progressCalls int + height, hash, removed, err := rewindCometBFT(blockStore, stateStore, 2, func(_, _ int64) { + progressCalls++ + }) + require.NoError(t, err) + require.EqualValues(t, 2, height) + require.Equal(t, testAppHash(2), hash) + require.EqualValues(t, 3, removed) + require.Equal(t, 3, progressCalls) + require.EqualValues(t, 2, blockStore.Height()) + require.Nil(t, blockStore.LoadBlockMeta(3)) + + state, err := stateStore.Load() + require.NoError(t, err) + require.EqualValues(t, 2, state.LastBlockHeight) + require.Equal(t, testAppHash(2), state.AppHash) +} + +func TestRewindCometBFTRejectsFutureTarget(t *testing.T) { + blockStore := cmtstore.NewBlockStore(dbm.NewMemDB()) + stateStore := cmtstate.NewStore(dbm.NewMemDB(), cmtstate.StoreOptions{}) + + _, _, _, err := rewindCometBFT(blockStore, stateStore, 2, nil) + require.ErrorContains(t, err, "no CometBFT state found") +} + +func testAppHash(height int64) []byte { + return tmhash.Sum([]byte(fmt.Sprintf("app-%d", height))) +} diff --git a/cmd/cronosd/cmd/root.go b/cmd/cronosd/cmd/root.go index 42c7abb..22515b7 100644 --- a/cmd/cronosd/cmd/root.go +++ b/cmd/cronosd/cmd/root.go @@ -184,7 +184,7 @@ func initRootCmd( rootCmd.AddCommand(changeSetCmd) } - rootCmd.AddCommand(RollbackToVersionCmd()) + rootCmd.AddCommand(RollbackToVersionCmd(), RewindCometBFTCmd()) // add keybase, auxiliary RPC, query, and tx child commands rootCmd.AddCommand( From 309a4e03840df2551c464be697032bea7af35d54 Mon Sep 17 00:00:00 2001 From: lihe Date: Tue, 1 Sep 2026 14:31:15 +0800 Subject: [PATCH 2/7] feat: add index rewind command Signed-off-by: lihe --- CHANGELOG.md | 2 +- cmd/cronosd/cmd/rewind_cometbft_test.go | 2 +- cmd/cronosd/cmd/rewind_indexes.go | 280 ++++++++++++++++++++++++ cmd/cronosd/cmd/rewind_indexes_test.go | 104 +++++++++ cmd/cronosd/cmd/root.go | 2 +- 5 files changed, 387 insertions(+), 3 deletions(-) create mode 100644 cmd/cronosd/cmd/rewind_indexes.go create mode 100644 cmd/cronosd/cmd/rewind_indexes_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e03e26..55b96f7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ ### Improvements -* Add `rewind-cometbft` for removing the CometBFT block and consensus-state suffix above a recovery height. +* Add recovery commands for removing CometBFT block/state and transaction-index suffixes above a target height. * [#9](https://github.com/Chaintable/cronos/pull/9) feat(blocksync): add an opt-in continuous sync mode for non-validator nodes. * [#2095](https://github.com/crypto-org-chain/cronos/pull/2095) feat: add hardcoded unblockable address list that is filtered out of both the mempool and consensus blocklists. diff --git a/cmd/cronosd/cmd/rewind_cometbft_test.go b/cmd/cronosd/cmd/rewind_cometbft_test.go index 7e477e1..c90b64a 100644 --- a/cmd/cronosd/cmd/rewind_cometbft_test.go +++ b/cmd/cronosd/cmd/rewind_cometbft_test.go @@ -93,7 +93,7 @@ func TestRewindCometBFT(t *testing.T) { require.Equal(t, testAppHash(2), state.AppHash) } -func TestRewindCometBFTRejectsFutureTarget(t *testing.T) { +func TestRewindCometBFTRejectsMissingState(t *testing.T) { blockStore := cmtstore.NewBlockStore(dbm.NewMemDB()) stateStore := cmtstate.NewStore(dbm.NewMemDB(), cmtstate.StoreOptions{}) diff --git a/cmd/cronosd/cmd/rewind_indexes.go b/cmd/cronosd/cmd/rewind_indexes.go new file mode 100644 index 0000000..aa28487 --- /dev/null +++ b/cmd/cronosd/cmd/rewind_indexes.go @@ -0,0 +1,280 @@ +package cmd + +import ( + "bytes" + "encoding/binary" + "fmt" + "strconv" + + cmtdb "github.com/cometbft/cometbft-db" + cmtabci "github.com/cometbft/cometbft/abci/types" + cmtcfg "github.com/cometbft/cometbft/config" + cosmosdb "github.com/cosmos/cosmos-db" + "github.com/cosmos/cosmos-sdk/server" + ethermintserver "github.com/evmos/ethermint/server" + "github.com/spf13/cobra" + + "github.com/cosmos/gogoproto/proto" +) + +const ( + deleteBatchSize = 10_000 + indexScanInterval = 1_000_000 +) + +var ( + blockEventPrefix = []byte("block_events") + evmTxHashPrefix = byte(1) + evmTxIndexPrefix = byte(2) +) + +type indexTrimStats struct { + Scanned int64 + TxEventKeys int64 + TxHashKeys int64 + BlockEventKeys int64 + EVMTxIndexKeys int64 + EVMTxHashKeys int64 +} + +// RewindIndexesCmd removes transaction and block-event index entries above a +// target height. It is idempotent and must be run with cronosd stopped. +func RewindIndexesCmd() *cobra.Command { + return &cobra.Command{ + Use: "rewind-indexes [height]", + Short: "Remove CometBFT and EVM index entries above a specific height", + Long: "Remove CometBFT transaction/block-event and EVM transaction index " + + "entries above a specific height. Stop cronosd before running this command.", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + target, err := strconv.ParseInt(args[0], 10, 64) + if err != nil { + return fmt.Errorf("parse target height: %w", err) + } + if target <= 0 { + return fmt.Errorf("target height must be positive: %d", target) + } + + ctx := server.GetServerContextFromCmd(cmd) + cfg := ctx.Config + + txIndexDB, err := cmtcfg.DefaultDBProvider(&cmtcfg.DBContext{ID: "tx_index", Config: cfg}) + if err != nil { + return fmt.Errorf("open CometBFT index database: %w", err) + } + stats, err := trimCometIndexes(txIndexDB, target, func(scanned int64) { + if scanned%indexScanInterval == 0 { + cmd.Printf("CometBFT index scan progress keys=%d\n", scanned) + } + }) + closeErr := txIndexDB.Close() + if err != nil { + return err + } + if closeErr != nil { + return fmt.Errorf("close CometBFT index database: %w", closeErr) + } + + evmIndexDB, err := ethermintserver.OpenIndexerDB(cfg.RootDir, server.GetAppDBBackend(ctx.Viper)) + if err != nil { + return fmt.Errorf("open EVM index database: %w", err) + } + evmStats, err := trimEVMIndexes(evmIndexDB, target) + closeErr = evmIndexDB.Close() + if err != nil { + return err + } + if closeErr != nil { + return fmt.Errorf("close EVM index database: %w", closeErr) + } + + stats.EVMTxIndexKeys = evmStats.EVMTxIndexKeys + stats.EVMTxHashKeys = evmStats.EVMTxHashKeys + cmd.Printf( + "rewound indexes to height %d scanned=%d tx_events=%d tx_hashes=%d block_events=%d evm_tx_indexes=%d evm_tx_hashes=%d\n", + target, + stats.Scanned, + stats.TxEventKeys, + stats.TxHashKeys, + stats.BlockEventKeys, + stats.EVMTxIndexKeys, + stats.EVMTxHashKeys, + ) + return nil + }, + } +} + +func trimCometIndexes(db cmtdb.DB, target int64, progress func(scanned int64)) (indexTrimStats, error) { + var stats indexTrimStats + var deleteKeys [][]byte + txHashes := make(map[string][]byte) + + iterator, err := db.Iterator(nil, nil) + if err != nil { + return stats, fmt.Errorf("iterate CometBFT indexes: %w", err) + } + for ; iterator.Valid(); iterator.Next() { + stats.Scanned++ + if progress != nil { + progress(stats.Scanned) + } + + key := iterator.Key() + value := iterator.Value() + if bytes.HasPrefix(key, blockEventPrefix) { + height, n := binary.Varint(value) + if n <= 0 { + _ = iterator.Close() + return stats, fmt.Errorf("decode block-event index height key=%X value=%X", key, value) + } + if height > target { + deleteKeys = append(deleteKeys, bytes.Clone(key)) + stats.BlockEventKeys++ + } + continue + } + + height, ok := txEventHeight(key, value) + if ok && height > target { + deleteKeys = append(deleteKeys, bytes.Clone(key)) + txHashes[string(value)] = bytes.Clone(value) + stats.TxEventKeys++ + } + } + if err := iterator.Error(); err != nil { + _ = iterator.Close() + return stats, fmt.Errorf("scan CometBFT indexes: %w", err) + } + if err := iterator.Close(); err != nil { + return stats, fmt.Errorf("close CometBFT index iterator: %w", err) + } + + for _, hash := range txHashes { + value, err := db.Get(hash) + if err != nil { + return stats, fmt.Errorf("read transaction index %X: %w", hash, err) + } + if value == nil { + continue + } + var result cmtabci.TxResult + if err := proto.Unmarshal(value, &result); err != nil { + return stats, fmt.Errorf("decode transaction index %X: %w", hash, err) + } + if result.Height > target { + deleteKeys = append(deleteKeys, hash) + stats.TxHashKeys++ + } + } + + if err := writeDeleteBatches(func() deleteBatch { return db.NewBatch() }, deleteKeys); err != nil { + return stats, fmt.Errorf("delete CometBFT index suffix: %w", err) + } + return stats, nil +} + +func trimEVMIndexes(db cosmosdb.DB, target int64) (indexTrimStats, error) { + var stats indexTrimStats + var deleteKeys [][]byte + + iterator, err := db.Iterator([]byte{evmTxIndexPrefix}, []byte{evmTxIndexPrefix + 1}) + if err != nil { + return stats, fmt.Errorf("iterate EVM indexes: %w", err) + } + for ; iterator.Valid(); iterator.Next() { + key := iterator.Key() + value := iterator.Value() + if len(key) != 17 || len(value) != 32 { + _ = iterator.Close() + return stats, fmt.Errorf("invalid EVM index entry key=%X value_length=%d", key, len(value)) + } + if int64(binary.BigEndian.Uint64(key[1:9])) > target { + deleteKeys = append(deleteKeys, bytes.Clone(key)) + deleteKeys = append(deleteKeys, append([]byte{evmTxHashPrefix}, value...)) + stats.EVMTxIndexKeys++ + stats.EVMTxHashKeys++ + } + } + if err := iterator.Error(); err != nil { + _ = iterator.Close() + return stats, fmt.Errorf("scan EVM indexes: %w", err) + } + if err := iterator.Close(); err != nil { + return stats, fmt.Errorf("close EVM index iterator: %w", err) + } + + if err := writeDeleteBatches(func() deleteBatch { return db.NewBatch() }, deleteKeys); err != nil { + return stats, fmt.Errorf("delete EVM index suffix: %w", err) + } + return stats, nil +} + +func txEventHeight(key, value []byte) (int64, bool) { + if len(value) != 32 { + return 0, false + } + lastSlash := bytes.LastIndexByte(key, '/') + if lastSlash <= 0 { + return 0, false + } + previousSlash := bytes.LastIndexByte(key[:lastSlash], '/') + if previousSlash <= 0 || bytes.LastIndexByte(key[:previousSlash], '/') < 0 { + return 0, false + } + if !validEventIndexSuffix(key[lastSlash+1:]) { + return 0, false + } + + height, err := strconv.ParseInt(string(key[previousSlash+1:lastSlash]), 10, 64) + if err != nil || height <= 0 { + return 0, false + } + return height, true +} + +func validEventIndexSuffix(value []byte) bool { + parts := bytes.Split(value, []byte("$es$")) + if len(parts) != 2 || len(parts[0]) == 0 || len(parts[1]) == 0 { + return false + } + for _, part := range parts { + for _, char := range part { + if char < '0' || char > '9' { + return false + } + } + } + return true +} + +type deleteBatch interface { + Delete(key []byte) error + WriteSync() error + Close() error +} + +func writeDeleteBatches(newBatch func() deleteBatch, keys [][]byte) error { + for start := 0; start < len(keys); start += deleteBatchSize { + end := start + deleteBatchSize + if end > len(keys) { + end = len(keys) + } + + batch := newBatch() + for _, key := range keys[start:end] { + if err := batch.Delete(key); err != nil { + _ = batch.Close() + return err + } + } + if err := batch.WriteSync(); err != nil { + _ = batch.Close() + return err + } + if err := batch.Close(); err != nil { + return err + } + } + return nil +} diff --git a/cmd/cronosd/cmd/rewind_indexes_test.go b/cmd/cronosd/cmd/rewind_indexes_test.go new file mode 100644 index 0000000..8ce1840 --- /dev/null +++ b/cmd/cronosd/cmd/rewind_indexes_test.go @@ -0,0 +1,104 @@ +package cmd + +import ( + "testing" + + cmtdb "github.com/cometbft/cometbft-db" + cmtabci "github.com/cometbft/cometbft/abci/types" + cmtquery "github.com/cometbft/cometbft/libs/pubsub/query" + cmtblockidxkv "github.com/cometbft/cometbft/state/indexer/block/kv" + cmttxidxkv "github.com/cometbft/cometbft/state/txindex/kv" + cmttypes "github.com/cometbft/cometbft/types" + cosmosdb "github.com/cosmos/cosmos-db" + "github.com/ethereum/go-ethereum/common" + ethindexer "github.com/evmos/ethermint/indexer" + "github.com/stretchr/testify/require" +) + +func TestTrimCometIndexes(t *testing.T) { + db := cmtdb.NewMemDB() + txIndexer := cmttxidxkv.NewTxIndex(db) + blockIndexer := cmtblockidxkv.New(cmtdb.NewPrefixDB(db, blockEventPrefix)) + + lowTx := indexTestTxResult(10, "low/value") + highTx := indexTestTxResult(11, "high/value") + require.NoError(t, txIndexer.Index(lowTx)) + require.NoError(t, txIndexer.Index(highTx)) + require.NoError(t, blockIndexer.Index(indexTestBlockEvents(10))) + require.NoError(t, blockIndexer.Index(indexTestBlockEvents(11))) + + stats, err := trimCometIndexes(db, 10, nil) + require.NoError(t, err) + require.EqualValues(t, 2, stats.TxEventKeys) + require.EqualValues(t, 1, stats.TxHashKeys) + require.EqualValues(t, 2, stats.BlockEventKeys) + + result, err := txIndexer.Get(cmttypes.Tx(lowTx.Tx).Hash()) + require.NoError(t, err) + require.NotNil(t, result) + result, err = txIndexer.Get(cmttypes.Tx(highTx.Tx).Hash()) + require.NoError(t, err) + require.Nil(t, result) + + lowBlocks, err := blockIndexer.Search(t.Context(), cmtquery.MustCompile("block.height = 10")) + require.NoError(t, err) + require.Equal(t, []int64{10}, lowBlocks) + highBlocks, err := blockIndexer.Search(t.Context(), cmtquery.MustCompile("block.height = 11")) + require.NoError(t, err) + require.Empty(t, highBlocks) +} + +func TestTrimEVMIndexes(t *testing.T) { + db := cosmosdb.NewMemDB() + lowHash := common.HexToHash("0x10") + highHash := common.HexToHash("0x11") + + require.NoError(t, db.Set(ethindexer.TxIndexKey(10, 0), lowHash.Bytes())) + require.NoError(t, db.Set(ethindexer.TxHashKey(lowHash), []byte("low"))) + require.NoError(t, db.Set(ethindexer.TxIndexKey(11, 0), highHash.Bytes())) + require.NoError(t, db.Set(ethindexer.TxHashKey(highHash), []byte("high"))) + + stats, err := trimEVMIndexes(db, 10) + require.NoError(t, err) + require.EqualValues(t, 1, stats.EVMTxIndexKeys) + require.EqualValues(t, 1, stats.EVMTxHashKeys) + + require.True(t, dbHas(t, db, ethindexer.TxIndexKey(10, 0))) + require.True(t, dbHas(t, db, ethindexer.TxHashKey(lowHash))) + require.False(t, dbHas(t, db, ethindexer.TxIndexKey(11, 0))) + require.False(t, dbHas(t, db, ethindexer.TxHashKey(highHash))) +} + +func indexTestTxResult(height int64, value string) *cmtabci.TxResult { + return &cmtabci.TxResult{ + Height: height, + Tx: []byte(value), + Result: cmtabci.ExecTxResult{ + Events: []cmtabci.Event{{ + Type: "test", + Attributes: []cmtabci.EventAttribute{{ + Key: "value", Value: value, Index: true, + }}, + }}, + }, + } +} + +func indexTestBlockEvents(height int64) cmttypes.EventDataNewBlockEvents { + return cmttypes.EventDataNewBlockEvents{ + Height: height, + Events: []cmtabci.Event{{ + Type: "test", + Attributes: []cmtabci.EventAttribute{{ + Key: "height", Value: "indexed", Index: true, + }}, + }}, + } +} + +func dbHas(t *testing.T, db cosmosdb.DB, key []byte) bool { + t.Helper() + found, err := db.Has(key) + require.NoError(t, err) + return found +} diff --git a/cmd/cronosd/cmd/root.go b/cmd/cronosd/cmd/root.go index 22515b7..bd2e4fe 100644 --- a/cmd/cronosd/cmd/root.go +++ b/cmd/cronosd/cmd/root.go @@ -184,7 +184,7 @@ func initRootCmd( rootCmd.AddCommand(changeSetCmd) } - rootCmd.AddCommand(RollbackToVersionCmd(), RewindCometBFTCmd()) + rootCmd.AddCommand(RollbackToVersionCmd(), RewindCometBFTCmd(), RewindIndexesCmd()) // add keybase, auxiliary RPC, query, and tx child commands rootCmd.AddCommand( From e489e87b68642da5db9c9f8fe79aa5ac78c7ac7f Mon Sep 17 00:00:00 2001 From: lihe Date: Tue, 1 Sep 2026 15:39:11 +0800 Subject: [PATCH 3/7] perf: seek directly to index suffixes Signed-off-by: lihe --- cmd/cronosd/cmd/rewind_indexes.go | 200 +++++++++++++++++++------ cmd/cronosd/cmd/rewind_indexes_test.go | 16 +- 2 files changed, 173 insertions(+), 43 deletions(-) diff --git a/cmd/cronosd/cmd/rewind_indexes.go b/cmd/cronosd/cmd/rewind_indexes.go index aa28487..d284677 100644 --- a/cmd/cronosd/cmd/rewind_indexes.go +++ b/cmd/cronosd/cmd/rewind_indexes.go @@ -4,6 +4,7 @@ import ( "bytes" "encoding/binary" "fmt" + "math" "strconv" cmtdb "github.com/cometbft/cometbft-db" @@ -24,12 +25,14 @@ const ( var ( blockEventPrefix = []byte("block_events") + txHeightPrefix = []byte("tx.height/") evmTxHashPrefix = byte(1) evmTxIndexPrefix = byte(2) ) type indexTrimStats struct { Scanned int64 + TxHeightKeys int64 TxEventKeys int64 TxHashKeys int64 BlockEventKeys int64 @@ -62,9 +65,9 @@ func RewindIndexesCmd() *cobra.Command { if err != nil { return fmt.Errorf("open CometBFT index database: %w", err) } - stats, err := trimCometIndexes(txIndexDB, target, func(scanned int64) { + stats, err := trimCometIndexes(txIndexDB, target, func(stage string, scanned int64) { if scanned%indexScanInterval == 0 { - cmd.Printf("CometBFT index scan progress keys=%d\n", scanned) + cmd.Printf("CometBFT %s scan progress keys=%d\n", stage, scanned) } }) closeErr := txIndexDB.Close() @@ -91,9 +94,10 @@ func RewindIndexesCmd() *cobra.Command { stats.EVMTxIndexKeys = evmStats.EVMTxIndexKeys stats.EVMTxHashKeys = evmStats.EVMTxHashKeys cmd.Printf( - "rewound indexes to height %d scanned=%d tx_events=%d tx_hashes=%d block_events=%d evm_tx_indexes=%d evm_tx_hashes=%d\n", + "rewound indexes to height %d scanned=%d tx_heights=%d tx_events=%d tx_hashes=%d block_events=%d evm_tx_indexes=%d evm_tx_hashes=%d\n", target, stats.Scanned, + stats.TxHeightKeys, stats.TxEventKeys, stats.TxHashKeys, stats.BlockEventKeys, @@ -105,67 +109,133 @@ func RewindIndexesCmd() *cobra.Command { } } -func trimCometIndexes(db cmtdb.DB, target int64, progress func(scanned int64)) (indexTrimStats, error) { +type indexedTransaction struct { + height int64 + heightKey []byte + hash []byte +} + +func trimCometIndexes( + db cmtdb.DB, + target int64, + progress func(stage string, scanned int64), +) (indexTrimStats, error) { var stats indexTrimStats var deleteKeys [][]byte - txHashes := make(map[string][]byte) + var transactions []indexedTransaction - iterator, err := db.Iterator(nil, nil) + txIterator, err := db.Iterator(txHeightPrefix, prefixEnd(txHeightPrefix)) if err != nil { - return stats, fmt.Errorf("iterate CometBFT indexes: %w", err) + return stats, fmt.Errorf("iterate CometBFT transaction heights: %w", err) } - for ; iterator.Valid(); iterator.Next() { + var txScanned int64 + for ; txIterator.Valid(); txIterator.Next() { + txScanned++ stats.Scanned++ if progress != nil { - progress(stats.Scanned) + progress("tx-height", txScanned) } - key := iterator.Key() - value := iterator.Value() - if bytes.HasPrefix(key, blockEventPrefix) { - height, n := binary.Varint(value) - if n <= 0 { - _ = iterator.Close() - return stats, fmt.Errorf("decode block-event index height key=%X value=%X", key, value) - } - if height > target { - deleteKeys = append(deleteKeys, bytes.Clone(key)) - stats.BlockEventKeys++ - } + height, ok := txEventHeight(txIterator.Key(), txIterator.Value()) + if !ok || height <= target { continue } - - height, ok := txEventHeight(key, value) - if ok && height > target { - deleteKeys = append(deleteKeys, bytes.Clone(key)) - txHashes[string(value)] = bytes.Clone(value) - stats.TxEventKeys++ - } + transactions = append(transactions, indexedTransaction{ + height: height, + heightKey: bytes.Clone(txIterator.Key()), + hash: bytes.Clone(txIterator.Value()), + }) + stats.TxHeightKeys++ } - if err := iterator.Error(); err != nil { - _ = iterator.Close() - return stats, fmt.Errorf("scan CometBFT indexes: %w", err) + if err := txIterator.Error(); err != nil { + _ = txIterator.Close() + return stats, fmt.Errorf("scan CometBFT transaction heights: %w", err) } - if err := iterator.Close(); err != nil { - return stats, fmt.Errorf("close CometBFT index iterator: %w", err) + if err := txIterator.Close(); err != nil { + return stats, fmt.Errorf("close CometBFT transaction-height iterator: %w", err) } - for _, hash := range txHashes { - value, err := db.Get(hash) + for _, transaction := range transactions { + value, err := db.Get(transaction.hash) if err != nil { - return stats, fmt.Errorf("read transaction index %X: %w", hash, err) + return stats, fmt.Errorf("read transaction index %X: %w", transaction.hash, err) } if value == nil { - continue + return stats, fmt.Errorf("transaction index %X is missing", transaction.hash) } var result cmtabci.TxResult if err := proto.Unmarshal(value, &result); err != nil { - return stats, fmt.Errorf("decode transaction index %X: %w", hash, err) + return stats, fmt.Errorf("decode transaction index %X: %w", transaction.hash, err) + } + if result.Height != transaction.height { + return stats, fmt.Errorf( + "transaction index %X height mismatch: height key=%d result=%d", + transaction.hash, + transaction.height, + result.Height, + ) + } + + deleteKeys = append(deleteKeys, transaction.heightKey, transaction.hash) + stats.TxHashKeys++ + for _, eventPrefix := range txEventPrefixes(&result) { + eventIterator, err := db.Iterator(eventPrefix, prefixEnd(eventPrefix)) + if err != nil { + return stats, fmt.Errorf("iterate transaction event prefix %X: %w", eventPrefix, err) + } + for ; eventIterator.Valid(); eventIterator.Next() { + if !bytes.Equal(eventIterator.Value(), transaction.hash) { + _ = eventIterator.Close() + return stats, fmt.Errorf( + "transaction event %X points to unexpected hash %X", + eventIterator.Key(), + eventIterator.Value(), + ) + } + deleteKeys = append(deleteKeys, bytes.Clone(eventIterator.Key())) + stats.TxEventKeys++ + } + if err := eventIterator.Error(); err != nil { + _ = eventIterator.Close() + return stats, fmt.Errorf("scan transaction event prefix %X: %w", eventPrefix, err) + } + if err := eventIterator.Close(); err != nil { + return stats, fmt.Errorf("close transaction event iterator: %w", err) + } + } + } + + blockIterator, err := db.Iterator(blockEventPrefix, prefixEnd(blockEventPrefix)) + if err != nil { + return stats, fmt.Errorf("iterate CometBFT block events: %w", err) + } + var blockScanned int64 + for ; blockIterator.Valid(); blockIterator.Next() { + blockScanned++ + stats.Scanned++ + if progress != nil { + progress("block-event", blockScanned) } - if result.Height > target { - deleteKeys = append(deleteKeys, hash) - stats.TxHashKeys++ + height, n := binary.Varint(blockIterator.Value()) + if n <= 0 { + _ = blockIterator.Close() + return stats, fmt.Errorf( + "decode block-event index height key=%X value=%X", + blockIterator.Key(), + blockIterator.Value(), + ) } + if height > target { + deleteKeys = append(deleteKeys, bytes.Clone(blockIterator.Key())) + stats.BlockEventKeys++ + } + } + if err := blockIterator.Error(); err != nil { + _ = blockIterator.Close() + return stats, fmt.Errorf("scan CometBFT block events: %w", err) + } + if err := blockIterator.Close(); err != nil { + return stats, fmt.Errorf("close CometBFT block-event iterator: %w", err) } if err := writeDeleteBatches(func() deleteBatch { return db.NewBatch() }, deleteKeys); err != nil { @@ -177,8 +247,14 @@ func trimCometIndexes(db cmtdb.DB, target int64, progress func(scanned int64)) ( func trimEVMIndexes(db cosmosdb.DB, target int64) (indexTrimStats, error) { var stats indexTrimStats var deleteKeys [][]byte + if target == math.MaxInt64 { + return stats, nil + } - iterator, err := db.Iterator([]byte{evmTxIndexPrefix}, []byte{evmTxIndexPrefix + 1}) + start := make([]byte, 9) + start[0] = evmTxIndexPrefix + binary.BigEndian.PutUint64(start[1:], uint64(target)+1) + iterator, err := db.Iterator(start, []byte{evmTxIndexPrefix + 1}) if err != nil { return stats, fmt.Errorf("iterate EVM indexes: %w", err) } @@ -210,6 +286,46 @@ func trimEVMIndexes(db cosmosdb.DB, target int64) (indexTrimStats, error) { return stats, nil } +func txEventPrefixes(result *cmtabci.TxResult) [][]byte { + prefixes := make(map[string][]byte) + for _, event := range result.Result.Events { + if len(event.Type) == 0 { + continue + } + for _, attribute := range event.Attributes { + if len(attribute.Key) == 0 || !attribute.GetIndex() { + continue + } + prefix := []byte(fmt.Sprintf( + "%s.%s/%s/%d/%d$es$", + event.Type, + attribute.Key, + attribute.Value, + result.Height, + result.Index, + )) + prefixes[string(prefix)] = prefix + } + } + + resultPrefixes := make([][]byte, 0, len(prefixes)) + for _, prefix := range prefixes { + resultPrefixes = append(resultPrefixes, prefix) + } + return resultPrefixes +} + +func prefixEnd(prefix []byte) []byte { + end := bytes.Clone(prefix) + for index := len(end) - 1; index >= 0; index-- { + if end[index] != math.MaxUint8 { + end[index]++ + return end[:index+1] + } + } + return nil +} + func txEventHeight(key, value []byte) (int64, bool) { if len(value) != 32 { return 0, false diff --git a/cmd/cronosd/cmd/rewind_indexes_test.go b/cmd/cronosd/cmd/rewind_indexes_test.go index 8ce1840..1c04318 100644 --- a/cmd/cronosd/cmd/rewind_indexes_test.go +++ b/cmd/cronosd/cmd/rewind_indexes_test.go @@ -29,7 +29,8 @@ func TestTrimCometIndexes(t *testing.T) { stats, err := trimCometIndexes(db, 10, nil) require.NoError(t, err) - require.EqualValues(t, 2, stats.TxEventKeys) + require.EqualValues(t, 1, stats.TxHeightKeys) + require.EqualValues(t, 1, stats.TxEventKeys) require.EqualValues(t, 1, stats.TxHashKeys) require.EqualValues(t, 2, stats.BlockEventKeys) @@ -39,6 +40,12 @@ func TestTrimCometIndexes(t *testing.T) { result, err = txIndexer.Get(cmttypes.Tx(highTx.Tx).Hash()) require.NoError(t, err) require.Nil(t, result) + lowTransactions, err := txIndexer.Search(t.Context(), cmtquery.MustCompile("test.value = 'low/value'")) + require.NoError(t, err) + require.Len(t, lowTransactions, 1) + highTransactions, err := txIndexer.Search(t.Context(), cmtquery.MustCompile("test.value = 'high/value'")) + require.NoError(t, err) + require.Empty(t, highTransactions) lowBlocks, err := blockIndexer.Search(t.Context(), cmtquery.MustCompile("block.height = 10")) require.NoError(t, err) @@ -57,6 +64,7 @@ func TestTrimEVMIndexes(t *testing.T) { require.NoError(t, db.Set(ethindexer.TxHashKey(lowHash), []byte("low"))) require.NoError(t, db.Set(ethindexer.TxIndexKey(11, 0), highHash.Bytes())) require.NoError(t, db.Set(ethindexer.TxHashKey(highHash), []byte("high"))) + require.NoError(t, db.Set([]byte{evmTxIndexPrefix, 0}, []byte("older malformed entry"))) stats, err := trimEVMIndexes(db, 10) require.NoError(t, err) @@ -67,6 +75,12 @@ func TestTrimEVMIndexes(t *testing.T) { require.True(t, dbHas(t, db, ethindexer.TxHashKey(lowHash))) require.False(t, dbHas(t, db, ethindexer.TxIndexKey(11, 0))) require.False(t, dbHas(t, db, ethindexer.TxHashKey(highHash))) + require.True(t, dbHas(t, db, []byte{evmTxIndexPrefix, 0})) +} + +func TestPrefixEnd(t *testing.T) { + require.Equal(t, []byte{0x01, 0x03}, prefixEnd([]byte{0x01, 0x02, 0xFF})) + require.Nil(t, prefixEnd([]byte{0xFF, 0xFF})) } func indexTestTxResult(height int64, value string) *cmtabci.TxResult { From cd7277981048df9b14e86b9b2527f4a8ce034d9a Mon Sep 17 00:00:00 2001 From: lihe Date: Tue, 1 Sep 2026 16:06:57 +0800 Subject: [PATCH 4/7] fix: handle duplicate transactions during index rewind Signed-off-by: lihe --- cmd/cronosd/cmd/rewind_indexes.go | 279 +++++++++++++++++++++---- cmd/cronosd/cmd/rewind_indexes_test.go | 68 +++++- 2 files changed, 301 insertions(+), 46 deletions(-) diff --git a/cmd/cronosd/cmd/rewind_indexes.go b/cmd/cronosd/cmd/rewind_indexes.go index d284677..cdc34c7 100644 --- a/cmd/cronosd/cmd/rewind_indexes.go +++ b/cmd/cronosd/cmd/rewind_indexes.go @@ -10,6 +10,8 @@ import ( cmtdb "github.com/cometbft/cometbft-db" cmtabci "github.com/cometbft/cometbft/abci/types" cmtcfg "github.com/cometbft/cometbft/config" + cmtstate "github.com/cometbft/cometbft/state" + cmttypes "github.com/cometbft/cometbft/types" cosmosdb "github.com/cosmos/cosmos-db" "github.com/cosmos/cosmos-sdk/server" ethermintserver "github.com/evmos/ethermint/server" @@ -31,13 +33,14 @@ var ( ) type indexTrimStats struct { - Scanned int64 - TxHeightKeys int64 - TxEventKeys int64 - TxHashKeys int64 - BlockEventKeys int64 - EVMTxIndexKeys int64 - EVMTxHashKeys int64 + Scanned int64 + TxHeightKeys int64 + TxEventKeys int64 + TxHashKeys int64 + TxHashesRestored int64 + BlockEventKeys int64 + EVMTxIndexKeys int64 + EVMTxHashKeys int64 } // RewindIndexesCmd removes transaction and block-event index entries above a @@ -65,17 +68,30 @@ func RewindIndexesCmd() *cobra.Command { if err != nil { return fmt.Errorf("open CometBFT index database: %w", err) } - stats, err := trimCometIndexes(txIndexDB, target, func(stage string, scanned int64) { + stateDB, err := cmtcfg.DefaultDBProvider(&cmtcfg.DBContext{ID: "state", Config: cfg}) + if err != nil { + _ = txIndexDB.Close() + return fmt.Errorf("open CometBFT state database: %w", err) + } + stateStore := cmtstate.NewStore(stateDB, cmtstate.StoreOptions{ + DiscardABCIResponses: cfg.Storage.DiscardABCIResponses, + }) + + stats, err := trimCometIndexes(txIndexDB, target, stateStore.LoadFinalizeBlockResponse, func(stage string, scanned int64) { if scanned%indexScanInterval == 0 { cmd.Printf("CometBFT %s scan progress keys=%d\n", stage, scanned) } }) - closeErr := txIndexDB.Close() + txCloseErr := txIndexDB.Close() + stateCloseErr := stateDB.Close() if err != nil { return err } - if closeErr != nil { - return fmt.Errorf("close CometBFT index database: %w", closeErr) + if txCloseErr != nil { + return fmt.Errorf("close CometBFT index database: %w", txCloseErr) + } + if stateCloseErr != nil { + return fmt.Errorf("close CometBFT state database: %w", stateCloseErr) } evmIndexDB, err := ethermintserver.OpenIndexerDB(cfg.RootDir, server.GetAppDBBackend(ctx.Viper)) @@ -83,7 +99,7 @@ func RewindIndexesCmd() *cobra.Command { return fmt.Errorf("open EVM index database: %w", err) } evmStats, err := trimEVMIndexes(evmIndexDB, target) - closeErr = evmIndexDB.Close() + closeErr := evmIndexDB.Close() if err != nil { return err } @@ -94,12 +110,13 @@ func RewindIndexesCmd() *cobra.Command { stats.EVMTxIndexKeys = evmStats.EVMTxIndexKeys stats.EVMTxHashKeys = evmStats.EVMTxHashKeys cmd.Printf( - "rewound indexes to height %d scanned=%d tx_heights=%d tx_events=%d tx_hashes=%d block_events=%d evm_tx_indexes=%d evm_tx_hashes=%d\n", + "rewound indexes to height %d scanned=%d tx_heights=%d tx_events=%d tx_hashes=%d tx_hashes_restored=%d block_events=%d evm_tx_indexes=%d evm_tx_hashes=%d\n", target, stats.Scanned, stats.TxHeightKeys, stats.TxEventKeys, stats.TxHashKeys, + stats.TxHashesRestored, stats.BlockEventKeys, stats.EVMTxIndexKeys, stats.EVMTxHashKeys, @@ -111,18 +128,28 @@ func RewindIndexesCmd() *cobra.Command { type indexedTransaction struct { height int64 + index uint32 heightKey []byte hash []byte } +type indexValue struct { + key []byte + value []byte +} + func trimCometIndexes( db cmtdb.DB, target int64, + loadFinalizeBlock func(int64) (*cmtabci.ResponseFinalizeBlock, error), progress func(stage string, scanned int64), ) (indexTrimStats, error) { var stats indexTrimStats - var deleteKeys [][]byte + var deleteDataKeys [][]byte + var deleteHeightKeys [][]byte + var restoreValues []indexValue var transactions []indexedTransaction + highHashes := make(map[string]struct{}) txIterator, err := db.Iterator(txHeightPrefix, prefixEnd(txHeightPrefix)) if err != nil { @@ -136,15 +163,21 @@ func trimCometIndexes( progress("tx-height", txScanned) } - height, ok := txEventHeight(txIterator.Key(), txIterator.Value()) - if !ok || height <= target { + height, index, err := txHeightOccurrence(txIterator.Key(), txIterator.Value()) + if err != nil { + _ = txIterator.Close() + return stats, err + } + if height <= target { continue } transactions = append(transactions, indexedTransaction{ height: height, + index: index, heightKey: bytes.Clone(txIterator.Key()), hash: bytes.Clone(txIterator.Value()), }) + highHashes[string(txIterator.Value())] = struct{}{} stats.TxHeightKeys++ } if err := txIterator.Error(); err != nil { @@ -155,6 +188,90 @@ func trimCometIndexes( return stats, fmt.Errorf("close CometBFT transaction-height iterator: %w", err) } + retainedTransactions := make(map[string]indexedTransaction) + if len(highHashes) > 0 { + retainedIterator, err := db.Iterator(txHeightPrefix, prefixEnd(txHeightPrefix)) + if err != nil { + return stats, fmt.Errorf("iterate retained CometBFT transaction heights: %w", err) + } + var retainedScanned int64 + for ; retainedIterator.Valid(); retainedIterator.Next() { + retainedScanned++ + stats.Scanned++ + if progress != nil { + progress("tx-retained", retainedScanned) + } + + height, index, err := txHeightOccurrence(retainedIterator.Key(), retainedIterator.Value()) + if err != nil { + _ = retainedIterator.Close() + return stats, err + } + hashKey := string(retainedIterator.Value()) + if height > target { + continue + } + if _, found := highHashes[hashKey]; !found { + continue + } + + retained, found := retainedTransactions[hashKey] + if !found || height > retained.height || height == retained.height && index > retained.index { + retainedTransactions[hashKey] = indexedTransaction{ + height: height, + index: index, + hash: bytes.Clone(retainedIterator.Value()), + } + } + } + if err := retainedIterator.Error(); err != nil { + _ = retainedIterator.Close() + return stats, fmt.Errorf("scan retained CometBFT transaction heights: %w", err) + } + if err := retainedIterator.Close(); err != nil { + return stats, fmt.Errorf("close retained CometBFT transaction-height iterator: %w", err) + } + } + + if len(transactions) > 0 && loadFinalizeBlock == nil { + return stats, fmt.Errorf("load FinalizeBlockResponse callback is required") + } + var cachedHeight int64 = -1 + var cachedResponse *cmtabci.ResponseFinalizeBlock + loadTransactionResult := func(transaction indexedTransaction, tx []byte) (*cmtabci.TxResult, error) { + if cachedHeight != transaction.height { + response, err := loadFinalizeBlock(transaction.height) + if err != nil { + return nil, fmt.Errorf("load FinalizeBlockResponse at height %d: %w", transaction.height, err) + } + cachedHeight = transaction.height + cachedResponse = response + } + resultCount := 0 + if cachedResponse != nil { + resultCount = len(cachedResponse.TxResults) + } + if uint64(transaction.index) >= uint64(resultCount) { + return nil, fmt.Errorf( + "transaction index %d is outside FinalizeBlockResponse at height %d (results=%d)", + transaction.index, + transaction.height, + resultCount, + ) + } + execResult := cachedResponse.TxResults[transaction.index] + if execResult == nil { + return nil, fmt.Errorf("transaction result %d is nil at height %d", transaction.index, transaction.height) + } + return &cmtabci.TxResult{ + Height: transaction.height, + Index: transaction.index, + Tx: bytes.Clone(tx), + Result: *execResult, + }, nil + } + + processedHashes := make(map[string]struct{}) for _, transaction := range transactions { value, err := db.Get(transaction.hash) if err != nil { @@ -167,18 +284,20 @@ func trimCometIndexes( if err := proto.Unmarshal(value, &result); err != nil { return stats, fmt.Errorf("decode transaction index %X: %w", transaction.hash, err) } - if result.Height != transaction.height { + if !bytes.Equal(cmttypes.Tx(result.Tx).Hash(), transaction.hash) { return stats, fmt.Errorf( - "transaction index %X height mismatch: height key=%d result=%d", + "transaction index %X contains transaction with hash %X", transaction.hash, - transaction.height, - result.Height, + cmttypes.Tx(result.Tx).Hash(), ) } + indexedResult, err := loadTransactionResult(transaction, result.Tx) + if err != nil { + return stats, err + } - deleteKeys = append(deleteKeys, transaction.heightKey, transaction.hash) - stats.TxHashKeys++ - for _, eventPrefix := range txEventPrefixes(&result) { + deleteHeightKeys = append(deleteHeightKeys, transaction.heightKey) + for _, eventPrefix := range txEventPrefixes(indexedResult) { eventIterator, err := db.Iterator(eventPrefix, prefixEnd(eventPrefix)) if err != nil { return stats, fmt.Errorf("iterate transaction event prefix %X: %w", eventPrefix, err) @@ -192,7 +311,7 @@ func trimCometIndexes( eventIterator.Value(), ) } - deleteKeys = append(deleteKeys, bytes.Clone(eventIterator.Key())) + deleteDataKeys = append(deleteDataKeys, bytes.Clone(eventIterator.Key())) stats.TxEventKeys++ } if err := eventIterator.Error(); err != nil { @@ -203,6 +322,30 @@ func trimCometIndexes( return stats, fmt.Errorf("close transaction event iterator: %w", err) } } + + hashKey := string(transaction.hash) + if _, processed := processedHashes[hashKey]; processed { + continue + } + processedHashes[hashKey] = struct{}{} + if retained, found := retainedTransactions[hashKey]; found { + retainedResult, err := loadTransactionResult(retained, result.Tx) + if err != nil { + return stats, err + } + encoded, err := proto.Marshal(retainedResult) + if err != nil { + return stats, fmt.Errorf("encode retained transaction index %X: %w", transaction.hash, err) + } + restoreValues = append(restoreValues, indexValue{ + key: bytes.Clone(transaction.hash), + value: encoded, + }) + stats.TxHashesRestored++ + } else { + deleteDataKeys = append(deleteDataKeys, bytes.Clone(transaction.hash)) + stats.TxHashKeys++ + } } blockIterator, err := db.Iterator(blockEventPrefix, prefixEnd(blockEventPrefix)) @@ -226,7 +369,7 @@ func trimCometIndexes( ) } if height > target { - deleteKeys = append(deleteKeys, bytes.Clone(blockIterator.Key())) + deleteDataKeys = append(deleteDataKeys, bytes.Clone(blockIterator.Key())) stats.BlockEventKeys++ } } @@ -238,8 +381,14 @@ func trimCometIndexes( return stats, fmt.Errorf("close CometBFT block-event iterator: %w", err) } - if err := writeDeleteBatches(func() deleteBatch { return db.NewBatch() }, deleteKeys); err != nil { - return stats, fmt.Errorf("delete CometBFT index suffix: %w", err) + if err := writeSetBatches(func() indexBatch { return db.NewBatch() }, restoreValues); err != nil { + return stats, fmt.Errorf("restore retained CometBFT transaction indexes: %w", err) + } + if err := writeDeleteBatches(func() indexBatch { return db.NewBatch() }, deleteDataKeys); err != nil { + return stats, fmt.Errorf("delete CometBFT transaction and block-event indexes: %w", err) + } + if err := writeDeleteBatches(func() indexBatch { return db.NewBatch() }, deleteHeightKeys); err != nil { + return stats, fmt.Errorf("delete CometBFT transaction-height indexes: %w", err) } return stats, nil } @@ -280,7 +429,7 @@ func trimEVMIndexes(db cosmosdb.DB, target int64) (indexTrimStats, error) { return stats, fmt.Errorf("close EVM index iterator: %w", err) } - if err := writeDeleteBatches(func() deleteBatch { return db.NewBatch() }, deleteKeys); err != nil { + if err := writeDeleteBatches(func() indexBatch { return db.NewBatch() }, deleteKeys); err != nil { return stats, fmt.Errorf("delete EVM index suffix: %w", err) } return stats, nil @@ -326,27 +475,43 @@ func prefixEnd(prefix []byte) []byte { return nil } -func txEventHeight(key, value []byte) (int64, bool) { +func txHeightOccurrence(key, value []byte) (int64, uint32, error) { if len(value) != 32 { - return 0, false + return 0, 0, fmt.Errorf("transaction-height index %X has invalid hash length %d", key, len(value)) } - lastSlash := bytes.LastIndexByte(key, '/') - if lastSlash <= 0 { - return 0, false + if !bytes.HasPrefix(key, txHeightPrefix) { + return 0, 0, fmt.Errorf("transaction-height index has invalid prefix: %X", key) } - previousSlash := bytes.LastIndexByte(key[:lastSlash], '/') - if previousSlash <= 0 || bytes.LastIndexByte(key[:previousSlash], '/') < 0 { - return 0, false + parts := bytes.Split(key[len(txHeightPrefix):], []byte{'/'}) + if len(parts) != 3 { + return 0, 0, fmt.Errorf("transaction-height index has invalid key: %X", key) } - if !validEventIndexSuffix(key[lastSlash+1:]) { - return 0, false + if !validEventIndexSuffix(parts[2]) { + return 0, 0, fmt.Errorf("transaction-height index has invalid event suffix: %X", key) } - height, err := strconv.ParseInt(string(key[previousSlash+1:lastSlash]), 10, 64) - if err != nil || height <= 0 { - return 0, false + valueHeight, err := strconv.ParseInt(string(parts[0]), 10, 64) + if err != nil || valueHeight <= 0 { + return 0, 0, fmt.Errorf("transaction-height index has invalid value height: %X", key) + } + eventHeight, err := strconv.ParseInt(string(parts[1]), 10, 64) + if err != nil || eventHeight <= 0 { + return 0, 0, fmt.Errorf("transaction-height index has invalid event height: %X", key) + } + if valueHeight != eventHeight { + return 0, 0, fmt.Errorf( + "transaction-height index has inconsistent heights value=%d event=%d: %X", + valueHeight, + eventHeight, + key, + ) + } + suffixParts := bytes.Split(parts[2], []byte("$es$")) + index, err := strconv.ParseUint(string(suffixParts[0]), 10, 32) + if err != nil { + return 0, 0, fmt.Errorf("transaction-height index has invalid transaction index: %X", key) } - return height, true + return eventHeight, uint32(index), nil } func validEventIndexSuffix(value []byte) bool { @@ -364,13 +529,39 @@ func validEventIndexSuffix(value []byte) bool { return true } -type deleteBatch interface { +type indexBatch interface { + Set(key, value []byte) error Delete(key []byte) error WriteSync() error Close() error } -func writeDeleteBatches(newBatch func() deleteBatch, keys [][]byte) error { +func writeSetBatches(newBatch func() indexBatch, values []indexValue) error { + for start := 0; start < len(values); start += deleteBatchSize { + end := start + deleteBatchSize + if end > len(values) { + end = len(values) + } + + batch := newBatch() + for _, value := range values[start:end] { + if err := batch.Set(value.key, value.value); err != nil { + _ = batch.Close() + return err + } + } + if err := batch.WriteSync(); err != nil { + _ = batch.Close() + return err + } + if err := batch.Close(); err != nil { + return err + } + } + return nil +} + +func writeDeleteBatches(newBatch func() indexBatch, keys [][]byte) error { for start := 0; start < len(keys); start += deleteBatchSize { end := start + deleteBatchSize if end > len(keys) { diff --git a/cmd/cronosd/cmd/rewind_indexes_test.go b/cmd/cronosd/cmd/rewind_indexes_test.go index 1c04318..00d04a2 100644 --- a/cmd/cronosd/cmd/rewind_indexes_test.go +++ b/cmd/cronosd/cmd/rewind_indexes_test.go @@ -1,6 +1,7 @@ package cmd import ( + "fmt" "testing" cmtdb "github.com/cometbft/cometbft-db" @@ -27,7 +28,7 @@ func TestTrimCometIndexes(t *testing.T) { require.NoError(t, blockIndexer.Index(indexTestBlockEvents(10))) require.NoError(t, blockIndexer.Index(indexTestBlockEvents(11))) - stats, err := trimCometIndexes(db, 10, nil) + stats, err := trimCometIndexes(db, 10, indexTestFinalizeLoader(lowTx, highTx), nil) require.NoError(t, err) require.EqualValues(t, 1, stats.TxHeightKeys) require.EqualValues(t, 1, stats.TxEventKeys) @@ -55,6 +56,41 @@ func TestTrimCometIndexes(t *testing.T) { require.Empty(t, highBlocks) } +func TestTrimCometIndexesRestoresRetainedDuplicateTransaction(t *testing.T) { + db := cmtdb.NewMemDB() + txIndexer := cmttxidxkv.NewTxIndex(db) + + lowTx := indexTestTxResultWithTx(10, "same transaction", "low/value") + highTx := indexTestTxResultWithTx(11, "same transaction", "high/value") + require.NoError(t, txIndexer.Index(lowTx)) + require.NoError(t, txIndexer.Index(highTx)) + + stats, err := trimCometIndexes(db, 10, indexTestFinalizeLoader(lowTx, highTx), nil) + require.NoError(t, err) + require.EqualValues(t, 1, stats.TxHeightKeys) + require.EqualValues(t, 1, stats.TxEventKeys) + require.EqualValues(t, 0, stats.TxHashKeys) + require.EqualValues(t, 1, stats.TxHashesRestored) + + hash := cmttypes.Tx(lowTx.Tx).Hash() + result, err := txIndexer.Get(hash) + require.NoError(t, err) + require.Equal(t, lowTx, result) + lowTransactions, err := txIndexer.Search(t.Context(), cmtquery.MustCompile("test.value = 'low/value'")) + require.NoError(t, err) + require.Len(t, lowTransactions, 1) + highTransactions, err := txIndexer.Search(t.Context(), cmtquery.MustCompile("test.value = 'high/value'")) + require.NoError(t, err) + require.Empty(t, highTransactions) + + stats, err = trimCometIndexes(db, 10, indexTestFinalizeLoader(lowTx, highTx), nil) + require.NoError(t, err) + require.Zero(t, stats.TxHeightKeys) + require.Zero(t, stats.TxEventKeys) + require.Zero(t, stats.TxHashKeys) + require.Zero(t, stats.TxHashesRestored) +} + func TestTrimEVMIndexes(t *testing.T) { db := cosmosdb.NewMemDB() lowHash := common.HexToHash("0x10") @@ -84,9 +120,13 @@ func TestPrefixEnd(t *testing.T) { } func indexTestTxResult(height int64, value string) *cmtabci.TxResult { + return indexTestTxResultWithTx(height, value, value) +} + +func indexTestTxResultWithTx(height int64, tx, value string) *cmtabci.TxResult { return &cmtabci.TxResult{ Height: height, - Tx: []byte(value), + Tx: []byte(tx), Result: cmtabci.ExecTxResult{ Events: []cmtabci.Event{{ Type: "test", @@ -98,6 +138,30 @@ func indexTestTxResult(height int64, value string) *cmtabci.TxResult { } } +func indexTestFinalizeLoader(results ...*cmtabci.TxResult) func(int64) (*cmtabci.ResponseFinalizeBlock, error) { + responses := make(map[int64]*cmtabci.ResponseFinalizeBlock) + for _, result := range results { + response := responses[result.Height] + if response == nil { + response = &cmtabci.ResponseFinalizeBlock{} + responses[result.Height] = response + } + for uint32(len(response.TxResults)) <= result.Index { + response.TxResults = append(response.TxResults, nil) + } + execResult := result.Result + response.TxResults[result.Index] = &execResult + } + + return func(height int64) (*cmtabci.ResponseFinalizeBlock, error) { + response := responses[height] + if response == nil { + return nil, fmt.Errorf("missing response at height %d", height) + } + return response, nil + } +} + func indexTestBlockEvents(height int64) cmttypes.EventDataNewBlockEvents { return cmttypes.EventDataNewBlockEvents{ Height: height, From 6ab71f1422db7954be6dba1ca20910d8b614d81b Mon Sep 17 00:00:00 2001 From: lihe Date: Tue, 1 Sep 2026 16:40:03 +0800 Subject: [PATCH 5/7] perf: seek directly to block event suffixes Signed-off-by: lihe --- cmd/cronosd/cmd/rewind_indexes.go | 166 ++++++++++++++++++++----- cmd/cronosd/cmd/rewind_indexes_test.go | 26 +++- 2 files changed, 157 insertions(+), 35 deletions(-) diff --git a/cmd/cronosd/cmd/rewind_indexes.go b/cmd/cronosd/cmd/rewind_indexes.go index cdc34c7..7b2bc81 100644 --- a/cmd/cronosd/cmd/rewind_indexes.go +++ b/cmd/cronosd/cmd/rewind_indexes.go @@ -10,11 +10,13 @@ import ( cmtdb "github.com/cometbft/cometbft-db" cmtabci "github.com/cometbft/cometbft/abci/types" cmtcfg "github.com/cometbft/cometbft/config" + cmtstateproto "github.com/cometbft/cometbft/proto/tendermint/state" cmtstate "github.com/cometbft/cometbft/state" cmttypes "github.com/cometbft/cometbft/types" cosmosdb "github.com/cosmos/cosmos-db" "github.com/cosmos/cosmos-sdk/server" ethermintserver "github.com/evmos/ethermint/server" + "github.com/google/orderedcode" "github.com/spf13/cobra" "github.com/cosmos/gogoproto/proto" @@ -76,9 +78,16 @@ func RewindIndexesCmd() *cobra.Command { stateStore := cmtstate.NewStore(stateDB, cmtstate.StoreOptions{ DiscardABCIResponses: cfg.Storage.DiscardABCIResponses, }) + sourceHeight, err := lastFinalizeBlockHeight(stateDB) + if err != nil { + _ = txIndexDB.Close() + _ = stateDB.Close() + return err + } - stats, err := trimCometIndexes(txIndexDB, target, stateStore.LoadFinalizeBlockResponse, func(stage string, scanned int64) { - if scanned%indexScanInterval == 0 { + stats, err := trimCometIndexes(txIndexDB, target, sourceHeight, stateStore.LoadFinalizeBlockResponse, func(stage string, scanned int64) { + if stage == "block-height" && scanned%rewindProgressInterval == 0 || + stage != "block-height" && scanned%indexScanInterval == 0 { cmd.Printf("CometBFT %s scan progress keys=%d\n", stage, scanned) } }) @@ -141,10 +150,14 @@ type indexValue struct { func trimCometIndexes( db cmtdb.DB, target int64, + sourceHeight int64, loadFinalizeBlock func(int64) (*cmtabci.ResponseFinalizeBlock, error), progress func(stage string, scanned int64), ) (indexTrimStats, error) { var stats indexTrimStats + if sourceHeight < target { + return stats, fmt.Errorf("source height %d is below target %d", sourceHeight, target) + } var deleteDataKeys [][]byte var deleteHeightKeys [][]byte var restoreValues []indexValue @@ -233,7 +246,7 @@ func trimCometIndexes( } } - if len(transactions) > 0 && loadFinalizeBlock == nil { + if sourceHeight > target && loadFinalizeBlock == nil { return stats, fmt.Errorf("load FinalizeBlockResponse callback is required") } var cachedHeight int64 = -1 @@ -348,38 +361,22 @@ func trimCometIndexes( } } - blockIterator, err := db.Iterator(blockEventPrefix, prefixEnd(blockEventPrefix)) - if err != nil { - return stats, fmt.Errorf("iterate CometBFT block events: %w", err) - } - var blockScanned int64 - for ; blockIterator.Valid(); blockIterator.Next() { - blockScanned++ - stats.Scanned++ - if progress != nil { - progress("block-event", blockScanned) + for height := target + 1; height <= sourceHeight; height++ { + response, err := loadFinalizeBlock(height) + if err != nil { + return stats, fmt.Errorf("load FinalizeBlockResponse at height %d for block indexes: %w", height, err) } - height, n := binary.Varint(blockIterator.Value()) - if n <= 0 { - _ = blockIterator.Close() - return stats, fmt.Errorf( - "decode block-event index height key=%X value=%X", - blockIterator.Key(), - blockIterator.Value(), - ) + keys, err := indexedBlockEventKeys(db, height, response.GetEvents()) + if err != nil { + return stats, err } - if height > target { - deleteDataKeys = append(deleteDataKeys, bytes.Clone(blockIterator.Key())) - stats.BlockEventKeys++ + deleteDataKeys = append(deleteDataKeys, keys...) + stats.BlockEventKeys += int64(len(keys)) + stats.Scanned += int64(len(keys)) + if progress != nil { + progress("block-height", height-target) } } - if err := blockIterator.Error(); err != nil { - _ = blockIterator.Close() - return stats, fmt.Errorf("scan CometBFT block events: %w", err) - } - if err := blockIterator.Close(); err != nil { - return stats, fmt.Errorf("close CometBFT block-event iterator: %w", err) - } if err := writeSetBatches(func() indexBatch { return db.NewBatch() }, restoreValues); err != nil { return stats, fmt.Errorf("restore retained CometBFT transaction indexes: %w", err) @@ -464,6 +461,95 @@ func txEventPrefixes(result *cmtabci.TxResult) [][]byte { return resultPrefixes } +func indexedBlockEventKeys(db cmtdb.DB, height int64, events []cmtabci.Event) ([][]byte, error) { + var keys [][]byte + heightKey, err := orderedcode.Append(nil, cmttypes.BlockHeightKey, height) + if err != nil { + return nil, fmt.Errorf("encode block-height index at height %d: %w", height, err) + } + heightKey = append(bytes.Clone(blockEventPrefix), heightKey...) + value, err := db.Get(heightKey) + if err != nil { + return nil, fmt.Errorf("read block-height index at height %d: %w", height, err) + } + if value != nil { + if err := validateBlockEventHeight(heightKey, value, height); err != nil { + return nil, err + } + keys = append(keys, heightKey) + } + + prefixes, err := blockEventPrefixes(events, height) + if err != nil { + return nil, err + } + for _, prefix := range prefixes { + prefix = append(bytes.Clone(blockEventPrefix), prefix...) + iterator, err := db.Iterator(prefix, prefixEnd(prefix)) + if err != nil { + return nil, fmt.Errorf("iterate block-event prefix %X: %w", prefix, err) + } + for ; iterator.Valid(); iterator.Next() { + if err := validateBlockEventHeight(iterator.Key(), iterator.Value(), height); err != nil { + _ = iterator.Close() + return nil, err + } + keys = append(keys, bytes.Clone(iterator.Key())) + } + if err := iterator.Error(); err != nil { + _ = iterator.Close() + return nil, fmt.Errorf("scan block-event prefix %X: %w", prefix, err) + } + if err := iterator.Close(); err != nil { + return nil, fmt.Errorf("close block-event iterator: %w", err) + } + } + return keys, nil +} + +func blockEventPrefixes(events []cmtabci.Event, height int64) ([][]byte, error) { + prefixes := make(map[string][]byte) + for _, event := range events { + if len(event.Type) == 0 { + continue + } + for _, attribute := range event.Attributes { + if len(attribute.Key) == 0 || !attribute.GetIndex() { + continue + } + compositeKey := fmt.Sprintf("%s.%s", event.Type, attribute.Key) + if compositeKey == cmttypes.BlockHeightKey { + return nil, fmt.Errorf("block event uses reserved index key %q at height %d", compositeKey, height) + } + prefix, err := orderedcode.Append(nil, compositeKey, attribute.Value, height) + if err != nil { + return nil, fmt.Errorf("encode block-event index at height %d: %w", height, err) + } + prefixes[string(prefix)] = prefix + } + } + + result := make([][]byte, 0, len(prefixes)) + for _, prefix := range prefixes { + result = append(result, prefix) + } + return result, nil +} + +func validateBlockEventHeight(key, value []byte, expected int64) error { + height, n := binary.Varint(value) + if n <= 0 || height != expected { + return fmt.Errorf( + "block-event index %X has height %d (value=%X), expected %d", + key, + height, + value, + expected, + ) + } + return nil +} + func prefixEnd(prefix []byte) []byte { end := bytes.Clone(prefix) for index := len(end) - 1; index >= 0; index-- { @@ -475,6 +561,24 @@ func prefixEnd(prefix []byte) []byte { return nil } +func lastFinalizeBlockHeight(db cmtdb.DB) (int64, error) { + value, err := db.Get([]byte("lastABCIResponseKey")) + if err != nil { + return 0, fmt.Errorf("read last FinalizeBlockResponse: %w", err) + } + if value == nil { + return 0, fmt.Errorf("last FinalizeBlockResponse is missing") + } + var info cmtstateproto.ABCIResponsesInfo + if err := info.Unmarshal(value); err != nil { + return 0, fmt.Errorf("decode last FinalizeBlockResponse: %w", err) + } + if info.Height <= 0 { + return 0, fmt.Errorf("last FinalizeBlockResponse has invalid height %d", info.Height) + } + return info.Height, nil +} + func txHeightOccurrence(key, value []byte) (int64, uint32, error) { if len(value) != 32 { return 0, 0, fmt.Errorf("transaction-height index %X has invalid hash length %d", key, len(value)) diff --git a/cmd/cronosd/cmd/rewind_indexes_test.go b/cmd/cronosd/cmd/rewind_indexes_test.go index 00d04a2..2278d05 100644 --- a/cmd/cronosd/cmd/rewind_indexes_test.go +++ b/cmd/cronosd/cmd/rewind_indexes_test.go @@ -7,6 +7,7 @@ import ( cmtdb "github.com/cometbft/cometbft-db" cmtabci "github.com/cometbft/cometbft/abci/types" cmtquery "github.com/cometbft/cometbft/libs/pubsub/query" + cmtstateproto "github.com/cometbft/cometbft/proto/tendermint/state" cmtblockidxkv "github.com/cometbft/cometbft/state/indexer/block/kv" cmttxidxkv "github.com/cometbft/cometbft/state/txindex/kv" cmttypes "github.com/cometbft/cometbft/types" @@ -28,7 +29,7 @@ func TestTrimCometIndexes(t *testing.T) { require.NoError(t, blockIndexer.Index(indexTestBlockEvents(10))) require.NoError(t, blockIndexer.Index(indexTestBlockEvents(11))) - stats, err := trimCometIndexes(db, 10, indexTestFinalizeLoader(lowTx, highTx), nil) + stats, err := trimCometIndexes(db, 10, 11, indexTestFinalizeLoader(lowTx, highTx), nil) require.NoError(t, err) require.EqualValues(t, 1, stats.TxHeightKeys) require.EqualValues(t, 1, stats.TxEventKeys) @@ -54,6 +55,9 @@ func TestTrimCometIndexes(t *testing.T) { highBlocks, err := blockIndexer.Search(t.Context(), cmtquery.MustCompile("block.height = 11")) require.NoError(t, err) require.Empty(t, highBlocks) + indexedBlocks, err := blockIndexer.Search(t.Context(), cmtquery.MustCompile("test.height = 'indexed'")) + require.NoError(t, err) + require.Equal(t, []int64{10}, indexedBlocks) } func TestTrimCometIndexesRestoresRetainedDuplicateTransaction(t *testing.T) { @@ -65,7 +69,7 @@ func TestTrimCometIndexesRestoresRetainedDuplicateTransaction(t *testing.T) { require.NoError(t, txIndexer.Index(lowTx)) require.NoError(t, txIndexer.Index(highTx)) - stats, err := trimCometIndexes(db, 10, indexTestFinalizeLoader(lowTx, highTx), nil) + stats, err := trimCometIndexes(db, 10, 11, indexTestFinalizeLoader(lowTx, highTx), nil) require.NoError(t, err) require.EqualValues(t, 1, stats.TxHeightKeys) require.EqualValues(t, 1, stats.TxEventKeys) @@ -83,7 +87,7 @@ func TestTrimCometIndexesRestoresRetainedDuplicateTransaction(t *testing.T) { require.NoError(t, err) require.Empty(t, highTransactions) - stats, err = trimCometIndexes(db, 10, indexTestFinalizeLoader(lowTx, highTx), nil) + stats, err = trimCometIndexes(db, 10, 11, indexTestFinalizeLoader(lowTx, highTx), nil) require.NoError(t, err) require.Zero(t, stats.TxHeightKeys) require.Zero(t, stats.TxEventKeys) @@ -119,6 +123,18 @@ func TestPrefixEnd(t *testing.T) { require.Nil(t, prefixEnd([]byte{0xFF, 0xFF})) } +func TestLastFinalizeBlockHeight(t *testing.T) { + db := cmtdb.NewMemDB() + info := &cmtstateproto.ABCIResponsesInfo{Height: 11} + value, err := info.Marshal() + require.NoError(t, err) + require.NoError(t, db.Set([]byte("lastABCIResponseKey"), value)) + + height, err := lastFinalizeBlockHeight(db) + require.NoError(t, err) + require.EqualValues(t, 11, height) +} + func indexTestTxResult(height int64, value string) *cmtabci.TxResult { return indexTestTxResultWithTx(height, value, value) } @@ -143,7 +159,9 @@ func indexTestFinalizeLoader(results ...*cmtabci.TxResult) func(int64) (*cmtabci for _, result := range results { response := responses[result.Height] if response == nil { - response = &cmtabci.ResponseFinalizeBlock{} + response = &cmtabci.ResponseFinalizeBlock{ + Events: indexTestBlockEvents(result.Height).Events, + } responses[result.Height] = response } for uint32(len(response.TxResults)) <= result.Index { From ac72fd406936ef0108b428183a99f1dac80e8c46 Mon Sep 17 00:00:00 2001 From: lihe Date: Tue, 1 Sep 2026 17:45:58 +0800 Subject: [PATCH 6/7] fix: trim CometBFT state history during rewind Signed-off-by: lihe --- cmd/cronosd/cmd/rewind_indexes.go | 134 ++++++++++++++++++++++++- cmd/cronosd/cmd/rewind_indexes_test.go | 65 ++++++++++++ 2 files changed, 194 insertions(+), 5 deletions(-) diff --git a/cmd/cronosd/cmd/rewind_indexes.go b/cmd/cronosd/cmd/rewind_indexes.go index 7b2bc81..ab09567 100644 --- a/cmd/cronosd/cmd/rewind_indexes.go +++ b/cmd/cronosd/cmd/rewind_indexes.go @@ -45,6 +45,12 @@ type indexTrimStats struct { EVMTxHashKeys int64 } +type stateTrimStats struct { + ABCIResponses int64 + Validators int64 + ConsensusParams int64 +} + // RewindIndexesCmd removes transaction and block-event index entries above a // target height. It is idempotent and must be run with cronosd stopped. func RewindIndexesCmd() *cobra.Command { @@ -78,12 +84,33 @@ func RewindIndexesCmd() *cobra.Command { stateStore := cmtstate.NewStore(stateDB, cmtstate.StoreOptions{ DiscardABCIResponses: cfg.Storage.DiscardABCIResponses, }) + currentState, err := stateStore.Load() + if err != nil { + _ = txIndexDB.Close() + _ = stateDB.Close() + return fmt.Errorf("load CometBFT state: %w", err) + } + if currentState.LastBlockHeight != target { + _ = txIndexDB.Close() + _ = stateDB.Close() + return fmt.Errorf( + "CometBFT state height %d does not match target %d", + currentState.LastBlockHeight, + target, + ) + } sourceHeight, err := lastFinalizeBlockHeight(stateDB) if err != nil { _ = txIndexDB.Close() _ = stateDB.Close() return err } + targetResponse, err := stateStore.LoadFinalizeBlockResponse(target) + if err != nil { + _ = txIndexDB.Close() + _ = stateDB.Close() + return fmt.Errorf("load target FinalizeBlockResponse at height %d: %w", target, err) + } stats, err := trimCometIndexes(txIndexDB, target, sourceHeight, stateStore.LoadFinalizeBlockResponse, func(stage string, scanned int64) { if stage == "block-height" && scanned%rewindProgressInterval == 0 || @@ -92,34 +119,44 @@ func RewindIndexesCmd() *cobra.Command { } }) txCloseErr := txIndexDB.Close() - stateCloseErr := stateDB.Close() if err != nil { + _ = stateDB.Close() return err } if txCloseErr != nil { + _ = stateDB.Close() return fmt.Errorf("close CometBFT index database: %w", txCloseErr) } - if stateCloseErr != nil { - return fmt.Errorf("close CometBFT state database: %w", stateCloseErr) - } evmIndexDB, err := ethermintserver.OpenIndexerDB(cfg.RootDir, server.GetAppDBBackend(ctx.Viper)) if err != nil { + _ = stateDB.Close() return fmt.Errorf("open EVM index database: %w", err) } evmStats, err := trimEVMIndexes(evmIndexDB, target) closeErr := evmIndexDB.Close() if err != nil { + _ = stateDB.Close() return err } if closeErr != nil { + _ = stateDB.Close() return fmt.Errorf("close EVM index database: %w", closeErr) } + stateStats, err := trimCometStateHistory(stateDB, target, sourceHeight, targetResponse) + stateCloseErr := stateDB.Close() + if err != nil { + return err + } + if stateCloseErr != nil { + return fmt.Errorf("close CometBFT state database: %w", stateCloseErr) + } + stats.EVMTxIndexKeys = evmStats.EVMTxIndexKeys stats.EVMTxHashKeys = evmStats.EVMTxHashKeys cmd.Printf( - "rewound indexes to height %d scanned=%d tx_heights=%d tx_events=%d tx_hashes=%d tx_hashes_restored=%d block_events=%d evm_tx_indexes=%d evm_tx_hashes=%d\n", + "rewound indexes and state history to height %d scanned=%d tx_heights=%d tx_events=%d tx_hashes=%d tx_hashes_restored=%d block_events=%d evm_tx_indexes=%d evm_tx_hashes=%d state_abci_responses=%d state_validators=%d state_consensus_params=%d\n", target, stats.Scanned, stats.TxHeightKeys, @@ -129,12 +166,99 @@ func RewindIndexesCmd() *cobra.Command { stats.BlockEventKeys, stats.EVMTxIndexKeys, stats.EVMTxHashKeys, + stateStats.ABCIResponses, + stateStats.Validators, + stateStats.ConsensusParams, ) return nil }, } } +func trimCometStateHistory( + db cmtdb.DB, + target int64, + sourceHeight int64, + targetResponse *cmtabci.ResponseFinalizeBlock, +) (stateTrimStats, error) { + var stats stateTrimStats + if sourceHeight < target { + return stats, fmt.Errorf("source height %d is below target %d", sourceHeight, target) + } + if targetResponse == nil { + return stats, fmt.Errorf("target FinalizeBlockResponse is nil") + } + + var deleteKeys [][]byte + appendIfPresent := func(key []byte, count *int64) error { + present, err := db.Has(key) + if err != nil { + return err + } + if present { + deleteKeys = append(deleteKeys, key) + *count = *count + 1 + } + return nil + } + + for height := target + 1; height <= sourceHeight; height++ { + if err := appendIfPresent(cometStateHeightKey("abciResponsesKey", height), &stats.ABCIResponses); err != nil { + return stats, fmt.Errorf("inspect FinalizeBlockResponse at height %d: %w", height, err) + } + } + // Saving CometBFT state at height H writes validator metadata for H+2 and + // consensus parameters for H+1. The final hard rollback recreated those + // boundary entries, so only keys beyond them belong to the discarded suffix. + for height := target + 3; height <= sourceHeight+2; height++ { + if err := appendIfPresent(cometStateHeightKey("validatorsKey", height), &stats.Validators); err != nil { + return stats, fmt.Errorf("inspect validators at height %d: %w", height, err) + } + } + for height := target + 2; height <= sourceHeight+1; height++ { + if err := appendIfPresent(cometStateHeightKey("consensusParamsKey", height), &stats.ConsensusParams); err != nil { + return stats, fmt.Errorf("inspect consensus params at height %d: %w", height, err) + } + } + + if err := writeDeleteBatches(func() indexBatch { return db.NewBatch() }, deleteKeys); err != nil { + return stats, fmt.Errorf("delete CometBFT state history suffix: %w", err) + } + for _, key := range deleteKeys { + present, err := db.Has(key) + if err != nil { + return stats, fmt.Errorf("verify deleted CometBFT state key %q: %w", key, err) + } + if present { + return stats, fmt.Errorf("CometBFT state key %q still exists after deletion", key) + } + } + + responseInfo := &cmtstateproto.ABCIResponsesInfo{ + ResponseFinalizeBlock: targetResponse, + Height: target, + } + value, err := responseInfo.Marshal() + if err != nil { + return stats, fmt.Errorf("encode last FinalizeBlockResponse at height %d: %w", target, err) + } + if err := db.SetSync([]byte("lastABCIResponseKey"), value); err != nil { + return stats, fmt.Errorf("set last FinalizeBlockResponse at height %d: %w", target, err) + } + lastHeight, err := lastFinalizeBlockHeight(db) + if err != nil { + return stats, err + } + if lastHeight != target { + return stats, fmt.Errorf("last FinalizeBlockResponse height %d does not match target %d", lastHeight, target) + } + return stats, nil +} + +func cometStateHeightKey(prefix string, height int64) []byte { + return []byte(fmt.Sprintf("%s:%d", prefix, height)) +} + type indexedTransaction struct { height int64 index uint32 diff --git a/cmd/cronosd/cmd/rewind_indexes_test.go b/cmd/cronosd/cmd/rewind_indexes_test.go index 2278d05..5f4152c 100644 --- a/cmd/cronosd/cmd/rewind_indexes_test.go +++ b/cmd/cronosd/cmd/rewind_indexes_test.go @@ -135,6 +135,64 @@ func TestLastFinalizeBlockHeight(t *testing.T) { require.EqualValues(t, 11, height) } +func TestTrimCometStateHistory(t *testing.T) { + db := cmtdb.NewMemDB() + target := int64(10) + sourceHeight := int64(12) + targetResponse := &cmtabci.ResponseFinalizeBlock{AppHash: []byte("target")} + + for height := int64(9); height <= sourceHeight; height++ { + require.NoError(t, db.Set(cometStateHeightKey("abciResponsesKey", height), []byte("response"))) + } + for height := int64(9); height <= sourceHeight+2; height++ { + require.NoError(t, db.Set(cometStateHeightKey("validatorsKey", height), []byte("validators"))) + } + for height := int64(9); height <= sourceHeight+1; height++ { + require.NoError(t, db.Set(cometStateHeightKey("consensusParamsKey", height), []byte("params"))) + } + sourceInfo := &cmtstateproto.ABCIResponsesInfo{ + ResponseFinalizeBlock: &cmtabci.ResponseFinalizeBlock{AppHash: []byte("source")}, + Height: sourceHeight, + } + value, err := sourceInfo.Marshal() + require.NoError(t, err) + require.NoError(t, db.Set([]byte("lastABCIResponseKey"), value)) + require.NoError(t, db.Set([]byte("unrelated"), []byte("keep"))) + + stats, err := trimCometStateHistory(db, target, sourceHeight, targetResponse) + require.NoError(t, err) + require.EqualValues(t, 2, stats.ABCIResponses) + require.EqualValues(t, 2, stats.Validators) + require.EqualValues(t, 2, stats.ConsensusParams) + + for height := target + 1; height <= sourceHeight; height++ { + require.False(t, cometDBHas(t, db, cometStateHeightKey("abciResponsesKey", height))) + } + for height := target + 3; height <= sourceHeight+2; height++ { + require.False(t, cometDBHas(t, db, cometStateHeightKey("validatorsKey", height))) + } + for height := target + 2; height <= sourceHeight+1; height++ { + require.False(t, cometDBHas(t, db, cometStateHeightKey("consensusParamsKey", height))) + } + require.True(t, cometDBHas(t, db, cometStateHeightKey("abciResponsesKey", target))) + require.True(t, cometDBHas(t, db, cometStateHeightKey("validatorsKey", target+2))) + require.True(t, cometDBHas(t, db, cometStateHeightKey("consensusParamsKey", target+1))) + require.True(t, cometDBHas(t, db, []byte("unrelated"))) + + lastHeight, err := lastFinalizeBlockHeight(db) + require.NoError(t, err) + require.Equal(t, target, lastHeight) +} + +func TestTrimCometStateHistoryRejectsInvalidInput(t *testing.T) { + db := cmtdb.NewMemDB() + + _, err := trimCometStateHistory(db, 10, 9, &cmtabci.ResponseFinalizeBlock{}) + require.EqualError(t, err, "source height 9 is below target 10") + _, err = trimCometStateHistory(db, 10, 10, nil) + require.EqualError(t, err, "target FinalizeBlockResponse is nil") +} + func indexTestTxResult(height int64, value string) *cmtabci.TxResult { return indexTestTxResultWithTx(height, value, value) } @@ -198,3 +256,10 @@ func dbHas(t *testing.T, db cosmosdb.DB, key []byte) bool { require.NoError(t, err) return found } + +func cometDBHas(t *testing.T, db cmtdb.DB, key []byte) bool { + t.Helper() + found, err := db.Has(key) + require.NoError(t, err) + return found +} From 3d277191e235603cc14b4e2f85d566b365a29ed8 Mon Sep 17 00:00:00 2001 From: lihe Date: Tue, 1 Sep 2026 17:55:17 +0800 Subject: [PATCH 7/7] fix: trim evidence suffix during rewind Signed-off-by: lihe --- cmd/cronosd/cmd/rewind_indexes.go | 85 +++++++++++++++++++++++++- cmd/cronosd/cmd/rewind_indexes_test.go | 35 +++++++++++ 2 files changed, 119 insertions(+), 1 deletion(-) diff --git a/cmd/cronosd/cmd/rewind_indexes.go b/cmd/cronosd/cmd/rewind_indexes.go index ab09567..59d2a8b 100644 --- a/cmd/cronosd/cmd/rewind_indexes.go +++ b/cmd/cronosd/cmd/rewind_indexes.go @@ -51,6 +51,13 @@ type stateTrimStats struct { ConsensusParams int64 } +type evidenceTrimStats struct { + Scanned int64 + Pending int64 + Committed int64 + Deleted int64 +} + // RewindIndexesCmd removes transaction and block-event index entries above a // target height. It is idempotent and must be run with cronosd stopped. func RewindIndexesCmd() *cobra.Command { @@ -144,6 +151,22 @@ func RewindIndexesCmd() *cobra.Command { return fmt.Errorf("close EVM index database: %w", closeErr) } + evidenceDB, err := cmtcfg.DefaultDBProvider(&cmtcfg.DBContext{ID: "evidence", Config: cfg}) + if err != nil { + _ = stateDB.Close() + return fmt.Errorf("open CometBFT evidence database: %w", err) + } + evidenceStats, err := trimEvidence(evidenceDB, target) + evidenceCloseErr := evidenceDB.Close() + if err != nil { + _ = stateDB.Close() + return err + } + if evidenceCloseErr != nil { + _ = stateDB.Close() + return fmt.Errorf("close CometBFT evidence database: %w", evidenceCloseErr) + } + stateStats, err := trimCometStateHistory(stateDB, target, sourceHeight, targetResponse) stateCloseErr := stateDB.Close() if err != nil { @@ -156,7 +179,7 @@ func RewindIndexesCmd() *cobra.Command { stats.EVMTxIndexKeys = evmStats.EVMTxIndexKeys stats.EVMTxHashKeys = evmStats.EVMTxHashKeys cmd.Printf( - "rewound indexes and state history to height %d scanned=%d tx_heights=%d tx_events=%d tx_hashes=%d tx_hashes_restored=%d block_events=%d evm_tx_indexes=%d evm_tx_hashes=%d state_abci_responses=%d state_validators=%d state_consensus_params=%d\n", + "rewound indexes and state history to height %d scanned=%d tx_heights=%d tx_events=%d tx_hashes=%d tx_hashes_restored=%d block_events=%d evm_tx_indexes=%d evm_tx_hashes=%d evidence_scanned=%d evidence_pending=%d evidence_committed=%d evidence_deleted=%d state_abci_responses=%d state_validators=%d state_consensus_params=%d\n", target, stats.Scanned, stats.TxHeightKeys, @@ -166,6 +189,10 @@ func RewindIndexesCmd() *cobra.Command { stats.BlockEventKeys, stats.EVMTxIndexKeys, stats.EVMTxHashKeys, + evidenceStats.Scanned, + evidenceStats.Pending, + evidenceStats.Committed, + evidenceStats.Deleted, stateStats.ABCIResponses, stateStats.Validators, stateStats.ConsensusParams, @@ -175,6 +202,62 @@ func RewindIndexesCmd() *cobra.Command { } } +func trimEvidence(db cmtdb.DB, target int64) (evidenceTrimStats, error) { + var stats evidenceTrimStats + var deleteKeys [][]byte + iterator, err := db.Iterator(nil, nil) + if err != nil { + return stats, fmt.Errorf("iterate CometBFT evidence: %w", err) + } + for ; iterator.Valid(); iterator.Next() { + key := iterator.Key() + stats.Scanned++ + if len(key) <= 18 || key[17] != '/' { + _ = iterator.Close() + return stats, fmt.Errorf("evidence key has invalid format: %X", key) + } + switch key[0] { + case 0: + stats.Committed++ + case 1: + stats.Pending++ + default: + _ = iterator.Close() + return stats, fmt.Errorf("evidence key has invalid prefix: %X", key) + } + height, err := strconv.ParseInt(string(key[1:17]), 16, 64) + if err != nil || height <= 0 { + _ = iterator.Close() + return stats, fmt.Errorf("evidence key has invalid height: %X", key) + } + if height > target { + deleteKeys = append(deleteKeys, bytes.Clone(key)) + stats.Deleted++ + } + } + if err := iterator.Error(); err != nil { + _ = iterator.Close() + return stats, fmt.Errorf("scan CometBFT evidence: %w", err) + } + if err := iterator.Close(); err != nil { + return stats, fmt.Errorf("close CometBFT evidence iterator: %w", err) + } + + if err := writeDeleteBatches(func() indexBatch { return db.NewBatch() }, deleteKeys); err != nil { + return stats, fmt.Errorf("delete CometBFT evidence suffix: %w", err) + } + for _, key := range deleteKeys { + present, err := db.Has(key) + if err != nil { + return stats, fmt.Errorf("verify deleted evidence key %X: %w", key, err) + } + if present { + return stats, fmt.Errorf("evidence key %X still exists after deletion", key) + } + } + return stats, nil +} + func trimCometStateHistory( db cmtdb.DB, target int64, diff --git a/cmd/cronosd/cmd/rewind_indexes_test.go b/cmd/cronosd/cmd/rewind_indexes_test.go index 5f4152c..2749cde 100644 --- a/cmd/cronosd/cmd/rewind_indexes_test.go +++ b/cmd/cronosd/cmd/rewind_indexes_test.go @@ -193,6 +193,41 @@ func TestTrimCometStateHistoryRejectsInvalidInput(t *testing.T) { require.EqualError(t, err, "target FinalizeBlockResponse is nil") } +func TestTrimEvidence(t *testing.T) { + db := cmtdb.NewMemDB() + lowCommitted := evidenceTestKey(0, 10, "AABB") + lowPending := evidenceTestKey(1, 10, "CCDD") + highCommitted := evidenceTestKey(0, 11, "EEFF") + highPending := evidenceTestKey(1, 12, "0011") + + for _, key := range [][]byte{lowCommitted, lowPending, highCommitted, highPending} { + require.NoError(t, db.Set(key, []byte("evidence"))) + } + + stats, err := trimEvidence(db, 10) + require.NoError(t, err) + require.EqualValues(t, 4, stats.Scanned) + require.EqualValues(t, 2, stats.Committed) + require.EqualValues(t, 2, stats.Pending) + require.EqualValues(t, 2, stats.Deleted) + require.True(t, cometDBHas(t, db, lowCommitted)) + require.True(t, cometDBHas(t, db, lowPending)) + require.False(t, cometDBHas(t, db, highCommitted)) + require.False(t, cometDBHas(t, db, highPending)) +} + +func TestTrimEvidenceRejectsMalformedKey(t *testing.T) { + db := cmtdb.NewMemDB() + require.NoError(t, db.Set([]byte("invalid"), []byte("evidence"))) + + _, err := trimEvidence(db, 10) + require.EqualError(t, err, "evidence key has invalid format: 696E76616C6964") +} + +func evidenceTestKey(prefix byte, height int64, hash string) []byte { + return append([]byte{prefix}, []byte(fmt.Sprintf("%016X/%s", height, hash))...) +} + func indexTestTxResult(height int64, value string) *cmtabci.TxResult { return indexTestTxResultWithTx(height, value, value) }