diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml new file mode 100644 index 0000000000..9e3749e228 --- /dev/null +++ b/.github/workflows/codeql.yml @@ -0,0 +1,57 @@ +name: "CodeQL" + +on: + pull_request: + paths: + - "**.go" + push: + branches: + - main + - release/** + paths: + - "**.go" + +jobs: + analyze: + name: Analyze + runs-on: ubuntu-latest + permissions: + actions: read + contents: read + security-events: write + + steps: + - name: Checkout repository + uses: actions/checkout@v3 + - uses: actions/setup-go@v6 + with: + go-version: '1.25.6' + # Initializes the CodeQL tools for scanning. + - name: Initialize CodeQL + uses: github/codeql-action/init@v2 + with: + languages: "go" + queries: crypto-com/cosmos-sdk-codeql@main,security-and-quality + # If you wish to specify custom queries, you can do so here or in a config file. + # By default, queries listed here will override any specified in a config file. + # Prefix the list here with "+" to use these queries and those in the config file. + # queries: ./path/to/local/query, your-org/your-repo/queries@main + + # Autobuild attempts to build any compiled languages (C/C++, C#, or Java). + # If this step fails, then you should remove it and run the build manually (see below) + - name: Autobuild + uses: github/codeql-action/autobuild@v2 + + # ℹī¸ Command-line programs to run using the OS shell. + # 📚 https://git.io/JvXDl + + # ✏ī¸ If the Autobuild fails above, remove it and uncomment the following three lines + # and modify them (or add more) to build your code if your project + # uses a compiled language + + #- run: | + # make bootstrap + # make release + + - name: Perform CodeQL Analysis + uses: github/codeql-action/analyze@v2 diff --git a/app/app.go b/app/app.go index 4af12a9bf6..85e844d6a6 100644 --- a/app/app.go +++ b/app/app.go @@ -35,6 +35,7 @@ import ( "github.com/sei-protocol/sei-chain/sei-cosmos/codec" "github.com/sei-protocol/sei-chain/sei-cosmos/codec/types" cryptotypes "github.com/sei-protocol/sei-chain/sei-cosmos/crypto/types" + "github.com/sei-protocol/sei-chain/sei-cosmos/server" "github.com/sei-protocol/sei-chain/sei-cosmos/server/api" "github.com/sei-protocol/sei-chain/sei-cosmos/server/config" servertypes "github.com/sei-protocol/sei-chain/sei-cosmos/server/types" @@ -117,6 +118,7 @@ import ( gigautils "github.com/sei-protocol/sei-chain/giga/executor/utils" "github.com/sei-protocol/sei-chain/precompiles" putils "github.com/sei-protocol/sei-chain/precompiles/utils" + ssconfig "github.com/sei-protocol/sei-chain/sei-db/config" "github.com/sei-protocol/sei-chain/sei-ibc-go/modules/apps/transfer" ibctransferkeeper "github.com/sei-protocol/sei-chain/sei-ibc-go/modules/apps/transfer/keeper" ibctransfertypes "github.com/sei-protocol/sei-chain/sei-ibc-go/modules/apps/transfer/types" @@ -676,10 +678,10 @@ func New( wasmOpts..., ) - receiptConfig, err := readReceiptStoreConfig(homePath, appOpts) - if err != nil { - panic(fmt.Sprintf("error reading receipt store config: %s", err)) - } + receiptStorePath := filepath.Join(homePath, "data", "receipt.db") + receiptConfig := ssconfig.DefaultReceiptStoreConfig() + receiptConfig.DBDirectory = receiptStorePath + receiptConfig.KeepRecent = cast.ToInt(appOpts.Get(server.FlagMinRetainBlocks)) if app.receiptStore == nil { receiptStore, err := receipt.NewReceiptStore(logger, receiptConfig, keys[evmtypes.StoreKey]) if err != nil { @@ -747,7 +749,7 @@ func New( if gigaExecutorConfig.OCCEnabled { logger.Info("benchmark: Giga Executor with OCC is ENABLED - using new EVM execution path with parallel execution") } else { - logger.Info("benchmark: Giga Executor is ENABLED - using new EVM execution path (sequential)") + logger.Info("benchmark: Giga Executor (evmone-based) is ENABLED - using new EVM execution path (sequential)") } } else { logger.Info("benchmark: Giga Executor is DISABLED - using default GETH interpreter") @@ -1544,89 +1546,71 @@ func (app *App) ProcessTXsWithOCCV2(ctx sdk.Context, txs [][]byte, typedTxs []sd func (app *App) ProcessTXsWithOCCGiga(ctx sdk.Context, txs [][]byte, typedTxs []sdk.Tx) ([]*abci.ExecTxResult, sdk.Context) { evmEntries := make([]*sdk.DeliverTxEntry, 0, len(txs)) v2Entries := make([]*sdk.DeliverTxEntry, 0, len(txs)) - firstCosmosSeen := false for txIndex, tx := range txs { if app.GetEVMMsg(typedTxs[txIndex]) != nil { - if firstCosmosSeen { - ctx.Logger().Error("Giga OCC cannot execute block due to tx ordering, falling back to V2") - // Oops! This isn't "all EVM txs, then all Cosmos txs" - we need to fallback to V2. - return app.ProcessTXsWithOCCV2(ctx, txs, typedTxs) - } - evmEntries = append(evmEntries, app.GetDeliverTxEntry(ctx, txIndex, tx, typedTxs[txIndex])) } else { - if !firstCosmosSeen { - firstCosmosSeen = true - } v2Entries = append(v2Entries, app.GetDeliverTxEntry(ctx, txIndex, tx, typedTxs[txIndex])) } } - var evmBatchResult []abci.ResponseDeliverTx - fallbackToV2 := false - - if len(evmEntries) > 0 { - // Run EVM txs against a cache so we can discard all changes on fallback. - evmCtx, evmCache := app.CacheContext(ctx) + // Run EVM txs against a cache so we can discard all changes on fallback. + evmCtx, evmCache := app.CacheContext(ctx) - // Cache block-level constants (identical for all txs in this block). - // Must use evmCtx (not ctx) because giga KV stores are registered in CacheContext. - cache, cacheErr := newGigaBlockCache(evmCtx, &app.GigaEvmKeeper) - if cacheErr != nil { - ctx.Logger().Error("failed to build giga block cache", "error", cacheErr, "height", ctx.BlockHeight()) - return nil, ctx - } + // Cache block-level constants (identical for all txs in this block). + // Must use evmCtx (not ctx) because giga KV stores are registered in CacheContext. + cache, cacheErr := newGigaBlockCache(evmCtx, &app.GigaEvmKeeper) + if cacheErr != nil { + ctx.Logger().Error("failed to build giga block cache", "error", cacheErr, "height", ctx.BlockHeight()) + return nil, ctx + } - // Create OCC scheduler with giga executor deliverTx capturing the cache. - evmScheduler := tasks.NewScheduler( - app.ConcurrencyWorkers(), - app.TracingInfo, - app.makeGigaDeliverTx(cache), - ) + // Create OCC scheduler with giga executor deliverTx capturing the cache. + evmScheduler := tasks.NewScheduler( + app.ConcurrencyWorkers(), + app.TracingInfo, + app.makeGigaDeliverTx(cache), + ) - var evmSchedErr error - evmBatchResult, evmSchedErr = evmScheduler.ProcessAll(evmCtx, evmEntries) - if evmSchedErr != nil { - ctx.Logger().Error("benchmark OCC scheduler error (EVM txs)", "error", evmSchedErr, "height", ctx.BlockHeight(), "txCount", len(evmEntries)) - return nil, ctx - } + evmBatchResult, evmSchedErr := evmScheduler.ProcessAll(evmCtx, evmEntries) + if evmSchedErr != nil { + // TODO: DeliverTxBatch panics in this case + // TODO: detect if it was interop, and use v2 if so + ctx.Logger().Error("benchmark OCC scheduler error (EVM txs)", "error", evmSchedErr, "height", ctx.BlockHeight(), "txCount", len(evmEntries)) + return nil, ctx + } - for _, r := range evmBatchResult { - if r.Code == gigautils.GigaAbortCode && r.Codespace == gigautils.GigaAbortCodespace { - fallbackToV2 = true - break - } + fallbackToV2 := false + for _, r := range evmBatchResult { + if r.Code == gigautils.GigaAbortCode && r.Codespace == gigautils.GigaAbortCodespace { + fallbackToV2 = true + break } + } - if fallbackToV2 { - metrics.IncrGigaFallbackToV2Counter() - // Discard all EVM changes by skipping cache writes, then re-run all txs via DeliverTx. - evmBatchResult = nil - v2Entries = make([]*sdk.DeliverTxEntry, len(txs)) - for txIndex, tx := range txs { - v2Entries[txIndex] = app.GetDeliverTxEntry(ctx, txIndex, tx, typedTxs[txIndex]) - } - } else { - // Commit EVM cache to main store before processing non-EVM txs. - evmCache.Write() - evmCtx.GigaMultiStore().WriteGiga() + if fallbackToV2 { + metrics.IncrGigaFallbackToV2Counter() + // Discard all EVM changes by skipping cache writes, then re-run all txs via DeliverTx. + evmBatchResult = nil + v2Entries = make([]*sdk.DeliverTxEntry, len(txs)) + for txIndex, tx := range txs { + v2Entries[txIndex] = app.GetDeliverTxEntry(ctx, txIndex, tx, typedTxs[txIndex]) } + } else { + // Commit EVM cache to main store before processing non-EVM txs. + evmCache.Write() + evmCtx.GigaMultiStore().WriteGiga() } - var v2BatchResult []abci.ResponseDeliverTx - - if len(v2Entries) > 0 { - v2Scheduler := tasks.NewScheduler( - app.ConcurrencyWorkers(), - app.TracingInfo, - app.DeliverTx, - ) - var v2SchedErr error - v2BatchResult, v2SchedErr = v2Scheduler.ProcessAll(ctx, v2Entries) - if v2SchedErr != nil { - ctx.Logger().Error("benchmark OCC scheduler error", "error", v2SchedErr, "height", ctx.BlockHeight(), "txCount", len(v2Entries)) - return nil, ctx - } + v2Scheduler := tasks.NewScheduler( + app.ConcurrencyWorkers(), + app.TracingInfo, + app.DeliverTx, + ) + v2BatchResult, v2SchedErr := v2Scheduler.ProcessAll(ctx, v2Entries) + if v2SchedErr != nil { + ctx.Logger().Error("benchmark OCC scheduler error", "error", v2SchedErr, "height", ctx.BlockHeight(), "txCount", len(v2Entries)) + return nil, ctx } execResults := make([]*abci.ExecTxResult, 0, len(evmBatchResult)+len(v2BatchResult)) diff --git a/app/receipt_store_config.go b/app/receipt_store_config.go deleted file mode 100644 index 8d2acfacf2..0000000000 --- a/app/receipt_store_config.go +++ /dev/null @@ -1,26 +0,0 @@ -package app - -import ( - "path/filepath" - - seidbconfig "github.com/sei-protocol/sei-chain/sei-db/config" -) - -const ( - receiptStoreBackendKey = "receipt-store.rs-backend" - receiptStoreDBDirectoryKey = "receipt-store.db-directory" - receiptStoreAsyncWriteBufferKey = "receipt-store.async-write-buffer" - receiptStoreKeepRecentKey = "receipt-store.keep-recent" - receiptStorePruneIntervalSecondsKey = "receipt-store.prune-interval-seconds" -) - -func readReceiptStoreConfig(homePath string, appOpts seidbconfig.AppOptions) (seidbconfig.ReceiptStoreConfig, error) { - receiptConfig, err := seidbconfig.ReadReceiptConfig(appOpts) - if err != nil { - return receiptConfig, err - } - if receiptConfig.DBDirectory == "" { - receiptConfig.DBDirectory = filepath.Join(homePath, "data", "receipt.db") - } - return receiptConfig, nil -} diff --git a/app/seidb.go b/app/seidb.go index ed1fdb7353..448de9960f 100644 --- a/app/seidb.go +++ b/app/seidb.go @@ -30,7 +30,6 @@ const ( FlagSCHistoricalProofBurst = "state-commit.sc-historical-proof-burst" FlagSCWriteMode = "state-commit.sc-write-mode" FlagSCReadMode = "state-commit.sc-read-mode" - FlagSCEnableLatticeHash = "state-commit.sc-enable-lattice-hash" // SS Store configs FlagSSEnable = "state-store.ss-enable" @@ -121,8 +120,6 @@ func parseSCConfigs(appOpts servertypes.AppOptions) config.StateCommitConfig { scConfig.ReadMode = parsedRM } - scConfig.EnableLatticeHash = cast.ToBool(appOpts.Get(FlagSCEnableLatticeHash)) - if v := appOpts.Get(FlagSCHistoricalProofMaxInFlight); v != nil { scConfig.HistoricalProofMaxInFlight = cast.ToInt(v) } diff --git a/app/seidb_test.go b/app/seidb_test.go index 8504f32848..94c7db5257 100644 --- a/app/seidb_test.go +++ b/app/seidb_test.go @@ -1,14 +1,10 @@ package app import ( - "path/filepath" "testing" - "github.com/sei-protocol/sei-chain/sei-cosmos/server" "github.com/sei-protocol/sei-chain/sei-db/config" - "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) type TestSeiDBAppOpts struct { @@ -17,7 +13,6 @@ type TestSeiDBAppOpts struct { func (t TestSeiDBAppOpts) Get(s string) interface{} { defaultSCConfig := config.DefaultStateCommitConfig() defaultSSConfig := config.DefaultStateStoreConfig() - defaultReceiptConfig := config.DefaultReceiptStoreConfig() switch s { case FlagSCEnable: return defaultSCConfig.Enable @@ -37,8 +32,6 @@ func (t TestSeiDBAppOpts) Get(s string) interface{} { return defaultSCConfig.MemIAVLConfig.SnapshotPrefetchThreshold case FlagSCSnapshotWriteRateMBps: return defaultSCConfig.MemIAVLConfig.SnapshotWriteRateMBps - case FlagSCEnableLatticeHash: - return defaultSCConfig.EnableLatticeHash case FlagSSEnable: return defaultSSConfig.Enable case FlagSSBackend: @@ -53,8 +46,6 @@ func (t TestSeiDBAppOpts) Get(s string) interface{} { return defaultSSConfig.PruneIntervalSeconds case FlagSSImportNumWorkers: return defaultSSConfig.ImportNumWorkers - case receiptStoreBackendKey: - return defaultReceiptConfig.Backend case FlagEVMSSDirectory: return defaultSSConfig.EVMDBDirectory case FlagEVMSSWriteMode: @@ -70,11 +61,8 @@ func TestNewDefaultConfig(t *testing.T) { appOpts := TestSeiDBAppOpts{} scConfig := parseSCConfigs(appOpts) ssConfig := parseSSConfigs(appOpts) - receiptConfig, err := config.ReadReceiptConfig(appOpts) - assert.NoError(t, err) assert.Equal(t, scConfig, config.DefaultStateCommitConfig()) assert.Equal(t, ssConfig, config.DefaultStateStoreConfig()) - assert.Equal(t, receiptConfig, config.DefaultReceiptStoreConfig()) } type mapAppOpts map[string]interface{} @@ -97,74 +85,3 @@ func TestParseSCConfigs_HistoricalProofFlags(t *testing.T) { assert.Equal(t, 12.5, scConfig.HistoricalProofRateLimit) assert.Equal(t, 3, scConfig.HistoricalProofBurst) } - -func TestParseReceiptConfigs_DefaultsToPebbleWhenUnset(t *testing.T) { - receiptConfig, err := config.ReadReceiptConfig(mapAppOpts{}) - assert.NoError(t, err) - assert.Equal(t, config.DefaultReceiptStoreConfig(), receiptConfig) -} - -func TestParseReceiptConfigs_UsesConfiguredBackend(t *testing.T) { - receiptConfig, err := config.ReadReceiptConfig(mapAppOpts{ - receiptStoreBackendKey: "parquet", - }) - assert.NoError(t, err) - assert.Equal(t, "parquet", receiptConfig.Backend) - assert.Equal(t, config.DefaultReceiptStoreConfig().AsyncWriteBuffer, receiptConfig.AsyncWriteBuffer) - assert.Equal(t, config.DefaultReceiptStoreConfig().KeepRecent, receiptConfig.KeepRecent) -} - -func TestParseReceiptConfigs_UsesConfiguredValues(t *testing.T) { - receiptConfig, err := config.ReadReceiptConfig(mapAppOpts{ - receiptStoreDBDirectoryKey: "/tmp/custom-receipt-db", - receiptStoreBackendKey: "parquet", - receiptStoreAsyncWriteBufferKey: 7, - receiptStoreKeepRecentKey: 42, - receiptStorePruneIntervalSecondsKey: 9, - }) - assert.NoError(t, err) - assert.Equal(t, "/tmp/custom-receipt-db", receiptConfig.DBDirectory) - assert.Equal(t, "parquet", receiptConfig.Backend) - assert.Equal(t, 7, receiptConfig.AsyncWriteBuffer) - assert.Equal(t, 42, receiptConfig.KeepRecent) - assert.Equal(t, 9, receiptConfig.PruneIntervalSeconds) -} - -func TestParseReceiptConfigs_RejectsInvalidBackend(t *testing.T) { - _, err := config.ReadReceiptConfig(mapAppOpts{ - receiptStoreBackendKey: "rocksdb", - }) - assert.Error(t, err) - assert.Contains(t, err.Error(), "unsupported receipt-store backend") - assert.Contains(t, err.Error(), "rocksdb") -} - -func TestReadReceiptStoreConfigUsesConfiguredValues(t *testing.T) { - homePath := t.TempDir() - receiptConfig, err := readReceiptStoreConfig(homePath, mapAppOpts{ - receiptStoreDBDirectoryKey: "/tmp/custom-receipt-db", - receiptStoreKeepRecentKey: 5, - server.FlagMinRetainBlocks: 100, - }) - require.NoError(t, err) - assert.Equal(t, "/tmp/custom-receipt-db", receiptConfig.DBDirectory) - assert.Equal(t, 5, receiptConfig.KeepRecent) -} - -func TestReadReceiptStoreConfigUsesDefaultDirectoryWhenUnset(t *testing.T) { - homePath := t.TempDir() - receiptConfig, err := readReceiptStoreConfig(homePath, mapAppOpts{}) - require.NoError(t, err) - assert.Equal(t, filepath.Join(homePath, "data", "receipt.db"), receiptConfig.DBDirectory) -} - -// TestFullAppPathWithParquetReceiptStore exercises the full app.New path with rs-backend = "parquet" -// and asserts the parquet receipt store is actually instantiated (not pebble). -func TestFullAppPathWithParquetReceiptStore(t *testing.T) { - app := SetupWithScReceiptFromOpts(t, false, false, TestAppOpts{ - UseSc: true, - ReceiptBackend: "parquet", - }) - require.NotNil(t, app.receiptStore, "receipt store should be created") - assert.Equal(t, "parquet", receipt.BackendTypeName(app.receiptStore), "receipt store backend should be parquet") -} diff --git a/app/test_helpers.go b/app/test_helpers.go index 8785e8c568..c9d5320548 100644 --- a/app/test_helpers.go +++ b/app/test_helpers.go @@ -76,10 +76,9 @@ func (t TestTx) GetGasEstimate() uint64 { } type TestAppOpts struct { - UseSc bool - EnableGiga bool - EnableGigaOCC bool - ReceiptBackend string // e.g. "parquet" to use parquet receipt store; empty = default (pebble) + UseSc bool + EnableGiga bool + EnableGigaOCC bool } func (t TestAppOpts) Get(s string) interface{} { @@ -100,9 +99,6 @@ func (t TestAppOpts) Get(s string) interface{} { if s == gigaconfig.FlagOCCEnabled { return t.EnableGigaOCC } - if s == receiptStoreBackendKey && t.ReceiptBackend != "" { - return t.ReceiptBackend - } // Disable EVM HTTP and WebSocket servers in tests to avoid port conflicts // when multiple tests run in parallel (all would try to bind to port 8545) if s == "evm.http_enabled" || s == "evm.ws_enabled" { @@ -476,55 +472,6 @@ func SetupWithDB(tb testing.TB, db dbm.DB, isCheckTx bool, enableEVMCustomPrecom return res } -// SetupWithScReceiptFromOpts is like SetupWithSc but does not inject a receipt store via AppOption. -// The receipt store is created inside New() from testAppOpts (e.g. testAppOpts.ReceiptBackend = "parquet"). -// Use this to test the full app path with rs-backend from config. -func SetupWithScReceiptFromOpts(t *testing.T, isCheckTx bool, enableEVMCustomPrecompiles bool, testAppOpts TestAppOpts, baseAppOptions ...func(*bam.BaseApp)) (res *App) { - db := dbm.NewMemDB() - encodingConfig := MakeEncodingConfig() - cdc := encodingConfig.Marshaler - - res = New( - log.NewNopLogger(), - db, - nil, - true, - map[int64]bool{}, - t.TempDir(), - 1, - enableEVMCustomPrecompiles, - config.TestConfig(), - encodingConfig, - wasm.EnableAllProposals, - testAppOpts, - EmptyWasmOpts, - nil, // no options: receipt store is created from testAppOpts inside New() - baseAppOptions..., - ) - if !isCheckTx { - genesisState := NewDefaultGenesisState(cdc) - stateBytes, err := json.MarshalIndent(genesisState, "", " ") - if err != nil { - panic(err) - } - - defer func() { _ = recover() }() - - _, err = res.InitChain( - context.Background(), &abci.RequestInitChain{ - Validators: []abci.ValidatorUpdate{}, - ConsensusParams: DefaultConsensusParams, - AppStateBytes: stateBytes, - }, - ) - if err != nil { - panic(err) - } - } - - return res -} - func SetupWithSc(t *testing.T, isCheckTx bool, enableEVMCustomPrecompiles bool, testAppOpts TestAppOpts, baseAppOptions ...func(*bam.BaseApp)) (res *App) { db := dbm.NewMemDB() encodingConfig := MakeEncodingConfig() diff --git a/cmd/seid/cmd/app_config.go b/cmd/seid/cmd/app_config.go index 22305f00cf..2ed7d4c3ae 100644 --- a/cmd/seid/cmd/app_config.go +++ b/cmd/seid/cmd/app_config.go @@ -22,25 +22,23 @@ type WASMConfig struct { type CustomAppConfig struct { srvconfig.Config - StateCommit seidbconfig.StateCommitConfig `mapstructure:"state-commit"` - StateStore seidbconfig.StateStoreConfig `mapstructure:"state-store"` - ReceiptStore seidbconfig.ReceiptStoreConfig `mapstructure:"receipt-store"` - WASM WASMConfig `mapstructure:"wasm"` - EVM evmrpcconfig.Config `mapstructure:"evm"` - GigaExecutor gigaconfig.Config `mapstructure:"giga_executor"` - ETHReplay replay.Config `mapstructure:"eth_replay"` - ETHBlockTest blocktest.Config `mapstructure:"eth_block_test"` - EvmQuery querier.Config `mapstructure:"evm_query"` - LightInvariance seiapp.LightInvarianceConfig `mapstructure:"light_invariance"` + StateCommit seidbconfig.StateCommitConfig `mapstructure:"state-commit"` + StateStore seidbconfig.StateStoreConfig `mapstructure:"state-store"` + WASM WASMConfig `mapstructure:"wasm"` + EVM evmrpcconfig.Config `mapstructure:"evm"` + GigaExecutor gigaconfig.Config `mapstructure:"giga_executor"` + ETHReplay replay.Config `mapstructure:"eth_replay"` + ETHBlockTest blocktest.Config `mapstructure:"eth_block_test"` + EvmQuery querier.Config `mapstructure:"evm_query"` + LightInvariance seiapp.LightInvarianceConfig `mapstructure:"light_invariance"` } // NewCustomAppConfig creates a CustomAppConfig with the given base config and EVM config func NewCustomAppConfig(baseConfig *srvconfig.Config, evmConfig evmrpcconfig.Config) CustomAppConfig { return CustomAppConfig{ - Config: *baseConfig, - StateCommit: seidbconfig.DefaultStateCommitConfig(), - StateStore: seidbconfig.DefaultStateStoreConfig(), - ReceiptStore: seidbconfig.DefaultReceiptStoreConfig(), + Config: *baseConfig, + StateCommit: seidbconfig.DefaultStateCommitConfig(), + StateStore: seidbconfig.DefaultStateStoreConfig(), WASM: WASMConfig{ QueryGasLimit: 300000, LruSize: 1, diff --git a/cmd/seid/cmd/init_test.go b/cmd/seid/cmd/init_test.go index 06fd629bc1..5c4f50abd4 100644 --- a/cmd/seid/cmd/init_test.go +++ b/cmd/seid/cmd/init_test.go @@ -1,13 +1,11 @@ package cmd import ( - "bytes" "encoding/json" "fmt" "os" "path/filepath" "testing" - "text/template" "github.com/sei-protocol/sei-chain/app" "github.com/sei-protocol/sei-chain/app/genesis" @@ -151,7 +149,6 @@ func TestInitModeConfiguration(t *testing.T) { // Build custom template with all sections customAppTemplate := srvconfig.ManualConfigTemplate + seidbconfig.StateCommitConfigTemplate + seidbconfig.StateStoreConfigTemplate + - seidbconfig.ReceiptStoreConfigTemplate + srvconfig.AutoManagedConfigTemplate // Simplified - just need the pruning config srvconfig.SetConfigTemplate(customAppTemplate) @@ -170,40 +167,10 @@ func TestInitModeConfiguration(t *testing.T) { if tt.validateApp != nil { tt.validateApp(t, configDir) } - - v := viper.New() - v.SetConfigFile(appTomlPath) - err = v.ReadInConfig() - require.NoError(t, err) - require.Equal(t, "pebbledb", v.GetString("receipt-store.rs-backend")) - require.Equal(t, "", v.GetString("receipt-store.db-directory")) - require.Equal(t, seidbconfig.DefaultReceiptStoreConfig().AsyncWriteBuffer, v.GetInt("receipt-store.async-write-buffer")) - require.Equal(t, seidbconfig.DefaultReceiptStoreConfig().KeepRecent, v.GetInt("receipt-store.keep-recent")) - require.Equal(t, seidbconfig.DefaultReceiptStoreConfig().PruneIntervalSeconds, v.GetInt("receipt-store.prune-interval-seconds")) }) } } -func TestInitAppConfigIncludesReceiptStoreDefaults(t *testing.T) { - customAppTemplate, customAppConfig := initAppConfig() - - tmpl, err := template.New("app").Parse(customAppTemplate) - require.NoError(t, err) - - var buf bytes.Buffer - err = tmpl.Execute(&buf, customAppConfig) - require.NoError(t, err) - - output := buf.String() - require.Contains(t, output, "[receipt-store]") - require.Contains(t, output, `rs-backend = "pebbledb"`) - require.Contains(t, output, `db-directory = ""`) - require.Contains(t, output, "async-write-buffer =") - require.Contains(t, output, "keep-recent =") - require.Contains(t, output, "prune-interval-seconds =") - require.NotContains(t, output, "use-default-comparer") -} - // TestInitModeFlag verifies the mode flag validation func TestInitModeFlag(t *testing.T) { validModes := []string{"validator", "full", "seed", "archive"} diff --git a/cmd/seid/cmd/root.go b/cmd/seid/cmd/root.go index 00625f9e11..235f98f0f1 100644 --- a/cmd/seid/cmd/root.go +++ b/cmd/seid/cmd/root.go @@ -430,7 +430,6 @@ func initAppConfig() (string, interface{}) { customAppTemplate := serverconfig.ManualConfigTemplate + seidbconfig.StateCommitConfigTemplate + seidbconfig.StateStoreConfigTemplate + - seidbconfig.ReceiptStoreConfigTemplate + evmrpcconfig.ConfigTemplate + gigaconfig.ConfigTemplate + serverconfig.AutoManagedConfigTemplate + ` diff --git a/sei-cosmos/server/config/config.go b/sei-cosmos/server/config/config.go index b3ac1b746c..7e9b2e383a 100644 --- a/sei-cosmos/server/config/config.go +++ b/sei-cosmos/server/config/config.go @@ -406,11 +406,10 @@ func GetConfig(v *viper.Viper) (Config, error) { SnapshotDirectory: v.GetString("state-sync.snapshot-directory"), }, StateCommit: config.StateCommitConfig{ - Enable: v.GetBool("state-commit.sc-enable"), - Directory: v.GetString("state-commit.sc-directory"), - WriteMode: config.WriteMode(v.GetString("state-commit.sc-write-mode")), - ReadMode: config.ReadMode(v.GetString("state-commit.sc-read-mode")), - EnableLatticeHash: v.GetBool("state-commit.sc-enable-lattice-hash"), + Enable: v.GetBool("state-commit.sc-enable"), + Directory: v.GetString("state-commit.sc-directory"), + WriteMode: config.WriteMode(v.GetString("state-commit.sc-write-mode")), + ReadMode: config.ReadMode(v.GetString("state-commit.sc-read-mode")), MemIAVLConfig: memiavl.Config{ AsyncCommitBuffer: v.GetInt("state-commit.sc-async-commit-buffer"), SnapshotKeepRecent: v.GetUint32("state-commit.sc-keep-recent"), diff --git a/sei-cosmos/storev2/rootmulti/store.go b/sei-cosmos/storev2/rootmulti/store.go index 10d378b630..b132b7723b 100644 --- a/sei-cosmos/storev2/rootmulti/store.go +++ b/sei-cosmos/storev2/rootmulti/store.go @@ -93,6 +93,9 @@ func NewStore( } ctx := context.Background() scStore := composite.NewCompositeCommitStore(ctx, scDir, logger, scConfig) + if err := scStore.CleanupCrashArtifacts(); err != nil { + panic(err) + } store := &Store{ logger: logger, scStore: scStore, diff --git a/sei-db/common/logger/logger.go b/sei-db/common/logger/logger.go index 6908199795..0e433a7af8 100644 --- a/sei-db/common/logger/logger.go +++ b/sei-db/common/logger/logger.go @@ -1,10 +1,5 @@ package logger -import ( - "fmt" - "strings" -) - // Logger is what any CometBFT library should take. type Logger interface { Debug(msg string, keyvals ...interface{}) @@ -14,8 +9,8 @@ type Logger interface { type nopLogger struct{} +// Interface assertions var _ Logger = (*nopLogger)(nil) -var _ Logger = (*consoleLogger)(nil) // NewNopLogger returns a logger that doesn't do anything. func NewNopLogger() Logger { return &nopLogger{} } @@ -23,40 +18,3 @@ func NewNopLogger() Logger { return &nopLogger{} } func (nopLogger) Info(string, ...interface{}) {} func (nopLogger) Debug(string, ...interface{}) {} func (nopLogger) Error(string, ...interface{}) {} - -type consoleLogger struct{} - -// NewConsoleLogger returns a logger that prints to stdout. -func NewConsoleLogger() Logger { return &consoleLogger{} } - -func (consoleLogger) Debug(msg string, keyvals ...interface{}) { - fmt.Println("[DEBUG]", msg, formatKeyvals(keyvals)) -} - -func (consoleLogger) Info(msg string, keyvals ...interface{}) { - fmt.Println("[INFO]", msg, formatKeyvals(keyvals)) -} - -func (consoleLogger) Error(msg string, keyvals ...interface{}) { - fmt.Println("[ERROR]", msg, formatKeyvals(keyvals)) -} - -func formatKeyvals(keyvals []interface{}) string { - if len(keyvals) == 0 { - return "" - } - var sb strings.Builder - for i := 0; i+1 < len(keyvals); i += 2 { - if sb.Len() > 0 { - sb.WriteByte(' ') - } - fmt.Fprintf(&sb, "%v=%v", keyvals[i], keyvals[i+1]) - } - if len(keyvals)%2 != 0 { - if sb.Len() > 0 { - sb.WriteByte(' ') - } - fmt.Fprintf(&sb, "%v=", keyvals[len(keyvals)-1]) - } - return sb.String() -} diff --git a/sei-db/config/receipt_config.go b/sei-db/config/receipt_config.go index 0d2887ed91..87b540251e 100644 --- a/sei-db/config/receipt_config.go +++ b/sei-db/config/receipt_config.go @@ -1,27 +1,5 @@ package config -import ( - "fmt" - "strings" - - "github.com/spf13/cast" -) - -// AppOptions is a minimal interface for reading config (e.g. from Viper). -// Implemented by sei-cosmos server/types.AppOptions; defined here to avoid import cycles. -type AppOptions interface { - Get(string) interface{} -} - -const ( - flagRSDBDirectory = "receipt-store.db-directory" - flagRSBackend = "receipt-store.rs-backend" - flagRSMisnamedBackend = "receipt-store.backend" - flagRSAsyncWriteBuffer = "receipt-store.async-write-buffer" - flagRSKeepRecent = "receipt-store.keep-recent" - flagRSPruneIntervalSeconds = "receipt-store.prune-interval-seconds" -) - // ReceiptStoreConfig defines configuration for the receipt store database. type ReceiptStoreConfig struct { // DBDirectory defines the directory to store the receipt store db files @@ -29,13 +7,12 @@ type ReceiptStoreConfig struct { // default to empty DBDirectory string `mapstructure:"db-directory"` - // Backend defines the backend database used for receipt-store. - // Supported backends: pebbledb (aka pebble), parquet + // Backend defines the backend database used for receipt-store + // Supported backends: pebbledb, rocksdb // defaults to pebbledb - Backend string `mapstructure:"rs-backend"` + Backend string `mapstructure:"backend"` // AsyncWriteBuffer defines the async queue length for commits to be applied to receipt store - // Applies only to the pebbledb backend. // Set <= 0 for synchronous writes. // defaults to 100 AsyncWriteBuffer int `mapstructure:"async-write-buffer"` @@ -48,6 +25,12 @@ type ReceiptStoreConfig struct { // PruneIntervalSeconds defines the interval in seconds to trigger pruning // default to every 600 seconds PruneIntervalSeconds int `mapstructure:"prune-interval-seconds"` + + // UseDefaultComparer uses Pebble's default lexicographic byte comparer instead of + // the custom MVCCComparer. This is NOT backwards compatible with existing databases + // that were created with MVCCComparer - only use this for NEW databases. + // defaults to false (use MVCCComparer for backwards compatibility) + UseDefaultComparer bool `mapstructure:"use-default-comparer"` } // DefaultReceiptStoreConfig returns the default ReceiptStoreConfig @@ -57,55 +40,6 @@ func DefaultReceiptStoreConfig() ReceiptStoreConfig { AsyncWriteBuffer: DefaultSSAsyncBuffer, KeepRecent: DefaultSSKeepRecent, PruneIntervalSeconds: DefaultSSPruneInterval, + UseDefaultComparer: false, } } - -// ReadReceiptConfig reads receipt store config from app options (e.g. TOML / Viper). -func ReadReceiptConfig(opts AppOptions) (ReceiptStoreConfig, error) { - cfg := DefaultReceiptStoreConfig() - if v := opts.Get(flagRSMisnamedBackend); v != nil { - return cfg, fmt.Errorf("unsupported receipt-store config key %q; use %q instead", flagRSMisnamedBackend, flagRSBackend) - } - if v := opts.Get(flagRSDBDirectory); v != nil { - dbDirectory, err := cast.ToStringE(v) - if err != nil { - return cfg, err - } - cfg.DBDirectory = strings.TrimSpace(dbDirectory) - } - if v := opts.Get(flagRSBackend); v != nil { - backend, err := cast.ToStringE(v) - if err != nil { - return cfg, err - } - backend = strings.ToLower(strings.TrimSpace(backend)) - switch backend { - case "pebbledb", "pebble", "parquet": - cfg.Backend = backend - default: - return cfg, fmt.Errorf("unsupported receipt-store backend %q; supported: pebbledb, parquet", backend) - } - } - if v := opts.Get(flagRSAsyncWriteBuffer); v != nil { - asyncWriteBuffer, err := cast.ToIntE(v) - if err != nil { - return cfg, err - } - cfg.AsyncWriteBuffer = asyncWriteBuffer - } - if v := opts.Get(flagRSKeepRecent); v != nil { - keepRecent, err := cast.ToIntE(v) - if err != nil { - return cfg, err - } - cfg.KeepRecent = keepRecent - } - if v := opts.Get(flagRSPruneIntervalSeconds); v != nil { - pruneIntervalSeconds, err := cast.ToIntE(v) - if err != nil { - return cfg, err - } - cfg.PruneIntervalSeconds = pruneIntervalSeconds - } - return cfg, nil -} diff --git a/sei-db/config/receipt_config_test.go b/sei-db/config/receipt_config_test.go deleted file mode 100644 index 68af52c524..0000000000 --- a/sei-db/config/receipt_config_test.go +++ /dev/null @@ -1,23 +0,0 @@ -package config - -import ( - "testing" - - "github.com/stretchr/testify/require" -) - -type mapAppOpts map[string]interface{} - -func (m mapAppOpts) Get(key string) interface{} { - return m[key] -} - -func TestReadReceiptConfigRejectsMisnamedBackendKey(t *testing.T) { - _, err := ReadReceiptConfig(mapAppOpts{ - "receipt-store.backend": "parquet", - }) - - require.Error(t, err) - require.ErrorContains(t, err, "receipt-store.backend") - require.ErrorContains(t, err, "receipt-store.rs-backend") -} diff --git a/sei-db/config/sc_config.go b/sei-db/config/sc_config.go index 830cc7ab48..b32995a77d 100644 --- a/sei-db/config/sc_config.go +++ b/sei-db/config/sc_config.go @@ -32,15 +32,12 @@ type StateCommitConfig struct { // WriteMode defines the write routing mode for EVM data // Valid values: cosmos_only, dual_write, split_write, evm_only // defaults to cosmos_only - WriteMode WriteMode `mapstructure:"write-mode"` + WriteMode WriteMode `mapstructure:"write_mode"` // ReadMode defines the read routing mode for EVM data // Valid values: cosmos_only, evm_first, split_read // defaults to cosmos_only - ReadMode ReadMode `mapstructure:"read-mode"` - - // EnableLatticeHash controls whether lattice hash will be participating in final app hash or not - EnableLatticeHash bool `mapstructure:"enable-lattice-hash"` + ReadMode ReadMode `mapstructure:"read_mode"` // MemIAVLConfig is the configuration for the MemIAVL (Cosmos) backend MemIAVLConfig memiavl.Config @@ -62,12 +59,11 @@ type StateCommitConfig struct { // DefaultStateCommitConfig returns the default StateCommitConfig func DefaultStateCommitConfig() StateCommitConfig { return StateCommitConfig{ - Enable: true, - WriteMode: CosmosOnlyWrite, - ReadMode: CosmosOnlyRead, - EnableLatticeHash: false, - MemIAVLConfig: memiavl.DefaultConfig(), - FlatKVConfig: flatkv.DefaultConfig(), + Enable: true, + WriteMode: CosmosOnlyWrite, + ReadMode: CosmosOnlyRead, + MemIAVLConfig: memiavl.DefaultConfig(), + FlatKVConfig: flatkv.DefaultConfig(), HistoricalProofMaxInFlight: DefaultSCHistoricalProofMaxInFlight, HistoricalProofRateLimit: DefaultSCHistoricalProofRateLimit, @@ -78,10 +74,10 @@ func DefaultStateCommitConfig() StateCommitConfig { // Validate checks if the StateCommitConfig is valid func (c StateCommitConfig) Validate() error { if !c.WriteMode.IsValid() { - return fmt.Errorf("invalid write-mode: %s", c.WriteMode) + return fmt.Errorf("invalid write_mode: %s", c.WriteMode) } if !c.ReadMode.IsValid() { - return fmt.Errorf("invalid read-mode: %s", c.ReadMode) + return fmt.Errorf("invalid read_mode: %s", c.ReadMode) } return nil } diff --git a/sei-db/config/toml.go b/sei-db/config/toml.go index 723832d0d6..8d972cf97f 100644 --- a/sei-db/config/toml.go +++ b/sei-db/config/toml.go @@ -23,10 +23,6 @@ sc-write-mode = "{{ .StateCommit.WriteMode }}" # defaults to cosmos_only sc-read-mode = "{{ .StateCommit.ReadMode }}" -# EnableLatticeHash controls whether the FlatKV lattice hash participates -# in the final app hash. Default: false. -sc-enable-lattice-hash = {{ .StateCommit.EnableLatticeHash }} - # Max concurrent historical proof queries (RPC /store path) sc-historical-proof-max-inflight = {{ .StateCommit.HistoricalProofMaxInFlight }} @@ -139,24 +135,6 @@ const ReceiptStoreConfigTemplate = ` # Supported backends: pebble (aka pebbledb), parquet # defaults to pebbledb rs-backend = "{{ .ReceiptStore.Backend }}" - -# Defines the receipt store directory. If unset, defaults to /data/receipt.db -db-directory = "{{ .ReceiptStore.DBDirectory }}" - -# AsyncWriteBuffer defines the async queue length for commits to be applied to receipt store. -# Applies only when rs-backend = "pebbledb"; parquet ignores this setting. -# Set <= 0 for synchronous writes. -# defaults to 100 -async-write-buffer = {{ .ReceiptStore.AsyncWriteBuffer }} - -# KeepRecent defines the number of versions to keep in receipt store -# Setting it to 0 means keep everything. -# Default to keep the last 100,000 blocks -keep-recent = {{ .ReceiptStore.KeepRecent }} - -# PruneIntervalSeconds defines the interval in seconds to trigger pruning. -# defaults to 600 seconds -prune-interval-seconds = {{ .ReceiptStore.PruneIntervalSeconds }} ` // DefaultConfigTemplate combines both templates for backward compatibility diff --git a/sei-db/config/toml_test.go b/sei-db/config/toml_test.go index 09a175850c..d35307766f 100644 --- a/sei-db/config/toml_test.go +++ b/sei-db/config/toml_test.go @@ -89,36 +89,6 @@ func TestStateStoreConfigTemplate(t *testing.T) { require.Contains(t, output, "ss-import-num-workers =", "Missing ss-import-num-workers") } -// TestReceiptStoreConfigTemplate verifies that all field paths in the receipt-store TOML template -// are valid and can be resolved against the actual config struct. -func TestReceiptStoreConfigTemplate(t *testing.T) { - type TemplateConfig struct { - ReceiptStore ReceiptStoreConfig - } - - cfg := TemplateConfig{ - ReceiptStore: DefaultReceiptStoreConfig(), - } - - tmpl, err := template.New("receipt").Parse(ReceiptStoreConfigTemplate) - require.NoError(t, err, "Failed to parse ReceiptStoreConfigTemplate") - - var buf bytes.Buffer - err = tmpl.Execute(&buf, cfg) - require.NoError(t, err, "Failed to execute ReceiptStoreConfigTemplate - field path mismatch detected") - - output := buf.String() - - require.Contains(t, output, "[receipt-store]", "Missing receipt-store section") - require.Contains(t, output, `rs-backend = "pebbledb"`, "Missing or incorrect rs-backend") - require.Contains(t, output, `db-directory = ""`, "Missing or incorrect db-directory") - require.Contains(t, output, "async-write-buffer =", "Missing async-write-buffer") - require.Contains(t, output, "keep-recent =", "Missing keep-recent") - require.Contains(t, output, "prune-interval-seconds =", "Missing prune-interval-seconds") - require.Contains(t, output, `Applies only when rs-backend = "pebbledb"`, "Missing pebble-only async-write-buffer note") - require.NotContains(t, output, "use-default-comparer", "use-default-comparer should not be in receipt-store template") -} - // TestDefaultConfigTemplate verifies the combined template works correctly func TestDefaultConfigTemplate(t *testing.T) { type TemplateConfig struct { @@ -146,9 +116,6 @@ func TestDefaultConfigTemplate(t *testing.T) { require.Contains(t, output, "[state-commit]") require.Contains(t, output, "[state-store]") require.Contains(t, output, "[receipt-store]") - require.Contains(t, output, "async-write-buffer =") - require.Contains(t, output, "keep-recent =") - require.Contains(t, output, "prune-interval-seconds =") } // TestWriteModeValues verifies WriteMode enum values match template output diff --git a/sei-db/ledger_db/parquet/store_config_test.go b/sei-db/ledger_db/parquet/store_config_test.go index b7d1d8b07f..2fba4afb66 100644 --- a/sei-db/ledger_db/parquet/store_config_test.go +++ b/sei-db/ledger_db/parquet/store_config_test.go @@ -65,19 +65,6 @@ func TestNewStoreUsesDefaultIntervalsWhenUnset(t *testing.T) { require.Equal(t, defaultMaxBlocksPerFile, store.CacheRotateInterval()) } -func TestNewStorePreservesKeepRecentAndPruneIntervalSettings(t *testing.T) { - store, err := NewStore(dbLogger.NewNopLogger(), StoreConfig{ - DBDirectory: t.TempDir(), - KeepRecent: 123, - PruneIntervalSeconds: 9, - }) - require.NoError(t, err) - t.Cleanup(func() { _ = store.Close() }) - - require.Equal(t, int64(123), store.config.KeepRecent) - require.Equal(t, int64(9), store.config.PruneIntervalSeconds) -} - func TestPruneOldFilesKeepsTrackingOnDeleteFailure(t *testing.T) { store, err := NewStore(dbLogger.NewNopLogger(), StoreConfig{ DBDirectory: t.TempDir(), diff --git a/sei-db/ledger_db/receipt/parquet_store_test.go b/sei-db/ledger_db/receipt/parquet_store_test.go index 6ba6142f4c..78e3f8a669 100644 --- a/sei-db/ledger_db/receipt/parquet_store_test.go +++ b/sei-db/ledger_db/receipt/parquet_store_test.go @@ -183,38 +183,12 @@ func TestParquetReceiptStoreWALReplay(t *testing.T) { require.Equal(t, receipt.TxHashHex, got.TxHashHex) } -func TestParquetReceiptStoreUsesConfiguredDirectory(t *testing.T) { - ctx, storeKey := newTestContext() - cfg := dbconfig.DefaultReceiptStoreConfig() - cfg.Backend = "parquet" - cfg.DBDirectory = t.TempDir() - - store, err := NewReceiptStore(dbLogger.NewNopLogger(), cfg, storeKey) - require.NoError(t, err) - - txHash := common.HexToHash("0x31") - receipt := makeTestReceipt(txHash, 11, 0, common.HexToAddress("0x401"), nil) - require.NoError(t, store.SetReceipts(ctx.WithBlockHeight(11), []ReceiptRecord{ - {TxHash: txHash, Receipt: receipt}, - })) - require.NoError(t, store.Close()) - - receiptFiles, err := filepath.Glob(filepath.Join(cfg.DBDirectory, "receipts_*.parquet")) - require.NoError(t, err) - require.NotEmpty(t, receiptFiles, "receipt parquet files should be written under the configured db directory") - - logFiles, err := filepath.Glob(filepath.Join(cfg.DBDirectory, "logs_*.parquet")) - require.NoError(t, err) - require.NotEmpty(t, logFiles, "log parquet files should be written under the configured db directory") -} - func TestParquetFilePruning(t *testing.T) { ctx, storeKey := newTestContext() cfg := dbconfig.DefaultReceiptStoreConfig() cfg.Backend = "parquet" cfg.DBDirectory = t.TempDir() cfg.KeepRecent = 600 // Keep 600 blocks, so files with blocks < (latest - 600) get pruned - cfg.PruneIntervalSeconds = 1 store, err := NewReceiptStore(dbLogger.NewNopLogger(), cfg, storeKey) require.NoError(t, err) @@ -235,36 +209,20 @@ func TestParquetFilePruning(t *testing.T) { // Close to flush all files require.NoError(t, store.Close()) - // Verify we have enough receipt and log files for pruning to remove at least one pair. + // Verify we have at least 2 receipt and log files receiptFiles, err := filepath.Glob(filepath.Join(cfg.DBDirectory, "receipts_*.parquet")) require.NoError(t, err) - require.GreaterOrEqual(t, len(receiptFiles), 3, "should have at least 3 receipt files") + require.GreaterOrEqual(t, len(receiptFiles), 2, "should have at least 2 receipt files") logFiles, err := filepath.Glob(filepath.Join(cfg.DBDirectory, "logs_*.parquet")) require.NoError(t, err) - require.GreaterOrEqual(t, len(logFiles), 3, "should have at least 3 log files") - - initialReceiptFileCount := len(receiptFiles) - initialLogFileCount := len(logFiles) - - const ( - pruneTimeout = 5_000_000_000 - prunePollInterval = 100_000_000 - ) + require.GreaterOrEqual(t, len(logFiles), 2, "should have at least 2 log files") - // Reopen store - pruning runs immediately on startup and then periodically afterward. + // Reopen store - pruning will run in background store, err = NewReceiptStore(dbLogger.NewNopLogger(), cfg, storeKey) require.NoError(t, err) t.Cleanup(func() { _ = store.Close() }) - require.Eventually(t, func() bool { - currentReceiptFiles, err := filepath.Glob(filepath.Join(cfg.DBDirectory, "receipts_*.parquet")) - require.NoError(t, err) - currentLogFiles, err := filepath.Glob(filepath.Join(cfg.DBDirectory, "logs_*.parquet")) - require.NoError(t, err) - return len(currentReceiptFiles) < initialReceiptFileCount && len(currentLogFiles) < initialLogFileCount - }, pruneTimeout, prunePollInterval, "configured keep-recent and prune-interval should trigger parquet file pruning") - // Verify we can still query recent data txHash := common.BigToHash(common.Big1.SetUint64(1400)) _, err = store.GetReceiptFromStore(ctx, txHash) diff --git a/sei-db/ledger_db/receipt/receipt_store.go b/sei-db/ledger_db/receipt/receipt_store.go index 250ac3ee30..8776a21a92 100644 --- a/sei-db/ledger_db/receipt/receipt_store.go +++ b/sei-db/ledger_db/receipt/receipt_store.go @@ -83,25 +83,6 @@ func NewReceiptStore(log dbLogger.Logger, config dbconfig.ReceiptStoreConfig, st return newCachedReceiptStore(backend), nil } -// BackendTypeName returns the backend implementation name ("parquet" or "pebble") for testing. -// Returns "" if store is nil or the backend type is unknown. -func BackendTypeName(store ReceiptStore) string { - if store == nil { - return "" - } - if c, ok := store.(*cachedReceiptStore); ok { - store = c.backend - } - switch store.(type) { - case *parquetReceiptStore: - return receiptBackendParquet - case *receiptStore: - return receiptBackendPebble - default: - return "unknown" - } -} - func newReceiptBackend(log dbLogger.Logger, config dbconfig.ReceiptStoreConfig, storeKey sdk.StoreKey) (ReceiptStore, error) { if log == nil { log = dbLogger.NewNopLogger() @@ -117,7 +98,6 @@ func newReceiptBackend(log dbLogger.Logger, config dbconfig.ReceiptStoreConfig, case receiptBackendPebble: ssConfig := dbconfig.DefaultStateStoreConfig() ssConfig.DBDirectory = config.DBDirectory - ssConfig.AsyncWriteBuffer = config.AsyncWriteBuffer ssConfig.KeepRecent = config.KeepRecent if config.PruneIntervalSeconds > 0 { ssConfig.PruneIntervalSeconds = config.PruneIntervalSeconds diff --git a/sei-db/state_db/bench/cryptosim/block.go b/sei-db/state_db/bench/cryptosim/block.go deleted file mode 100644 index a7833f957d..0000000000 --- a/sei-db/state_db/bench/cryptosim/block.go +++ /dev/null @@ -1,94 +0,0 @@ -package cryptosim - -import "iter" - -// A simulated block of transactions. -type block struct { - config *CryptoSimConfig - - // The transactions in the block. - transactions []*transaction - - // The block number. This is not currently preserved across benchmark restarts, but otherwise monotonically - // increases as you'd expect. - blockNumber int64 - - // The next account ID to be used when creating a new account, as of the end of this block. - nextAccountID int64 - - // The number of cold accounts, as of the end of this block. - numberOfColdAccounts int64 - - // The next ERC20 contract ID to be used when creating a new ERC20 contract, as of the end of this block. - nextErc20ContractID int64 - - metrics *CryptosimMetrics -} - -// Creates a new block with the given capacity. -func NewBlock( - config *CryptoSimConfig, - metrics *CryptosimMetrics, - blockNumber int64, - capacity int, -) *block { - return &block{ - config: config, - blockNumber: blockNumber, - transactions: make([]*transaction, 0, capacity), - metrics: metrics, - } -} - -// Returns an iterator over the transactions in the block. -func (b *block) Iterator() iter.Seq[*transaction] { - return func(yield func(*transaction) bool) { - for _, txn := range b.transactions { - if !yield(txn) { - return - } - } - } -} - -// Adds a transaction to the block. -func (b *block) AddTransaction(txn *transaction) { - b.transactions = append(b.transactions, txn) -} - -// Returns the block number. -func (b *block) BlockNumber() int64 { - return b.blockNumber -} - -// Sets information about account state as of the end of this block. -func (b *block) SetBlockAccountStats( - nextAccountID int64, - numberOfColdAccounts int64, - nextErc20ContractID int64, -) { - b.nextAccountID = nextAccountID - b.numberOfColdAccounts = numberOfColdAccounts - b.nextErc20ContractID = nextErc20ContractID -} - -// This method should be called after a block is finished executing and finalized. -// Reports metrics about the block. -func (b *block) ReportBlockMetrics() { - b.metrics.SetTotalNumberOfAccounts(b.nextAccountID, int64(b.config.NumberOfHotAccounts), b.numberOfColdAccounts) -} - -// Returns the next account ID to be used when creating a new account, as of the end of this block. -func (b *block) NextAccountID() int64 { - return b.nextAccountID -} - -// Returns the next ERC20 contract ID to be used when creating a new ERC20 contract, as of the end of this block. -func (b *block) NextErc20ContractID() int64 { - return b.nextErc20ContractID -} - -// Returns the number of transactions in the block. -func (b *block) TransactionCount() int64 { - return int64(len(b.transactions)) -} diff --git a/sei-db/state_db/bench/cryptosim/block_builder.go b/sei-db/state_db/bench/cryptosim/block_builder.go deleted file mode 100644 index eb36b5d281..0000000000 --- a/sei-db/state_db/bench/cryptosim/block_builder.go +++ /dev/null @@ -1,82 +0,0 @@ -package cryptosim - -import ( - "context" - "fmt" -) - -// A builder for blocks of transactions. -type blockBuilder struct { - ctx context.Context - - config *CryptoSimConfig - - // Metrics for the benchmark. - metrics *CryptosimMetrics - - // Produces random data. - dataGenerator *DataGenerator - - // Blocks are sent to this channel. - blocksChan chan *block - - // The next block number to be used. - nextBlockNumber int64 -} - -// Asyncronously produces blocks of transactions. -func NewBlockBuilder( - ctx context.Context, - config *CryptoSimConfig, - metrics *CryptosimMetrics, - dataGenerator *DataGenerator, -) *blockBuilder { - return &blockBuilder{ - ctx: ctx, - config: config, - metrics: metrics, - dataGenerator: dataGenerator, - blocksChan: make(chan *block, config.BlockChannelCapacity), - } -} - -// Starts the block builder. This should not be called until all other threads are done using the data generator, -// as the data generator is not thread-safe. -func (b *blockBuilder) Start() { - go b.mainLoop() -} - -// Builds blocks and sends them to the blocks channel. -func (b *blockBuilder) mainLoop() { - for { - block := b.buildBlock() - select { - case <-b.ctx.Done(): - return - case b.blocksChan <- block: - } - } -} - -func (b *blockBuilder) buildBlock() *block { - blk := NewBlock(b.config, b.metrics, b.nextBlockNumber, b.config.TransactionsPerBlock) - b.nextBlockNumber++ - - for i := 0; i < b.config.TransactionsPerBlock; i++ { - txn, err := BuildTransaction(b.dataGenerator) - if err != nil { - fmt.Printf("failed to build transaction: %v\n", err) - continue - } - blk.AddTransaction(txn) - } - - blk.SetBlockAccountStats( - b.dataGenerator.NextAccountID(), - b.dataGenerator.NumberOfColdAccounts(), - b.dataGenerator.NextErc20ContractID()) - - b.dataGenerator.ReportEndOfBlock() - - return blk -} diff --git a/sei-db/state_db/bench/cryptosim/config/basic-config.json b/sei-db/state_db/bench/cryptosim/config/basic-config.json index 53cdf98724..dea1b2a229 100644 --- a/sei-db/state_db/bench/cryptosim/config/basic-config.json +++ b/sei-db/state_db/bench/cryptosim/config/basic-config.json @@ -11,7 +11,7 @@ "Erc20ContractSize": 2048, "Erc20InteractionsPerAccount": 10, "Erc20StorageSlotSize": 32, - "ExecutorQueueSize": 1024, + "ExecutorQueueSize": 64, "HotAccountProbability": 0.1, "HotErc20ContractProbability": 0.5, "HotErc20ContractSetSize": 100, @@ -29,6 +29,5 @@ "TransactionsPerBlock": 1024, "MaxRuntimeSeconds": 0, "TransactionMetricsSampleRate": 0.001, - "BackgroundMetricsScrapeInterval": 60, - "BlockChannelCapacity": 8 + "BackgroundMetricsScrapeInterval": 60 } diff --git a/sei-db/state_db/bench/cryptosim/cryptosim.go b/sei-db/state_db/bench/cryptosim/cryptosim.go index 94fdd02d3a..04fc441b1a 100644 --- a/sei-db/state_db/bench/cryptosim/cryptosim.go +++ b/sei-db/state_db/bench/cryptosim/cryptosim.go @@ -48,9 +48,6 @@ type CryptoSim struct { // The data generator for the benchmark. dataGenerator *DataGenerator - // Builds blocks of transactions. - blockBuilder *blockBuilder - // The database for the benchmark. database *Database @@ -67,14 +64,6 @@ type CryptoSim struct { // benchmark, sending "false" will resume it. Suspending an already suspended benchmark will have no effect, // and resuming an already resumed benchmark will likewise have no effect. suspendChan chan bool - - // The most recent block that has been processed. - mostRecentBlock *block - - // The next ERC20 contract ID to be used when creating a new ERC20 contract. - // This is fixed after initial setup is complete, since we don't currently simulate - // the creation of new ERC20 contracts during the benchmark. - nextERC20ContractID int64 } // Creates a new cryptosim benchmark runner. @@ -153,8 +142,6 @@ func NewCryptoSim( ctx, cancel, database, dataGenerator.FeeCollectionAddress(), config.ExecutorQueueSize, metrics) } - blockBuilder := NewBlockBuilder(ctx, config, metrics, dataGenerator) - c := &CryptoSim{ ctx: ctx, cancel: cancel, @@ -164,7 +151,6 @@ func NewCryptoSim( lastConsoleUpdateTransactionCount: 0, closeChan: make(chan struct{}, 1), dataGenerator: dataGenerator, - blockBuilder: blockBuilder, database: database, executors: executors, metrics: metrics, @@ -180,10 +166,7 @@ func NewCryptoSim( c.database.ResetTransactionCount() c.startTimestamp = time.Now() - - // Now that we are done generating initial data, it is thread safe to start the block builder. - // (dataGenerator is not thread safe, and is used both for initial setup and for transaction generation) - c.blockBuilder.Start() + c.metrics.StartBackgroundSampling(c.startTimestamp) go c.run() return c, nil @@ -227,7 +210,7 @@ func (c *CryptoSim) setupAccounts() error { if err != nil { return fmt.Errorf("failed to create new account: %w", err) } - c.database.IncrementTransactionCount(1) + c.database.IncrementTransactionCount() finalized, err := c.database.MaybeFinalizeBlock( c.dataGenerator.NextAccountID(), c.dataGenerator.NextErc20ContractID()) if err != nil { @@ -235,7 +218,7 @@ func (c *CryptoSim) setupAccounts() error { } if finalized { c.dataGenerator.ReportAccountCounts() - c.dataGenerator.ReportEndOfBlock() + c.dataGenerator.ReportFinalizeBlock() } if c.dataGenerator.NextAccountID()%c.config.SetupUpdateIntervalCount == 0 { @@ -254,7 +237,7 @@ func (c *CryptoSim) setupAccounts() error { return fmt.Errorf("failed to finalize block: %w", err) } c.dataGenerator.ReportAccountCounts() - c.dataGenerator.ReportEndOfBlock() + c.dataGenerator.ReportFinalizeBlock() fmt.Printf("There are now %s accounts in the database.\n", int64Commas(c.dataGenerator.NextAccountID())) @@ -284,7 +267,7 @@ func (c *CryptoSim) setupErc20Contracts() error { break } - c.database.IncrementTransactionCount(1) + c.database.IncrementTransactionCount() _, _, err := c.dataGenerator.CreateNewErc20Contract(c.config.Erc20ContractSize, true) if err != nil { @@ -296,7 +279,7 @@ func (c *CryptoSim) setupErc20Contracts() error { return fmt.Errorf("failed to maybe commit batch: %w", err) } if finalized { - c.dataGenerator.ReportEndOfBlock() + c.dataGenerator.ReportFinalizeBlock() c.metrics.SetTotalNumberOfERC20Contracts(c.dataGenerator.NextErc20ContractID()) } @@ -318,14 +301,12 @@ func (c *CryptoSim) setupErc20Contracts() error { if err != nil { return fmt.Errorf("failed to finalize block: %w", err) } - c.dataGenerator.ReportEndOfBlock() + c.dataGenerator.ReportFinalizeBlock() c.metrics.SetTotalNumberOfERC20Contracts(c.dataGenerator.NextErc20ContractID()) fmt.Printf("There are now %s simulated ERC20 contracts in the database.\n", int64Commas(c.dataGenerator.NextErc20ContractID())) - c.nextERC20ContractID = c.dataGenerator.NextErc20ContractID() - return nil } @@ -335,14 +316,10 @@ func (c *CryptoSim) run() { defer c.teardown() haltTime := time.Now().Add(time.Duration(c.config.MaxRuntimeSeconds) * time.Second) - var timeoutChan <-chan time.Time - if c.config.MaxRuntimeSeconds > 0 { - timeoutChan = time.After(time.Until(haltTime)) - } - for { - c.metrics.SetMainThreadPhase("get_block") + c.metrics.SetMainThreadPhase("executing") + for { select { case <-c.ctx.Done(): if c.database.TransactionCount() > 0 { @@ -354,52 +331,55 @@ func (c *CryptoSim) run() { if isSuspended { c.suspend() } - case <-timeoutChan: - fmt.Printf("\nBenchmark timed out after %s.\n", formatDuration(time.Since(c.startTimestamp), 1)) - c.cancel() - return - case blk := <-c.blockBuilder.blocksChan: - c.handleNextBlock(blk) + default: + c.handleNextCycle(haltTime) } - - c.generateConsoleReport(false) } } -// Execute and finalize the next block. -func (c *CryptoSim) handleNextBlock(blk *block) { - c.mostRecentBlock = blk - c.metrics.SetMainThreadPhase("send_to_executors") - - c.database.IncrementTransactionCount(blk.TransactionCount()) - - for txn := range blk.Iterator() { - c.executors[c.nextExecutorIndex].ScheduleForExecution(txn) - c.nextExecutorIndex = (c.nextExecutorIndex + 1) % len(c.executors) +// Process the next benchmark cycle, creating a new transaction and executing it. +func (c *CryptoSim) handleNextCycle(haltTime time.Time) { + txn, err := BuildTransaction(c.dataGenerator) + if err != nil { + fmt.Printf("\nfailed to build transaction: %v\n", err) + c.cancel() + return } - if err := c.database.FinalizeBlock(blk.NextAccountID(), blk.NextErc20ContractID(), false); err != nil { - fmt.Printf("failed to finalize block: %v\n", err) + c.executors[c.nextExecutorIndex].ScheduleForExecution(txn) + c.nextExecutorIndex = (c.nextExecutorIndex + 1) % len(c.executors) + + finalized, err := c.database.MaybeFinalizeBlock( + c.dataGenerator.NextAccountID(), c.dataGenerator.NextErc20ContractID()) + if err != nil { + fmt.Printf("error finalizing block: %v\n", err) c.cancel() return } - blk.ReportBlockMetrics() + if finalized { + c.dataGenerator.ReportAccountCounts() + c.dataGenerator.ReportFinalizeBlock() + + if c.config.MaxRuntimeSeconds > 0 && time.Now().After(haltTime) { + c.cancel() + } + } + + c.database.IncrementTransactionCount() + c.generateConsoleReport(false) } // Suspends the benchmark. This method blocks until the benchmark is resumed or shut down. func (c *CryptoSim) suspend() { - if c.mostRecentBlock != nil { - err := c.database.FinalizeBlock(c.mostRecentBlock.nextAccountID, c.nextERC20ContractID, true) - if err != nil { - fmt.Printf("failed to finalize block: %v\n", err) - c.cancel() - return - } + err := c.database.FinalizeBlock(c.dataGenerator.NextAccountID(), c.dataGenerator.NextErc20ContractID(), true) + if err != nil { + fmt.Printf("failed to finalize block: %v\n", err) + c.cancel() + return } fmt.Printf("Benchmark suspended.\n") - c.metrics.SetMainThreadPhase("suspended") for { select { @@ -423,16 +403,9 @@ func (c *CryptoSim) suspend() { // Clean up the benchmark and release any resources. func (c *CryptoSim) teardown() { - if c.mostRecentBlock == nil { - err := c.database.CloseWithoutFinalizing() - if err != nil { - fmt.Printf("failed to close database: %v\n", err) - } - } else { - err := c.database.Close(c.mostRecentBlock.nextAccountID, c.nextERC20ContractID) - if err != nil { - fmt.Printf("failed to close database: %v\n", err) - } + err := c.database.Close(c.dataGenerator.NextAccountID(), c.dataGenerator.NextErc20ContractID()) + if err != nil { + fmt.Printf("failed to close database: %v\n", err) } c.dataGenerator.Close() diff --git a/sei-db/state_db/bench/cryptosim/cryptosim_config.go b/sei-db/state_db/bench/cryptosim/cryptosim_config.go index a4735a775b..a33e5a42ff 100644 --- a/sei-db/state_db/bench/cryptosim/cryptosim_config.go +++ b/sei-db/state_db/bench/cryptosim/cryptosim_config.go @@ -132,9 +132,6 @@ type CryptoSimConfig struct { // If true, pressing Enter in the terminal will toggle suspend/resume of the benchmark. // If false, Enter has no effect. EnableSuspension bool - - // The capacity of the channel that holds blocks awaiting execution. - BlockChannelCapacity int } // Returns the default configuration for the cryptosim benchmark. @@ -158,7 +155,7 @@ func DefaultCryptoSimConfig() *CryptoSimConfig { Erc20StorageSlotSize: 32, Erc20InteractionsPerAccount: 10, TransactionsPerBlock: 1024, - BlocksPerCommit: 1, + BlocksPerCommit: 32, Seed: 1337, CannedRandomSize: 1024 * 1024 * 1024, // 1GB Backend: wrappers.FlatKV, @@ -167,13 +164,12 @@ func DefaultCryptoSimConfig() *CryptoSimConfig { SetupUpdateIntervalCount: 100_000, ThreadsPerCore: 2.0, ConstantThreadCount: 0, - ExecutorQueueSize: 1024, + ExecutorQueueSize: 64, MaxRuntimeSeconds: 0, MetricsAddr: ":9090", TransactionMetricsSampleRate: 0.001, BackgroundMetricsScrapeInterval: 60, EnableSuspension: true, - BlockChannelCapacity: 8, } } @@ -246,9 +242,6 @@ func (c *CryptoSimConfig) Validate() error { if c.BackgroundMetricsScrapeInterval < 0 { return fmt.Errorf("BackgroundMetricsScrapeInterval must be non-negative (got %d)", c.BackgroundMetricsScrapeInterval) } - if c.BlockChannelCapacity < 1 { - return fmt.Errorf("BlockChannelCapacity must be at least 1 (got %d)", c.BlockChannelCapacity) - } return nil } diff --git a/sei-db/state_db/bench/cryptosim/cryptosim_metrics.go b/sei-db/state_db/bench/cryptosim/cryptosim_metrics.go index c5b16408b8..a25ac9ebc6 100644 --- a/sei-db/state_db/bench/cryptosim/cryptosim_metrics.go +++ b/sei-db/state_db/bench/cryptosim/cryptosim_metrics.go @@ -172,11 +172,20 @@ func NewCryptosimMetrics( m.startDataDirSizeSampling(dataDir, config.BackgroundMetricsScrapeInterval) } m.startProcessIOSampling(config.BackgroundMetricsScrapeInterval) - m.startUptimeSampling(time.Now()) } return m } +// StartBackgroundSampling starts goroutines that periodically update gauges +// (uptime, etc.). Call this when the benchmark is about to run, after initial +// state is loaded. Does not start any HTTP server; the caller configures export. +func (m *CryptosimMetrics) StartBackgroundSampling(startTime time.Time) { + if m == nil { + return + } + m.startUptimeSampling(startTime) +} + func (m *CryptosimMetrics) startUptimeSampling(startTime time.Time) { if m == nil || m.uptimeSeconds == nil { return diff --git a/sei-db/state_db/bench/cryptosim/data_generator.go b/sei-db/state_db/bench/cryptosim/data_generator.go index 8347d6bd2d..5920366814 100644 --- a/sei-db/state_db/bench/cryptosim/data_generator.go +++ b/sei-db/state_db/bench/cryptosim/data_generator.go @@ -138,6 +138,7 @@ func (d *DataGenerator) CreateNewAccount( accountID := d.nextAccountID d.nextAccountID++ + // Use EVMKeyCode for account data (balance+padding); EVMKeyNonce only accepts 8-byte values. addr := d.rand.Address(accountPrefix, accountID, AddressLen) address = evm.BuildMemIAVLEVMKey(evm.EVMKeyCode, addr) @@ -288,8 +289,7 @@ func (d *DataGenerator) FeeCollectionAddress() []byte { return d.feeCollectionAddress } -// Call this to signal that we have reached the end of a block. This is a signal that it is now safe to use -// recently created accounts as read/write targets. -func (d *DataGenerator) ReportEndOfBlock() { +// This method should be called after a block is finalized. +func (d *DataGenerator) ReportFinalizeBlock() { d.highestSafeAccountIDInBlock = d.nextAccountID - 1 } diff --git a/sei-db/state_db/bench/cryptosim/database.go b/sei-db/state_db/bench/cryptosim/database.go index 21d9a48937..2b723fbdbc 100644 --- a/sei-db/state_db/bench/cryptosim/database.go +++ b/sei-db/state_db/bench/cryptosim/database.go @@ -80,10 +80,10 @@ func (d *Database) Get(key []byte) ([]byte, bool, error) { return nil, false, nil } -// Signal that transactions have been added to the current block. -func (d *Database) IncrementTransactionCount(count int64) { - d.transactionCount += count - d.transactionsInCurrentBlock += count +// Signal that a transaction has been executed. +func (d *Database) IncrementTransactionCount() { + d.transactionCount++ + d.transactionsInCurrentBlock++ } // Reset the transaction count. Useful for when changing test phases. @@ -120,8 +120,6 @@ func (d *Database) FinalizeBlock( forceCommit bool, ) error { - d.metrics.SetMainThreadPhase("execute_block") - // Wait for all transactions in the current block to be executed. if d.flushFunc != nil { d.flushFunc() @@ -206,17 +204,6 @@ func (d *Database) Close(nextAccountID int64, nextErc20ContractID int64) error { return nil } -// Close the database and release any resources without finalizing the last batch. -func (d *Database) CloseWithoutFinalizing() error { - fmt.Printf("Closing database.\n") - err := d.db.Close() - if err != nil { - return fmt.Errorf("failed to close database: %w", err) - } - - return nil -} - // Set the function that flushes the executors. This setter is required to break a circular dependency. func (d *Database) SetFlushFunc(flushFunc func()) { d.flushFunc = flushFunc diff --git a/sei-db/state_db/bench/wrappers/db_implementations.go b/sei-db/state_db/bench/wrappers/db_implementations.go index a827918382..bd99044b46 100644 --- a/sei-db/state_db/bench/wrappers/db_implementations.go +++ b/sei-db/state_db/bench/wrappers/db_implementations.go @@ -31,10 +31,9 @@ const ( func newMemIAVLCommitStore(dbDir string) (DBWrapper, error) { cfg := memiavl.DefaultConfig() cfg.AsyncCommitBuffer = 10 - cfg.SnapshotInterval = 1000 - cfg.SnapshotMinTimeInterval = 60 + cfg.SnapshotInterval = 100 fmt.Printf("Opening memIAVL from directory %s\n", dbDir) - cs := memiavl.NewCommitStore(dbDir, logger.NewConsoleLogger(), cfg) + cs := memiavl.NewCommitStore(dbDir, logger.NewNopLogger(), cfg) cs.Initialize([]string{EVMStoreName}) _, err := cs.LoadVersion(0, false) if err != nil { @@ -50,8 +49,8 @@ func newFlatKVCommitStore(ctx context.Context, dbDir string) (DBWrapper, error) cfg := flatkv.DefaultConfig() cfg.Fsync = false fmt.Printf("Opening flatKV from directory %s\n", dbDir) - cs := flatkv.NewCommitStore(ctx, dbDir, logger.NewConsoleLogger(), cfg) - _, err := cs.LoadVersion(0) + cs := flatkv.NewCommitStore(ctx, dbDir, logger.NewNopLogger(), cfg) + _, err := cs.LoadVersion(0, false) if err != nil { if closeErr := cs.Close(); closeErr != nil { fmt.Printf("failed to close commit store during error recovery: %v\n", closeErr) @@ -67,7 +66,10 @@ func newCompositeCommitStore(ctx context.Context, dbDir string, writeMode config cfg.MemIAVLConfig.AsyncCommitBuffer = 10 cfg.MemIAVLConfig.SnapshotInterval = 100 - cs := composite.NewCompositeCommitStore(ctx, dbDir, logger.NewConsoleLogger(), cfg) + cs := composite.NewCompositeCommitStore(ctx, dbDir, logger.NewNopLogger(), cfg) + if err := cs.CleanupCrashArtifacts(); err != nil { + return nil, fmt.Errorf("failed to cleanup crash artifacts: %w", err) + } cs.Initialize([]string{EVMStoreName}) loaded, err := cs.LoadVersion(0, false) @@ -89,7 +91,7 @@ func openSSComposite(dir string) (*ssComposite.CompositeStateStore, error) { cfg.AsyncWriteBuffer = 0 cfg.WriteMode = config.DualWrite cfg.ReadMode = config.EVMFirstRead - return ssComposite.NewCompositeStateStore(cfg, dir, logger.NewConsoleLogger()) + return ssComposite.NewCompositeStateStore(cfg, dir, logger.NewNopLogger()) } func newSSCompositeStateStore(dbDir string) (DBWrapper, error) { diff --git a/sei-db/state_db/bench/wrappers/flatkv_wrapper.go b/sei-db/state_db/bench/wrappers/flatkv_wrapper.go index 748923569c..567936f593 100644 --- a/sei-db/state_db/bench/wrappers/flatkv_wrapper.go +++ b/sei-db/state_db/bench/wrappers/flatkv_wrapper.go @@ -41,7 +41,7 @@ func (f *flatKVWrapper) Commit() (int64, error) { } func (f *flatKVWrapper) LoadVersion(version int64) error { - _, err := f.base.LoadVersion(version) + _, err := f.base.LoadVersion(version, false) return err } diff --git a/sei-db/state_db/sc/composite/store.go b/sei-db/state_db/sc/composite/store.go index d0e8b5a4c6..aee2c29991 100644 --- a/sei-db/state_db/sc/composite/store.go +++ b/sei-db/state_db/sc/composite/store.go @@ -76,15 +76,27 @@ func (cs *CompositeCommitStore) Initialize(initialStores []string) { cs.cosmosCommitter.Initialize(initialStores) } +// CleanupCrashArtifacts removes temporary/orphaned files left by a +// previous process crash (e.g. FlatKV readonly-* working directories). +// Must be called once at process startup, before any read-only clones +// are created. Any writer lock acquired during cleanup is retained for +// the subsequent LoadVersion(..., false) call. +func (cs *CompositeCommitStore) CleanupCrashArtifacts() error { + if fkv, ok := cs.evmCommitter.(*flatkv.CommitStore); ok { + if err := fkv.CleanupOrphanedReadOnlyDirs(); err != nil { + return err + } + } + return nil +} + // SetInitialVersion sets the initial version for the store func (cs *CompositeCommitStore) SetInitialVersion(initialVersion int64) error { return cs.cosmosCommitter.SetInitialVersion(initialVersion) } -// LoadVersion loads the specified version of the database. -// Being used for two scenarios: -// ReadOnly: Either for state sync or for historical proof -// Writable: Opened during initialization for root multistore +// LoadVersion opens the database at the given version (0 = latest). +// When readOnly is true an isolated composite store is returned. func (cs *CompositeCommitStore) LoadVersion(targetVersion int64, readOnly bool) (types.Committer, error) { cosmosSC, err := cs.cosmosCommitter.LoadVersion(targetVersion, readOnly) if err != nil { @@ -96,7 +108,6 @@ func (cs *CompositeCommitStore) LoadVersion(targetVersion int64, readOnly bool) return nil, fmt.Errorf("unexpected committer type from cosmos LoadVersion") } - // Read only mode should return a new SC if readOnly { newStore := &CompositeCommitStore{ logger: cs.logger, @@ -104,16 +115,21 @@ func (cs *CompositeCommitStore) LoadVersion(targetVersion int64, readOnly bool) homeDir: cs.homeDir, config: cs.config, } - // TODO: Support loading FlatKV at target version for read only + if cs.evmCommitter != nil { + evmStore, err := cs.evmCommitter.LoadVersion(targetVersion, true) + if err != nil { + cs.logger.Info("FlatKV unavailable for readonly load, EVM data will not be served", + "version", targetVersion, "err", err) + } else { + newStore.evmCommitter = evmStore + } + } return newStore, nil } cs.cosmosCommitter = cosmosCommitter - // Load evmCommitter if initialized (nil when WriteMode is CosmosOnlyWrite). - // This is the single entry point for evmCommitter.LoadVersion — CMS calls - // CompositeCommitStore.LoadVersion(), which internally loads both backends. if cs.evmCommitter != nil { - _, err := cs.evmCommitter.LoadVersion(targetVersion) + _, err := cs.evmCommitter.LoadVersion(targetVersion, false) if err != nil { return nil, fmt.Errorf("failed to load FlatKV version: %w", err) } @@ -219,44 +235,16 @@ func (cs *CompositeCommitStore) GetEarliestVersion() (int64, error) { return cs.cosmosCommitter.GetEarliestVersion() } -// appendEvmLatticeHash returns a new CommitInfo with the EVM lattice hash -// appended, without mutating the original. Returns the original unchanged -// when lattice hashing is disabled. -func (cs *CompositeCommitStore) appendEvmLatticeHash(ci *proto.CommitInfo, evmHash []byte) *proto.CommitInfo { - if !cs.config.EnableLatticeHash { - return ci - } - combined := make([]proto.StoreInfo, len(ci.StoreInfos)+1) - copy(combined, ci.StoreInfos) - combined[len(combined)-1] = proto.StoreInfo{ - Name: "evm_lattice", - CommitId: proto.CommitID{ - Version: ci.Version, - Hash: evmHash, - }, - } - return &proto.CommitInfo{ - Version: ci.Version, - StoreInfos: combined, - } -} - // WorkingCommitInfo returns the working commit info func (cs *CompositeCommitStore) WorkingCommitInfo() *proto.CommitInfo { - ci := cs.cosmosCommitter.WorkingCommitInfo() - if cs.evmCommitter != nil { - return cs.appendEvmLatticeHash(ci, cs.evmCommitter.RootHash()) - } - return ci + // TODO: Need to combine hash for cosmos and evm + return cs.cosmosCommitter.WorkingCommitInfo() } // LastCommitInfo returns the last commit info func (cs *CompositeCommitStore) LastCommitInfo() *proto.CommitInfo { - ci := cs.cosmosCommitter.LastCommitInfo() - if cs.evmCommitter != nil { - return cs.appendEvmLatticeHash(ci, cs.evmCommitter.CommittedRootHash()) - } - return ci + // TODO: Need to combine hash for cosmos and evm + return cs.cosmosCommitter.LastCommitInfo() } // GetChildStoreByName returns the underlying child store by module name. diff --git a/sei-db/state_db/sc/composite/store_test.go b/sei-db/state_db/sc/composite/store_test.go index 551b505a05..395290a708 100644 --- a/sei-db/state_db/sc/composite/store_test.go +++ b/sei-db/state_db/sc/composite/store_test.go @@ -1,17 +1,43 @@ package composite import ( + "fmt" "testing" "github.com/stretchr/testify/require" - "github.com/sei-protocol/sei-chain/sei-db/common/evm" "github.com/sei-protocol/sei-chain/sei-db/common/logger" + "github.com/sei-protocol/sei-chain/sei-db/common/metrics" "github.com/sei-protocol/sei-chain/sei-db/config" "github.com/sei-protocol/sei-chain/sei-db/proto" + "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/flatkv" + "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/types" iavl "github.com/sei-protocol/sei-chain/sei-iavl" ) +// failingEVMStore is a mock flatkv.Store whose LoadVersion always fails. +type failingEVMStore struct{} + +var _ flatkv.Store = (*failingEVMStore)(nil) + +func (f *failingEVMStore) LoadVersion(int64, bool) (flatkv.Store, error) { + return nil, fmt.Errorf("flatkv unavailable") +} +func (f *failingEVMStore) ApplyChangeSets([]*proto.NamedChangeSet) error { return nil } +func (f *failingEVMStore) Commit() (int64, error) { return 0, nil } +func (f *failingEVMStore) Get([]byte) ([]byte, bool) { return nil, false } +func (f *failingEVMStore) Has([]byte) bool { return false } +func (f *failingEVMStore) Iterator(_, _ []byte) flatkv.Iterator { return nil } +func (f *failingEVMStore) IteratorByPrefix([]byte) flatkv.Iterator { return nil } +func (f *failingEVMStore) RootHash() []byte { return nil } +func (f *failingEVMStore) Version() int64 { return 0 } +func (f *failingEVMStore) WriteSnapshot(string) error { return nil } +func (f *failingEVMStore) Rollback(int64) error { return nil } +func (f *failingEVMStore) Exporter(int64) (types.Exporter, error) { return nil, nil } +func (f *failingEVMStore) Importer(int64) (types.Importer, error) { return nil, nil } +func (f *failingEVMStore) GetPhaseTimer() *metrics.PhaseTimer { return nil } +func (f *failingEVMStore) Close() error { return nil } + func TestCompositeStoreBasicOperations(t *testing.T) { dir := t.TempDir() cfg := config.DefaultStateCommitConfig() @@ -154,149 +180,6 @@ func TestWorkingAndLastCommitInfo(t *testing.T) { require.Equal(t, int64(1), lastInfo.Version) } -func TestLatticeHashCommitInfo(t *testing.T) { - addr := [20]byte{0xAA} - slot := [32]byte{0xBB} - evmStorageKey := evm.BuildMemIAVLEVMKey(evm.EVMKeyStorage, append(addr[:], slot[:]...)) - - makeChangesets := func(round byte) []*proto.NamedChangeSet { - return []*proto.NamedChangeSet{ - { - Name: "test", - Changeset: iavl.ChangeSet{ - Pairs: []*iavl.KVPair{ - {Key: []byte("key"), Value: []byte{round}}, - }, - }, - }, - { - Name: EVMStoreName, - Changeset: iavl.ChangeSet{ - Pairs: []*iavl.KVPair{ - {Key: evmStorageKey, Value: []byte{round}}, - }, - }, - }, - } - } - - tests := []struct { - name string - writeMode config.WriteMode - enableLattice bool - expectLattice bool - }{ - {"CosmosOnly/lattice_off", config.CosmosOnlyWrite, false, false}, - {"CosmosOnly/lattice_on", config.CosmosOnlyWrite, true, false}, - {"DualWrite/lattice_off", config.DualWrite, false, false}, - {"DualWrite/lattice_on", config.DualWrite, true, true}, - {"SplitWrite/lattice_off", config.SplitWrite, false, false}, - {"SplitWrite/lattice_on", config.SplitWrite, true, true}, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - dir := t.TempDir() - cfg := config.DefaultStateCommitConfig() - cfg.WriteMode = tt.writeMode - cfg.EnableLatticeHash = tt.enableLattice - - cs := NewCompositeCommitStore(t.Context(), dir, logger.NewNopLogger(), cfg) - cs.Initialize([]string{"test", EVMStoreName}) - _, err := cs.LoadVersion(0, false) - require.NoError(t, err) - defer cs.Close() - - var prevLastHash []byte - - for round := byte(1); round <= 3; round++ { - require.NoError(t, cs.ApplyChangeSets(makeChangesets(round))) - - // --- Working commit info --- - expectedCosmos := cs.cosmosCommitter.WorkingCommitInfo() - var expectedEvmHash []byte - if tt.expectLattice { - expectedEvmHash = cs.evmCommitter.RootHash() - } - - workingInfo := cs.WorkingCommitInfo() - cosmosCount := len(expectedCosmos.StoreInfos) - if tt.expectLattice { - require.Equal(t, cosmosCount+1, len(workingInfo.StoreInfos)) - } else { - require.Equal(t, cosmosCount, len(workingInfo.StoreInfos)) - } - for i, si := range expectedCosmos.StoreInfos { - require.Equal(t, si.Name, workingInfo.StoreInfos[i].Name) - require.Equal(t, si.CommitId.Hash, workingInfo.StoreInfos[i].CommitId.Hash) - } - if tt.expectLattice { - entry := workingInfo.StoreInfos[len(workingInfo.StoreInfos)-1] - require.Equal(t, "evm_lattice", entry.Name) - require.Equal(t, expectedEvmHash, entry.CommitId.Hash) - require.Equal(t, workingInfo.Version, entry.CommitId.Version) - - // Verify no duplicate names — important for app hash merkle tree - names := make(map[string]int) - for _, si := range workingInfo.StoreInfos { - names[si.Name]++ - } - for name, count := range names { - require.Equal(t, 1, count, "duplicate store name %q in WorkingCommitInfo", name) - } - } - - // --- Commit --- - _, err = cs.Commit() - require.NoError(t, err) - - // --- Last commit info --- - expectedCosmosLast := cs.cosmosCommitter.LastCommitInfo() - var expectedEvmCommitted []byte - if tt.expectLattice { - expectedEvmCommitted = cs.evmCommitter.CommittedRootHash() - require.Equal(t, expectedEvmHash, expectedEvmCommitted) - } - - lastInfo := cs.LastCommitInfo() - require.Equal(t, int64(round), lastInfo.Version) - cosmosLastCount := len(expectedCosmosLast.StoreInfos) - if tt.expectLattice { - require.Equal(t, cosmosLastCount+1, len(lastInfo.StoreInfos)) - } else { - require.Equal(t, cosmosLastCount, len(lastInfo.StoreInfos)) - } - for i, si := range expectedCosmosLast.StoreInfos { - require.Equal(t, si.Name, lastInfo.StoreInfos[i].Name) - require.Equal(t, si.CommitId.Hash, lastInfo.StoreInfos[i].CommitId.Hash) - } - if tt.expectLattice { - entry := lastInfo.StoreInfos[len(lastInfo.StoreInfos)-1] - require.Equal(t, "evm_lattice", entry.Name) - require.Equal(t, expectedEvmCommitted, entry.CommitId.Hash) - require.Equal(t, lastInfo.Version, entry.CommitId.Version) - - // Verify no duplicate names — important for app hash merkle tree - names := make(map[string]int) - for _, si := range lastInfo.StoreInfos { - names[si.Name]++ - } - for name, count := range names { - require.Equal(t, 1, count, "duplicate store name %q in LastCommitInfo", name) - } - - // Hash must change between rounds since data differs - if prevLastHash != nil { - require.NotEqual(t, prevLastHash, entry.CommitId.Hash, - "lattice hash should change across commits") - } - prevLastHash = entry.CommitId.Hash - } - } - }) - } -} - func TestRollback(t *testing.T) { dir := t.TempDir() cfg := config.DefaultStateCommitConfig() @@ -367,3 +250,46 @@ func TestGetVersions(t *testing.T) { require.NoError(t, err) require.Equal(t, int64(3), latestVersion) } + +func TestReadOnlyLoadVersionSoftFailsWhenFlatKVUnavailable(t *testing.T) { + dir := t.TempDir() + cfg := config.DefaultStateCommitConfig() + + cs := NewCompositeCommitStore(t.Context(), dir, logger.NewNopLogger(), cfg) + cs.Initialize([]string{"test"}) + + _, err := cs.LoadVersion(0, false) + require.NoError(t, err) + + err = cs.ApplyChangeSets([]*proto.NamedChangeSet{ + { + Name: "test", + Changeset: iavl.ChangeSet{ + Pairs: []*iavl.KVPair{ + {Key: []byte("key1"), Value: []byte("value1")}, + }, + }, + }, + }) + require.NoError(t, err) + _, err = cs.Commit() + require.NoError(t, err) + + // Inject a failing EVM committer to simulate FlatKV being unavailable + // for historical versions (different retention, late enablement, etc). + cs.evmCommitter = &failingEVMStore{} + + readOnly, err := cs.LoadVersion(0, true) + require.NoError(t, err, "readonly LoadVersion should succeed even when FlatKV fails") + defer func() { _ = readOnly.Close() }() + + compositeRO, ok := readOnly.(*CompositeCommitStore) + require.True(t, ok) + require.Nil(t, compositeRO.evmCommitter, "evmCommitter should be nil when FlatKV failed") + + // Cosmos data should still be accessible + store := compositeRO.GetChildStoreByName("test") + require.NotNil(t, store) + val := store.Get([]byte("key1")) + require.Equal(t, []byte("value1"), val) +} diff --git a/sei-db/state_db/sc/flatkv/api.go b/sei-db/state_db/sc/flatkv/api.go index 3bc76e9785..f764812667 100644 --- a/sei-db/state_db/sc/flatkv/api.go +++ b/sei-db/state_db/sc/flatkv/api.go @@ -21,10 +21,10 @@ type Options struct { // Read path: Get/Has/Iterator read committed state only. // Key format: x/evm memiavl keys (mapped internally to account/code/storage DBs). type Store interface { - // LoadVersion opens the database at the specified version. - // targetVersion == 0 opens the latest; targetVersion > 0 seeks the best - // snapshot <= target and replays WAL to reach it exactly. - LoadVersion(targetVersion int64) (Store, error) + // LoadVersion opens the database at the given version (0 = latest). + // When readOnly is true an isolated, read-only store is returned; + // the caller must Close it when done. + LoadVersion(targetVersion int64, readOnly bool) (Store, error) // ApplyChangeSets buffers EVM changesets (x/evm memiavl keys) and updates LtHash. // Non-EVM modules are ignored. Call Commit to persist. @@ -58,9 +58,6 @@ type Store interface { // raw LtHash vector. RootHash() []byte - // CommittedRootHash returns the 32-byte checksum of the last committed LtHash. - CommittedRootHash() []byte - // Version returns the latest committed version. Version() int64 diff --git a/sei-db/state_db/sc/flatkv/lthash_correctness_test.go b/sei-db/state_db/sc/flatkv/lthash_correctness_test.go index 597019e447..7c3c9f96fe 100644 --- a/sei-db/state_db/sc/flatkv/lthash_correctness_test.go +++ b/sei-db/state_db/sc/flatkv/lthash_correctness_test.go @@ -641,7 +641,7 @@ func TestLtHashPersistenceAfterReopen(t *testing.T) { // Phase 1: create state and close s1 := NewCommitStore(t.Context(), dir, nil, DefaultConfig()) - _, err := s1.LoadVersion(0) + _, err := s1.LoadVersion(0, false) require.NoError(t, err) for i := 1; i <= 10; i++ { @@ -659,7 +659,7 @@ func TestLtHashPersistenceAfterReopen(t *testing.T) { // Phase 2: reopen and verify s2 := NewCommitStore(t.Context(), dir, nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) defer s2.Close() diff --git a/sei-db/state_db/sc/flatkv/snapshot.go b/sei-db/state_db/sc/flatkv/snapshot.go index bfbcc76021..9c9ce87c51 100644 --- a/sei-db/state_db/sc/flatkv/snapshot.go +++ b/sei-db/state_db/sc/flatkv/snapshot.go @@ -423,6 +423,9 @@ func (s *CommitStore) migrateFlatLayout(flatkvDir string) (string, error) { // (e.g. flatkv/snapshot-00000000000000000100) and the current symlink is updated. // The dir parameter is ignored; snapshots are always stored alongside the live data. func (s *CommitStore) WriteSnapshot(_ string) error { + if s.readOnly { + return errReadOnly + } version := s.committedVersion if version <= 0 { return fmt.Errorf("cannot snapshot uncommitted store (version %d)", version) @@ -530,6 +533,9 @@ func (s *CommitStore) pruneSnapshots(dir string, currentVersion int64) { // completes, the next restart will simply re-run catchup against the // already-truncated WAL, converging to targetVersion. func (s *CommitStore) Rollback(targetVersion int64) error { + if s.readOnly { + return errReadOnly + } s.log.Info("FlatKV Rollback", "targetVersion", targetVersion) dir := s.flatkvDir() diff --git a/sei-db/state_db/sc/flatkv/snapshot_test.go b/sei-db/state_db/sc/flatkv/snapshot_test.go index d6b43e4c24..321dee24ac 100644 --- a/sei-db/state_db/sc/flatkv/snapshot_test.go +++ b/sei-db/state_db/sc/flatkv/snapshot_test.go @@ -33,7 +33,7 @@ func commitStorageEntry(t *testing.T, s *CommitStore, addr Address, slot Slot, v func TestSnapshotCreatesDir(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -60,7 +60,7 @@ func TestSnapshotCreatesDir(t *testing.T) { func TestSnapshotIdempotent(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -80,7 +80,7 @@ func TestOpenFromSnapshot(t *testing.T) { // Phase 1: create store, commit v1 and v2, snapshot at v2, commit v3 s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s1.LoadVersion(0) + _, err := s1.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s1, Address{0x10}, Slot{0x01}, []byte{0x01}) @@ -97,7 +97,7 @@ func TestOpenFromSnapshot(t *testing.T) { // Phase 2: reopen - should catchup from v2 snapshot + WAL entry for v3 s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) defer s2.Close() @@ -119,7 +119,7 @@ func TestCatchupUpdatesLtHash(t *testing.T) { dir := t.TempDir() s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s1.LoadVersion(0) + _, err := s1.LoadVersion(0, false) require.NoError(t, err) // Commit 5 versions, snapshot at v2 @@ -137,7 +137,7 @@ func TestCatchupUpdatesLtHash(t *testing.T) { // Reopen: catchup from v2 snapshot through v3,v4,v5 via WAL s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) defer s2.Close() @@ -151,7 +151,7 @@ func TestRollbackRewindsState(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) // Commit v1..v5, snapshot at v3 @@ -188,7 +188,7 @@ func TestRollbackToSnapshotExact(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s, Address{0x40}, Slot{0x01}, []byte{0x01}) @@ -209,7 +209,7 @@ func TestRollbackToSnapshotExact(t *testing.T) { func TestPartialSnapshotCleanup(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s, Address{0x50}, Slot{0x01}, []byte{0x01}) @@ -266,7 +266,7 @@ func TestMigrationFromFlatLayout(t *testing.T) { // Open the store - should trigger migration s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s.LoadVersion(0) + _, err = s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -297,7 +297,7 @@ func TestOpenVersionValidation(t *testing.T) { // Phase 1: create store, commit some data s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s1.LoadVersion(0) + _, err := s1.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s1, Address{0x60}, Slot{0x01}, []byte{0x11}) @@ -320,7 +320,7 @@ func TestOpenVersionValidation(t *testing.T) { // Phase 3: reopen - should detect skew and catchup s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) defer s2.Close() @@ -389,7 +389,7 @@ func TestLoadVersionWithTarget(t *testing.T) { dir := t.TempDir() s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s1.LoadVersion(0) + _, err := s1.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s1, Address{0x70}, Slot{0x01}, []byte{0x01}) @@ -402,7 +402,7 @@ func TestLoadVersionWithTarget(t *testing.T) { // Reopen at specific version 3 s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(3) + _, err = s2.LoadVersion(3, false) require.NoError(t, err) defer s2.Close() @@ -421,7 +421,7 @@ func TestSnapshotThenCatchupThenVerifyCorrectness(t *testing.T) { // Phase 1: build baseline at v2 and snapshot it. s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s1.LoadVersion(0) + _, err := s1.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s1, addr, slot, []byte{0x01}) // v1 @@ -442,7 +442,7 @@ func TestSnapshotThenCatchupThenVerifyCorrectness(t *testing.T) { // Phase 3: reopen exactly at v2. If later commits had mutated the snapshot // baseline in place, we'd incorrectly read 0x04 here. s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(2) + _, err = s2.LoadVersion(2, false) require.NoError(t, err) gotV2, ok := s2.Get(key) require.True(t, ok) @@ -451,7 +451,7 @@ func TestSnapshotThenCatchupThenVerifyCorrectness(t *testing.T) { // Phase 4: reopen latest again to ensure catchup/replay still reaches v4. s3 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s3.LoadVersion(0) + _, err = s3.LoadVersion(0, false) require.NoError(t, err) defer s3.Close() @@ -471,7 +471,7 @@ func TestLoadVersionMixedSequence(t *testing.T) { key := evm.BuildMemIAVLEVMKey(evm.EVMKeyStorage, StorageKey(addr, slot)) s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s, addr, slot, []byte{0x01}) @@ -486,7 +486,7 @@ func TestLoadVersionMixedSequence(t *testing.T) { // Round 1: load exactly v2 s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s1.LoadVersion(2) + _, err = s1.LoadVersion(2, false) require.NoError(t, err) require.Equal(t, int64(2), s1.Version()) require.Equal(t, hashAtV2, s1.RootHash()) @@ -497,7 +497,7 @@ func TestLoadVersionMixedSequence(t *testing.T) { // Round 2: load latest (catches up through v3, v4) s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) require.Equal(t, int64(4), s2.Version()) require.Equal(t, hashAtV4, s2.RootHash()) @@ -508,7 +508,7 @@ func TestLoadVersionMixedSequence(t *testing.T) { // Round 3: load v2 AGAIN — snapshot must still be clean. s3 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s3.LoadVersion(2) + _, err = s3.LoadVersion(2, false) require.NoError(t, err, "LoadVersion(2) must succeed after LoadVersion(0) dirtied working dir") require.Equal(t, int64(2), s3.Version()) require.Equal(t, hashAtV2, s3.RootHash()) @@ -524,7 +524,7 @@ func TestRollbackTargetBeforeWALStart(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) // Build: v1..v5, snapshot at v2 @@ -558,7 +558,7 @@ func TestRollbackTargetBeforeWALStart(t *testing.T) { // Simulate restart: should stay at v2. require.NoError(t, s.Close()) s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) defer s2.Close() @@ -756,7 +756,7 @@ func TestCreateWorkingDirReclones(t *testing.T) { func TestPruneSnapshotsKeepsRecent(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, Config{SnapshotKeepRecent: 1}) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) for i := 0; i < 5; i++ { @@ -780,7 +780,7 @@ func TestPruneSnapshotsKeepsRecent(t *testing.T) { func TestPruneSnapshotsKeepAll(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, Config{SnapshotKeepRecent: 100}) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -816,7 +816,7 @@ func TestOrphanSnapshotRecovery(t *testing.T) { require.True(t, os.IsNotExist(err), "no current symlink should exist") s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s.LoadVersion(0) + _, err = s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -878,7 +878,7 @@ func TestTraverseSnapshotsEarlyStop(t *testing.T) { func TestVerifyWALTailSuccess(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -892,7 +892,7 @@ func TestVerifyWALTailSuccess(t *testing.T) { func TestVerifyWALTailMismatch(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -914,7 +914,7 @@ func TestTryTruncateWAL(t *testing.T) { // the manual snapshot at v5 is created; this makes v5 the earliest // snapshot and gives tryTruncateWAL a positive truncation offset. s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, Config{SnapshotKeepRecent: 0}) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -939,7 +939,7 @@ func TestTryTruncateWAL(t *testing.T) { func TestTryTruncateWALNoSnapshot(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -960,7 +960,7 @@ func TestTryTruncateWALNoSnapshot(t *testing.T) { func TestRollbackRemovesPostTargetSnapshots(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) for i := 0; i < 3; i++ { @@ -1058,7 +1058,7 @@ func TestSeekSnapshotExact(t *testing.T) { func TestMultipleSnapshotsAndReopen(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, Config{SnapshotKeepRecent: 10}) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) var hashes [][]byte @@ -1072,7 +1072,7 @@ func TestMultipleSnapshotsAndReopen(t *testing.T) { for i, expectedHash := range hashes { ver := int64(i + 1) s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, Config{SnapshotKeepRecent: 10}) - _, err := s2.LoadVersion(ver) + _, err := s2.LoadVersion(ver, false) require.NoError(t, err) require.Equal(t, ver, s2.Version()) require.Equal(t, expectedHash, s2.RootHash(), "hash mismatch at version %d", ver) @@ -1087,7 +1087,7 @@ func TestMultipleSnapshotsAndReopen(t *testing.T) { func TestWriteSnapshotUpdatesSnapshotBase(t *testing.T) { dir := t.TempDir() s := NewCommitStore(context.Background(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s, Address{0xF0}, Slot{0x01}, []byte{0x01}) @@ -1113,7 +1113,7 @@ func TestWriteSnapshotUpdatesSnapshotBase(t *testing.T) { // so committedVersion should be 5 (from working dir metadata), not 2 // (from the snapshot). Catchup should replay 0 entries. s2 := NewCommitStore(context.Background(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) defer s2.Close() @@ -1124,7 +1124,7 @@ func TestWriteSnapshotUpdatesSnapshotBase(t *testing.T) { func TestSnapshotPreservesAllKeyTypes(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) addr := Address{0xAB} @@ -1145,7 +1145,7 @@ func TestSnapshotPreservesAllKeyTypes(t *testing.T) { require.NoError(t, s.Close()) s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) defer s2.Close() diff --git a/sei-db/state_db/sc/flatkv/store.go b/sei-db/state_db/sc/flatkv/store.go index a833e60f18..6410a44efa 100644 --- a/sei-db/state_db/sc/flatkv/store.go +++ b/sei-db/state_db/sc/flatkv/store.go @@ -2,6 +2,7 @@ package flatkv import ( "context" + "errors" "fmt" "io" "os" @@ -37,6 +38,8 @@ const ( tmpSuffix = "-tmp" removingSuffix = "-removing" + readOnlyDirPrefix = "readonly-" + // Metadata DB keys MetaGlobalVersion = "_meta/version" // Global committed version watermark (8 bytes) MetaGlobalLtHash = "_meta/hash" // Global LtHash (2048 bytes) @@ -98,8 +101,10 @@ type CommitStore struct { // File lock prevents multiple processes from opening the same DB. fileLock filelock.TryLockerSafe - // Used to track time spent in various phases of execution. phaseTimer *metrics.PhaseTimer + + readOnly bool // Set by readonly LoadVersion; guards all write methods. + readOnlyWorkDir string // Temp working dir for readonly store; removed by Close. } var _ Store = (*CommitStore)(nil) @@ -138,13 +143,20 @@ func (s *CommitStore) flatkvDir() string { return s.dbDir } -// LoadVersion loads the specified version of the database. -// -// - targetVersion == 0: open latest (follow current symlink + catchup to end of WAL). -// - targetVersion > 0: seek the best snapshot <= targetVersion, open it, then -// catchup via WAL to reach targetVersion exactly. -func (s *CommitStore) LoadVersion(targetVersion int64) (_ Store, retErr error) { - s.log.Info("FlatKV LoadVersion", "targetVersion", targetVersion) +var errReadOnly = errors.New("flatkv: store is read-only") + +// LoadVersion opens the database at the given version (0 = latest). +// When readOnly is true an isolated, read-only CommitStore is returned; +// the caller must Close it when done. +func (s *CommitStore) LoadVersion(targetVersion int64, readOnly bool) (_ Store, retErr error) { + s.log.Info("FlatKV LoadVersion", "targetVersion", targetVersion, "readOnly", readOnly) + + if readOnly { + if s.readOnly { + return nil, errReadOnly + } + return s.loadVersionReadOnly(targetVersion) + } _ = s.closeDBsOnly() @@ -196,6 +208,91 @@ func (s *CommitStore) LoadVersion(targetVersion int64) (_ Store, retErr error) { return s, nil } +// loadVersionReadOnly creates an isolated, read-only CommitStore at the +// requested version. +func (s *CommitStore) loadVersionReadOnly(targetVersion int64) (_ Store, retErr error) { + ro := NewCommitStore(s.ctx, s.dbDir, s.log, s.config) + + workDir, err := os.MkdirTemp(ro.flatkvDir(), readOnlyDirPrefix) + if err != nil { + return nil, fmt.Errorf("create readonly temp dir: %w", err) + } + ro.readOnlyWorkDir = workDir + + defer func() { + if retErr != nil { + if closeErr := ro.Close(); closeErr != nil { + s.log.Error("failed to close readonly store during error cleanup", "err", closeErr) + } + } + }() + + if err := ro.openReadOnly(targetVersion); err != nil { + return nil, fmt.Errorf("readonly open: %w", err) + } + + return ro, nil +} + +// openReadOnly opens PebbleDBs in readOnlyWorkDir, replays the WAL to +// targetVersion, then closes the WAL and marks the store as read-only. +// It never modifies the global "current" symlink. +func (s *CommitStore) openReadOnly(targetVersion int64) error { + s.clearPendingWrites() + + dir := s.flatkvDir() + + var snapDir string + if targetVersion > 0 { + baseVer, err := seekSnapshot(dir, targetVersion) + if err != nil { + return fmt.Errorf("seek snapshot for readonly: %w", err) + } + snapDir = filepath.Join(dir, snapshotName(baseVer)) + } else { + var err error + snapDir, _, err = currentSnapshotDir(dir) + if err != nil { + return fmt.Errorf("resolve current snapshot for readonly: %w", err) + } + } + + if err := createWorkingDir(snapDir, s.readOnlyWorkDir); err != nil { + return fmt.Errorf("create readonly working dir: %w", err) + } + + if err := s.openDBs(s.readOnlyWorkDir, dir); err != nil { + return err + } + + if err := s.loadGlobalMetadata(); err != nil { + return err + } + + if err := s.catchup(targetVersion); err != nil { + return fmt.Errorf("readonly catchup: %w", err) + } + + if targetVersion > 0 && s.committedVersion != targetVersion { + return fmt.Errorf("readonly version mismatch: requested %d, reached %d", + targetVersion, s.committedVersion) + } + + if s.changelog != nil { + closeErr := s.changelog.Close() + s.changelog = nil + if closeErr != nil { + return fmt.Errorf("close readonly changelog: %w", closeErr) + } + } + + s.readOnly = true + + s.log.Info("FlatKV readonly store opened", "version", s.committedVersion, + "dir", s.readOnlyWorkDir) + return nil +} + // openTo opens all DBs and catches up via WAL to the given version. // - 0 -> replay to end of WAL (latest). // - >0 -> replay up to (and including) that version. @@ -258,7 +355,7 @@ func (s *CommitStore) open() (retErr error) { return fmt.Errorf("create working dir: %w", err) } - if err := s.openAllDBs(workDir, dir); err != nil { + if err := s.openDBs(workDir, dir); err != nil { return err } @@ -290,20 +387,19 @@ func (s *CommitStore) acquireFileLock(dir string) error { return nil } -// openAllDBs opens the 5 PebbleDBs from the snapshot directory, the changelog -// WAL from the flatkv root, and loads per-DB local metadata. On failure all -// already-opened handles are closed. -func (s *CommitStore) openAllDBs(snapDir, flatkvRoot string) (retErr error) { +// openDBs opens all PebbleDBs from dbDir and optionally the changelog WAL +// from changelogRoot. On failure all already-opened handles are closed. +func (s *CommitStore) openDBs(dbDir, changelogRoot string) (retErr error) { type namedPath struct { name string path string } dbPaths := []namedPath{ - {accountDBDir, filepath.Join(snapDir, accountDBDir)}, - {codeDBDir, filepath.Join(snapDir, codeDBDir)}, - {storageDBDir, filepath.Join(snapDir, storageDBDir)}, - {legacyDBDir, filepath.Join(snapDir, legacyDBDir)}, - {metadataDir, filepath.Join(snapDir, metadataDir)}, + {accountDBDir, filepath.Join(dbDir, accountDBDir)}, + {codeDBDir, filepath.Join(dbDir, codeDBDir)}, + {storageDBDir, filepath.Join(dbDir, storageDBDir)}, + {legacyDBDir, filepath.Join(dbDir, legacyDBDir)}, + {metadataDir, filepath.Join(dbDir, metadataDir)}, } for _, np := range dbPaths { @@ -354,18 +450,19 @@ func (s *CommitStore) openAllDBs(snapDir, flatkvRoot string) (retErr error) { return err } - changelogPath := filepath.Join(flatkvRoot, changelogDir) - s.changelog, err = wal.NewChangelogWAL(s.log, changelogPath, wal.Config{ - WriteBufferSize: 0, - KeepRecent: 0, - PruneInterval: 0, - }) - if err != nil { - return fmt.Errorf("failed to open changelog: %w", err) + if changelogRoot != "" { + changelogPath := filepath.Join(changelogRoot, changelogDir) + s.changelog, err = wal.NewChangelogWAL(s.log, changelogPath, wal.Config{ + WriteBufferSize: 0, + KeepRecent: 0, + PruneInterval: 0, + }) + if err != nil { + return fmt.Errorf("failed to open changelog: %w", err) + } + toClose = append(toClose, s.changelog) } - toClose = append(toClose, s.changelog) - // Load per-DB local metadata (or initialize if not present) dataDBs := map[string]seidbtypes.KeyValueDB{ accountDBDir: s.accountDB, codeDBDir: s.codeDB, @@ -437,13 +534,10 @@ func (s *CommitStore) RootHash() []byte { return checksum[:] } -// CommittedRootHash returns the Blake3-256 digest of the last committed LtHash. -func (s *CommitStore) CommittedRootHash() []byte { - checksum := s.committedLtHash.Checksum() - return checksum[:] -} - func (s *CommitStore) Importer(version int64) (types.Importer, error) { + if s.readOnly { + return nil, errReadOnly + } // rootmulti.Restore closes the store before creating an importer. // Reopen the DBs so the importer can write data. if s.isClosed() { diff --git a/sei-db/state_db/sc/flatkv/store_lifecycle.go b/sei-db/state_db/sc/flatkv/store_lifecycle.go index 4c73f50136..95e93196bd 100644 --- a/sei-db/state_db/sc/flatkv/store_lifecycle.go +++ b/sei-db/state_db/sc/flatkv/store_lifecycle.go @@ -3,6 +3,9 @@ package flatkv import ( "errors" "fmt" + "os" + "path/filepath" + "strings" "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/types" ) @@ -22,8 +25,8 @@ func (s *CommitStore) closeDBsOnly() error { if err := s.changelog.Close(); err != nil { errs = append(errs, fmt.Errorf("changelog close: %w", err)) } - s.changelog = nil } + s.changelog = nil if s.metadataDB != nil { if err := s.metadataDB.Close(); err != nil { @@ -77,6 +80,10 @@ func (s *CommitStore) Close() error { s.fileLock = nil } + if s.readOnlyWorkDir != "" { + _ = os.RemoveAll(s.readOnlyWorkDir) + } + if err != nil { return err } @@ -85,6 +92,35 @@ func (s *CommitStore) Close() error { return nil } +// CleanupOrphanedReadOnlyDirs acquires the writer lock and removes readonly-* +// working directories left behind by a previous process crash. It is a +// startup-only API and must be called before any read-only instances are +// created in the current process. The acquired writer lock is retained for +// subsequent LoadVersion(..., false) calls. +func (s *CommitStore) CleanupOrphanedReadOnlyDirs() error { + dir := s.flatkvDir() + if err := os.MkdirAll(dir, 0o750); err != nil { + return fmt.Errorf("create flatkv dir: %w", err) + } + if s.fileLock == nil { + if err := s.acquireFileLock(dir); err != nil { + return err + } + } + + entries, err := os.ReadDir(dir) + if err != nil { + return nil + } + for _, e := range entries { + if e.IsDir() && strings.HasPrefix(e.Name(), readOnlyDirPrefix) { + s.log.Info("removing orphaned readonly dir", "dir", e.Name()) + _ = os.RemoveAll(filepath.Join(dir, e.Name())) + } + } + return nil +} + // Exporter creates an exporter for the given version. // NOTE: Not yet implemented. Will be added with state-sync support. // The future implementation will export each DB separately with internal key format. diff --git a/sei-db/state_db/sc/flatkv/store_test.go b/sei-db/state_db/sc/flatkv/store_test.go index dadba42c1b..3b51ceb15c 100644 --- a/sei-db/state_db/sc/flatkv/store_test.go +++ b/sei-db/state_db/sc/flatkv/store_test.go @@ -68,7 +68,7 @@ func setupTestStore(t *testing.T) *CommitStore { t.Helper() dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) return s } @@ -78,7 +78,7 @@ func setupTestStoreWithConfig(t *testing.T, cfg Config) *CommitStore { t.Helper() dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, cfg) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) return s } @@ -98,7 +98,7 @@ func commitAndCheck(t *testing.T, s *CommitStore) int64 { func TestStoreOpenClose(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) require.NoError(t, s.Close()) @@ -107,7 +107,7 @@ func TestStoreOpenClose(t *testing.T) { func TestStoreClose(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) // Close should succeed @@ -302,7 +302,7 @@ func TestStorePersistence(t *testing.T) { // Write and close s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s1.LoadVersion(0) + _, err := s1.LoadVersion(0, false) require.NoError(t, err) cs := makeChangeSet(key, value, false) @@ -312,7 +312,7 @@ func TestStorePersistence(t *testing.T) { // Reopen and verify s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) defer s2.Close() @@ -429,17 +429,17 @@ func TestFileLockPreventsDoubleOpen(t *testing.T) { dir := t.TempDir() s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s1.LoadVersion(0) + _, err := s1.LoadVersion(0, false) require.NoError(t, err) s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.Error(t, err, "second open on same dir should fail due to file lock") require.Contains(t, err.Error(), "file lock") require.NoError(t, s1.Close()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err, "should succeed after first store releases lock") require.NoError(t, s2.Close()) } @@ -451,7 +451,7 @@ func TestFileLockPreventsDoubleOpen(t *testing.T) { func TestClearChangelog(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -476,7 +476,7 @@ func TestClearChangelog(t *testing.T) { func TestCloseDBsOnlyIdempotent(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) require.NoError(t, s.closeDBsOnly()) @@ -494,7 +494,7 @@ func TestLoadVersionTargetBeyondWALFails(t *testing.T) { dir := t.TempDir() s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s1.LoadVersion(0) + _, err := s1.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s1, Address{0x01}, Slot{0x01}, []byte{0x01}) @@ -503,7 +503,7 @@ func TestLoadVersionTargetBeyondWALFails(t *testing.T) { require.NoError(t, s1.Close()) s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(100) + _, err = s2.LoadVersion(100, false) require.Error(t, err, "loading version beyond WAL should fail") } @@ -515,7 +515,7 @@ func TestReopenReusesWorkingDir(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s, Address{0x01}, Slot{0x01}, []byte{0x01}) @@ -528,7 +528,7 @@ func TestReopenReusesWorkingDir(t *testing.T) { require.NoError(t, err, "SNAPSHOT_BASE should exist after close") s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) defer s2.Close() @@ -542,7 +542,7 @@ func TestReopenReusesWorkingDir(t *testing.T) { func TestWalOffsetForVersionFastPath(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -564,7 +564,7 @@ func TestWalOffsetForVersionFastPath(t *testing.T) { func TestWalOffsetForVersionBeforeWAL(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -580,7 +580,7 @@ func TestWalOffsetForVersionBeforeWAL(t *testing.T) { func TestWalOffsetForVersionNotFound(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -598,7 +598,7 @@ func TestWalOffsetForVersionNotFound(t *testing.T) { func TestCatchupFromSpecificVersion(t *testing.T) { dir := t.TempDir() s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s1.LoadVersion(0) + _, err := s1.LoadVersion(0, false) require.NoError(t, err) for i := 0; i < 10; i++ { @@ -610,7 +610,7 @@ func TestCatchupFromSpecificVersion(t *testing.T) { require.NoError(t, s1.Close()) s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) defer s2.Close() @@ -659,7 +659,7 @@ func TestPersistenceAllKeyTypes(t *testing.T) { slot := Slot{0xBB} s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s1.LoadVersion(0) + _, err := s1.LoadVersion(0, false) require.NoError(t, err) storageKey := evm.BuildMemIAVLEVMKey(evm.EVMKeyStorage, StorageKey(addr, slot)) @@ -678,7 +678,7 @@ func TestPersistenceAllKeyTypes(t *testing.T) { require.NoError(t, s1.Close()) s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err = s2.LoadVersion(0) + _, err = s2.LoadVersion(0, false) require.NoError(t, err) defer s2.Close() @@ -697,3 +697,264 @@ func TestPersistenceAllKeyTypes(t *testing.T) { require.True(t, ok) require.Equal(t, []byte{0x60, 0x80}, v) } + +// ============================================================================= +// ReadOnly LoadVersion Tests +// ============================================================================= + +func TestReadOnlyBasicLoadAndRead(t *testing.T) { + dir := t.TempDir() + s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) + _, err := s.LoadVersion(0, false) + require.NoError(t, err) + + addr := Address{0xAA} + slot := Slot{0xBB} + key := memiavlStorageKey(addr, slot) + value := []byte{0xCC} + + require.NoError(t, s.ApplyChangeSets([]*proto.NamedChangeSet{makeChangeSet(key, value, false)})) + commitAndCheck(t, s) + + ro, err := s.LoadVersion(0, true) + require.NoError(t, err) + defer ro.Close() + + require.Equal(t, int64(1), ro.Version()) + got, found := ro.Get(key) + require.True(t, found) + require.Equal(t, value, got) + require.NotNil(t, ro.RootHash()) + require.Len(t, ro.RootHash(), 32) +} + +func TestReadOnlyLoadFromUnopenedStore(t *testing.T) { + dir := t.TempDir() + writer := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) + _, err := writer.LoadVersion(0, false) + require.NoError(t, err) + + addr := Address{0xCC} + slot := Slot{0xDD} + key := memiavlStorageKey(addr, slot) + value := []byte{0xEE} + + require.NoError(t, writer.ApplyChangeSets([]*proto.NamedChangeSet{makeChangeSet(key, value, false)})) + commitAndCheck(t, writer) + require.NoError(t, writer.Close()) + + fresh := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) + ro, err := fresh.LoadVersion(0, true) + require.NoError(t, err) + defer ro.Close() + + require.Equal(t, int64(1), ro.Version()) + got, found := ro.Get(key) + require.True(t, found) + require.Equal(t, value, got) +} + +func TestReadOnlyAtSpecificVersion(t *testing.T) { + dir := t.TempDir() + s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, Config{SnapshotInterval: 2, SnapshotKeepRecent: 10}) + _, err := s.LoadVersion(0, false) + require.NoError(t, err) + + addr := Address{0x11} + slot := Slot{0x22} + key := memiavlStorageKey(addr, slot) + + for i := byte(1); i <= 5; i++ { + require.NoError(t, s.ApplyChangeSets([]*proto.NamedChangeSet{ + makeChangeSet(key, []byte{i}, false), + })) + commitAndCheck(t, s) + } + + ro, err := s.LoadVersion(3, true) + require.NoError(t, err) + defer ro.Close() + + require.Equal(t, int64(3), ro.Version()) + got, found := ro.Get(key) + require.True(t, found) + require.Equal(t, []byte{3}, got) +} + +func TestReadOnlyWriteGuards(t *testing.T) { + dir := t.TempDir() + s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) + _, err := s.LoadVersion(0, false) + require.NoError(t, err) + + addr := Address{0xAA} + slot := Slot{0xBB} + key := memiavlStorageKey(addr, slot) + require.NoError(t, s.ApplyChangeSets([]*proto.NamedChangeSet{makeChangeSet(key, []byte{1}, false)})) + commitAndCheck(t, s) + + ro, err := s.LoadVersion(0, true) + require.NoError(t, err) + defer ro.Close() + + require.ErrorIs(t, ro.ApplyChangeSets(nil), errReadOnly) + _, err = ro.Commit() + require.ErrorIs(t, err, errReadOnly) + require.ErrorIs(t, ro.WriteSnapshot(""), errReadOnly) + require.ErrorIs(t, ro.Rollback(1), errReadOnly) + _, err = ro.(*CommitStore).Importer(1) + require.ErrorIs(t, err, errReadOnly) + + _, err = ro.LoadVersion(0, true) + require.ErrorIs(t, err, errReadOnly) +} + +func TestReadOnlyParentWritesDuringReadOnly(t *testing.T) { + dir := t.TempDir() + s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) + _, err := s.LoadVersion(0, false) + require.NoError(t, err) + + addr := Address{0xAA} + slot := Slot{0xBB} + key := memiavlStorageKey(addr, slot) + require.NoError(t, s.ApplyChangeSets([]*proto.NamedChangeSet{makeChangeSet(key, []byte{1}, false)})) + commitAndCheck(t, s) + + ro, err := s.LoadVersion(0, true) + require.NoError(t, err) + defer ro.Close() + + require.NoError(t, s.ApplyChangeSets([]*proto.NamedChangeSet{makeChangeSet(key, []byte{2}, false)})) + commitAndCheck(t, s) + require.NoError(t, s.ApplyChangeSets([]*proto.NamedChangeSet{makeChangeSet(key, []byte{3}, false)})) + commitAndCheck(t, s) + + require.Equal(t, int64(3), s.Version()) + + require.Equal(t, int64(1), ro.Version()) + got, found := ro.Get(key) + require.True(t, found) + require.Equal(t, []byte{1}, got) +} + +func TestReadOnlyConcurrentInstances(t *testing.T) { + dir := t.TempDir() + s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, Config{SnapshotInterval: 2, SnapshotKeepRecent: 10}) + _, err := s.LoadVersion(0, false) + require.NoError(t, err) + + addr := Address{0x11} + slot := Slot{0x22} + key := memiavlStorageKey(addr, slot) + + for i := byte(1); i <= 4; i++ { + require.NoError(t, s.ApplyChangeSets([]*proto.NamedChangeSet{ + makeChangeSet(key, []byte{i}, false), + })) + commitAndCheck(t, s) + } + + ro1, err := s.LoadVersion(0, true) + require.NoError(t, err) + defer ro1.Close() + + ro2, err := s.LoadVersion(0, true) + require.NoError(t, err) + defer ro2.Close() + + require.Equal(t, int64(4), ro1.Version()) + require.Equal(t, int64(4), ro2.Version()) + + g1, ok1 := ro1.Get(key) + g2, ok2 := ro2.Get(key) + require.True(t, ok1) + require.True(t, ok2) + require.Equal(t, []byte{4}, g1) + require.Equal(t, []byte{4}, g2) +} + +func TestReadOnlyFailureDoesNotAffectParent(t *testing.T) { + dir := t.TempDir() + s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) + _, err := s.LoadVersion(0, false) + require.NoError(t, err) + + addr := Address{0xAA} + slot := Slot{0xBB} + key := memiavlStorageKey(addr, slot) + require.NoError(t, s.ApplyChangeSets([]*proto.NamedChangeSet{makeChangeSet(key, []byte{1}, false)})) + commitAndCheck(t, s) + + _, err = s.LoadVersion(999, true) + require.Error(t, err) + + require.NoError(t, s.ApplyChangeSets([]*proto.NamedChangeSet{makeChangeSet(key, []byte{2}, false)})) + v, err := s.Commit() + require.NoError(t, err) + require.Equal(t, int64(2), v) + + got, found := s.Get(key) + require.True(t, found) + require.Equal(t, []byte{2}, got) +} + +func TestReadOnlyCloseRemovesTempDir(t *testing.T) { + dir := t.TempDir() + s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) + _, err := s.LoadVersion(0, false) + require.NoError(t, err) + + addr := Address{0xAA} + slot := Slot{0xBB} + key := memiavlStorageKey(addr, slot) + require.NoError(t, s.ApplyChangeSets([]*proto.NamedChangeSet{makeChangeSet(key, []byte{1}, false)})) + commitAndCheck(t, s) + + roStore, err := s.LoadVersion(0, true) + require.NoError(t, err) + + roCommit := roStore.(*CommitStore) + tmpDir := roCommit.readOnlyWorkDir + require.DirExists(t, tmpDir) + + require.NoError(t, roStore.Close()) + require.NoDirExists(t, tmpDir) +} + +func TestCleanupOrphanedReadOnlyDirs(t *testing.T) { + dir := t.TempDir() + s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) + defer func() { require.NoError(t, s.Close()) }() + + fkvDir := s.flatkvDir() + require.NoError(t, os.MkdirAll(fkvDir, 0o755)) + + // Simulate orphaned dirs left by a crash. + orphan1 := filepath.Join(fkvDir, "readonly-old1") + orphan2 := filepath.Join(fkvDir, "readonly-old2") + require.NoError(t, os.Mkdir(orphan1, 0o755)) + require.NoError(t, os.Mkdir(orphan2, 0o755)) + + require.NoError(t, s.CleanupOrphanedReadOnlyDirs()) + + require.NoDirExists(t, orphan1) + require.NoDirExists(t, orphan2) + + _, err := s.LoadVersion(0, false) + require.NoError(t, err) +} + +func TestCleanupOrphanedReadOnlyDirsHoldsWriterLock(t *testing.T) { + dir := t.TempDir() + s1 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) + defer func() { require.NoError(t, s1.Close()) }() + require.NoError(t, s1.CleanupOrphanedReadOnlyDirs()) + + s2 := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) + defer func() { require.NoError(t, s2.Close()) }() + + err := s2.CleanupOrphanedReadOnlyDirs() + require.Error(t, err) + require.ErrorContains(t, err, "acquire file lock") +} diff --git a/sei-db/state_db/sc/flatkv/store_write.go b/sei-db/state_db/sc/flatkv/store_write.go index cf1350f9be..7bf6a2d93b 100644 --- a/sei-db/state_db/sc/flatkv/store_write.go +++ b/sei-db/state_db/sc/flatkv/store_write.go @@ -20,9 +20,11 @@ import ( // - codeDB: key=addr, value=bytecode // - legacyDB: key=full original key (with prefix), value=raw value func (s *CommitStore) ApplyChangeSets(cs []*proto.NamedChangeSet) error { - s.phaseTimer.SetPhase("apply_change_sets_prepare") + if s.readOnly { + return errReadOnly + } - // Save original changesets for changelog + s.phaseTimer.SetPhase("apply_change_sets_prepare") s.pendingChangeSets = append(s.pendingChangeSets, cs...) // Collect LtHash pairs per DB (using internal key format) @@ -231,6 +233,9 @@ func (s *CommitStore) ApplyChangeSets(cs []*proto.NamedChangeSet) error { // Protocol: WAL → per-DB batch (with LocalMeta) → flush → update metaDB. // On crash, catchup replays WAL to recover incomplete commits. func (s *CommitStore) Commit() (int64, error) { + if s.readOnly { + return 0, errReadOnly + } // Auto-increment version version := s.committedVersion + 1 diff --git a/sei-db/state_db/sc/flatkv/store_write_test.go b/sei-db/state_db/sc/flatkv/store_write_test.go index 04915daf51..de8192dabb 100644 --- a/sei-db/state_db/sc/flatkv/store_write_test.go +++ b/sei-db/state_db/sc/flatkv/store_write_test.go @@ -524,7 +524,7 @@ func TestStoreFsyncConfig(t *testing.T) { t.Run("DefaultConfig", func(t *testing.T) { dir := t.TempDir() store := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := store.LoadVersion(0) + _, err := store.LoadVersion(0, false) require.NoError(t, err) defer store.Close() @@ -538,7 +538,7 @@ func TestStoreFsyncConfig(t *testing.T) { store := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, Config{ Fsync: false, }) - _, err := store.LoadVersion(0) + _, err := store.LoadVersion(0, false) require.NoError(t, err) defer store.Close() @@ -572,7 +572,7 @@ func TestAutoSnapshotTriggeredByInterval(t *testing.T) { SnapshotKeepRecent: 2, } s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, cfg) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -596,7 +596,7 @@ func TestAutoSnapshotNotTriggeredBeforeInterval(t *testing.T) { SnapshotKeepRecent: 2, } s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, cfg) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -623,7 +623,7 @@ func TestAutoSnapshotDisabledWhenIntervalZero(t *testing.T) { dir := t.TempDir() cfg := Config{SnapshotInterval: 0} s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, cfg) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -715,7 +715,7 @@ func TestLtHashDeterministicAcrossReopen(t *testing.T) { writeAndGetHash := func() []byte { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s, Address{0x01}, Slot{0x01}, []byte{0xAA}) @@ -839,7 +839,7 @@ func TestStoreFsyncEnabled(t *testing.T) { dir := t.TempDir() cfg := Config{Fsync: true} s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, cfg) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -860,7 +860,7 @@ func TestStoreFsyncEnabled(t *testing.T) { func TestLastSnapshotTimeUpdated(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) defer s.Close() @@ -880,7 +880,7 @@ func TestLastSnapshotTimeUpdated(t *testing.T) { func TestWALRecordsChangesets(t *testing.T) { dir := t.TempDir() s := NewCommitStore(t.Context(), filepath.Join(dir, flatkvRootDir), nil, DefaultConfig()) - _, err := s.LoadVersion(0) + _, err := s.LoadVersion(0, false) require.NoError(t, err) commitStorageEntry(t, s, Address{0x01}, Slot{0x01}, []byte{0xAA})