diff --git a/app/config_fuzz_test.go b/app/config_fuzz_test.go index 9d42328eee..37f717e8c1 100644 --- a/app/config_fuzz_test.go +++ b/app/config_fuzz_test.go @@ -167,14 +167,14 @@ var genesisKeys = []configtest.KeySpec{ // else, so the compiler keeps this list out of CheckRow and out of the discriminating-seed check, // where a row that predicted a resolved value would be wrong for all three. // -// The four flatkv names are here for a different reason than the three above them. Nothing in this +// The five flatkv names are here for a different reason than the three above them. Nothing in this // package reads them: sei-cosmos/server/config.GetConfig is their only reader, and its guardedKeys // target drives them. They are recorded on this section's record rather than on a second one in that // package so that [state-commit] has one list of operator-facing names, which is where someone // checking a spelling will look. // // What that record does and does not do is worth being exact about, in both directions. Renaming one -// of these four in GetConfig fails that package's own targets, not this record, because nothing +// of these five in GetConfig fails that package's own targets, not this record, because nothing // compares this list against the read site. Verified by renaming // state-commit.flatkv.snapshot-interval in GetConfig, which reddens three tests in sei-cosmos and // none here. And deleting one of them from GetConfig leaves this record green while it names a key no @@ -192,6 +192,7 @@ var scKeysWithTargetsOfTheirOwn = []configtest.KeyName{ "state-commit.flatkv.async-write-buffer", "state-commit.flatkv.snapshot-interval", "state-commit.flatkv.snapshot-keep-recent", + "state-commit.flatkv.max-snapshot-lag-blocks", } // genesisKeysWithTargetsOfTheirOwn are the [genesis] names no row claims. diff --git a/app/testdata/state-commit.golden b/app/testdata/state-commit.golden index a92860ef0a..b7f555c219 100644 --- a/app/testdata/state-commit.golden +++ b/app/testdata/state-commit.golden @@ -15,6 +15,7 @@ FlatKVConfig.Fsync = bool(false) FlatKVConfig.AsyncWriteBuffer = int(0) FlatKVConfig.SnapshotInterval = uint32(10000) FlatKVConfig.SnapshotKeepRecent = uint32(1) +FlatKVConfig.MaxSnapshotLagBlocks = uint32(64) FlatKVConfig.ExternalPruning = bool(false) FlatKVConfig.EnablePebbleMetrics = bool(true) FlatKVConfig.EnableReadWriteMetrics = bool(false) @@ -28,7 +29,7 @@ FlatKVConfig.AccountStoreConfig.EstimatedOverheadPerEntry = uint64(256) FlatKVConfig.AccountStoreConfig.Name = string("account") FlatKVConfig.AccountStoreConfig.MetricsEnabled = bool(true) FlatKVConfig.AccountStoreConfig.MetricsScrapeIntervalSeconds = float64(10) -FlatKVConfig.AccountStoreConfig.MaxUnflushedVersions = uint64(4) +FlatKVConfig.AccountStoreConfig.MaxUnflushedVersions = uint64(128) FlatKVConfig.AccountStoreConfig.TargetBytesPerFlush = uint64(4194304) FlatKVConfig.AccountStoreConfig.ReservedPrefix = string("_meta/") FlatKVConfig.AccountStoreConfig.FlushSync = bool(false) @@ -42,7 +43,7 @@ FlatKVConfig.CodeStoreConfig.EstimatedOverheadPerEntry = uint64(256) FlatKVConfig.CodeStoreConfig.Name = string("code") FlatKVConfig.CodeStoreConfig.MetricsEnabled = bool(true) FlatKVConfig.CodeStoreConfig.MetricsScrapeIntervalSeconds = float64(10) -FlatKVConfig.CodeStoreConfig.MaxUnflushedVersions = uint64(4) +FlatKVConfig.CodeStoreConfig.MaxUnflushedVersions = uint64(128) FlatKVConfig.CodeStoreConfig.TargetBytesPerFlush = uint64(4194304) FlatKVConfig.CodeStoreConfig.ReservedPrefix = string("_meta/") FlatKVConfig.CodeStoreConfig.FlushSync = bool(false) @@ -56,7 +57,7 @@ FlatKVConfig.StorageStoreConfig.EstimatedOverheadPerEntry = uint64(256) FlatKVConfig.StorageStoreConfig.Name = string("storage") FlatKVConfig.StorageStoreConfig.MetricsEnabled = bool(true) FlatKVConfig.StorageStoreConfig.MetricsScrapeIntervalSeconds = float64(10) -FlatKVConfig.StorageStoreConfig.MaxUnflushedVersions = uint64(4) +FlatKVConfig.StorageStoreConfig.MaxUnflushedVersions = uint64(128) FlatKVConfig.StorageStoreConfig.TargetBytesPerFlush = uint64(4194304) FlatKVConfig.StorageStoreConfig.ReservedPrefix = string("_meta/") FlatKVConfig.StorageStoreConfig.FlushSync = bool(false) @@ -70,7 +71,7 @@ FlatKVConfig.MiscStoreConfig.EstimatedOverheadPerEntry = uint64(256) FlatKVConfig.MiscStoreConfig.Name = string("misc") FlatKVConfig.MiscStoreConfig.MetricsEnabled = bool(true) FlatKVConfig.MiscStoreConfig.MetricsScrapeIntervalSeconds = float64(10) -FlatKVConfig.MiscStoreConfig.MaxUnflushedVersions = uint64(4) +FlatKVConfig.MiscStoreConfig.MaxUnflushedVersions = uint64(128) FlatKVConfig.MiscStoreConfig.TargetBytesPerFlush = uint64(4194304) FlatKVConfig.MiscStoreConfig.ReservedPrefix = string("_meta/") FlatKVConfig.MiscStoreConfig.FlushSync = bool(false) diff --git a/app/testdata/state-commit.keys.golden b/app/testdata/state-commit.keys.golden index e77e8ba067..f84c7fccd9 100644 --- a/app/testdata/state-commit.keys.golden +++ b/app/testdata/state-commit.keys.golden @@ -23,3 +23,4 @@ "state-commit.flatkv.async-write-buffer" "state-commit.flatkv.snapshot-interval" "state-commit.flatkv.snapshot-keep-recent" +"state-commit.flatkv.max-snapshot-lag-blocks" diff --git a/sei-cosmos/server/config/config.go b/sei-cosmos/server/config/config.go index ee732b85c9..87e90b6136 100644 --- a/sei-cosmos/server/config/config.go +++ b/sei-cosmos/server/config/config.go @@ -489,6 +489,9 @@ func GetConfig(v *viper.Viper) (Config, error) { if v.IsSet("state-commit.flatkv.snapshot-keep-recent") { flatKVConfig.SnapshotKeepRecent = v.GetUint32("state-commit.flatkv.snapshot-keep-recent") } + if v.IsSet("state-commit.flatkv.max-snapshot-lag-blocks") { + flatKVConfig.MaxSnapshotLagBlocks = v.GetUint32("state-commit.flatkv.max-snapshot-lag-blocks") + } if v.IsSet("state-commit.flatkv.enable-read-write-metrics") { flatKVConfig.EnableReadWriteMetrics = v.GetBool("state-commit.flatkv.enable-read-write-metrics") } diff --git a/sei-cosmos/server/config/config_fuzz_test.go b/sei-cosmos/server/config/config_fuzz_test.go index 04ef74c8a4..74e0ffb24d 100644 --- a/sei-cosmos/server/config/config_fuzz_test.go +++ b/sei-cosmos/server/config/config_fuzz_test.go @@ -315,6 +315,10 @@ var guardedKeys = []guardedKey{ }, {Key: "state-commit.flatkv.snapshot-interval", Path: "StateCommit.FlatKVConfig.SnapshotInterval", Set: 777}, {Key: "state-commit.flatkv.snapshot-keep-recent", Path: "StateCommit.FlatKVConfig.SnapshotKeepRecent", Set: 6}, + { + Key: "state-commit.flatkv.max-snapshot-lag-blocks", Path: "StateCommit.FlatKVConfig.MaxSnapshotLagBlocks", + Set: 32, + }, { Key: "state-commit.flatkv.enable-read-write-metrics", Path: "StateCommit.FlatKVConfig.EnableReadWriteMetrics", Set: true, DefaultIsZero: true, diff --git a/sei-cosmos/server/config/testdata/server_config.golden b/sei-cosmos/server/config/testdata/server_config.golden index 0adf2cc51b..3b99cc7fe9 100644 --- a/sei-cosmos/server/config/testdata/server_config.golden +++ b/sei-cosmos/server/config/testdata/server_config.golden @@ -73,6 +73,7 @@ StateCommit.FlatKVConfig.Fsync = bool(false) StateCommit.FlatKVConfig.AsyncWriteBuffer = int(0) StateCommit.FlatKVConfig.SnapshotInterval = uint32(10000) StateCommit.FlatKVConfig.SnapshotKeepRecent = uint32(1) +StateCommit.FlatKVConfig.MaxSnapshotLagBlocks = uint32(64) StateCommit.FlatKVConfig.ExternalPruning = bool(false) StateCommit.FlatKVConfig.EnablePebbleMetrics = bool(true) StateCommit.FlatKVConfig.EnableReadWriteMetrics = bool(false) @@ -86,7 +87,7 @@ StateCommit.FlatKVConfig.AccountStoreConfig.EstimatedOverheadPerEntry = uint64(2 StateCommit.FlatKVConfig.AccountStoreConfig.Name = string("account") StateCommit.FlatKVConfig.AccountStoreConfig.MetricsEnabled = bool(true) StateCommit.FlatKVConfig.AccountStoreConfig.MetricsScrapeIntervalSeconds = float64(10) -StateCommit.FlatKVConfig.AccountStoreConfig.MaxUnflushedVersions = uint64(4) +StateCommit.FlatKVConfig.AccountStoreConfig.MaxUnflushedVersions = uint64(128) StateCommit.FlatKVConfig.AccountStoreConfig.TargetBytesPerFlush = uint64(4194304) StateCommit.FlatKVConfig.AccountStoreConfig.ReservedPrefix = string("_meta/") StateCommit.FlatKVConfig.AccountStoreConfig.FlushSync = bool(false) @@ -100,7 +101,7 @@ StateCommit.FlatKVConfig.CodeStoreConfig.EstimatedOverheadPerEntry = uint64(256) StateCommit.FlatKVConfig.CodeStoreConfig.Name = string("code") StateCommit.FlatKVConfig.CodeStoreConfig.MetricsEnabled = bool(true) StateCommit.FlatKVConfig.CodeStoreConfig.MetricsScrapeIntervalSeconds = float64(10) -StateCommit.FlatKVConfig.CodeStoreConfig.MaxUnflushedVersions = uint64(4) +StateCommit.FlatKVConfig.CodeStoreConfig.MaxUnflushedVersions = uint64(128) StateCommit.FlatKVConfig.CodeStoreConfig.TargetBytesPerFlush = uint64(4194304) StateCommit.FlatKVConfig.CodeStoreConfig.ReservedPrefix = string("_meta/") StateCommit.FlatKVConfig.CodeStoreConfig.FlushSync = bool(false) @@ -114,7 +115,7 @@ StateCommit.FlatKVConfig.StorageStoreConfig.EstimatedOverheadPerEntry = uint64(2 StateCommit.FlatKVConfig.StorageStoreConfig.Name = string("storage") StateCommit.FlatKVConfig.StorageStoreConfig.MetricsEnabled = bool(true) StateCommit.FlatKVConfig.StorageStoreConfig.MetricsScrapeIntervalSeconds = float64(10) -StateCommit.FlatKVConfig.StorageStoreConfig.MaxUnflushedVersions = uint64(4) +StateCommit.FlatKVConfig.StorageStoreConfig.MaxUnflushedVersions = uint64(128) StateCommit.FlatKVConfig.StorageStoreConfig.TargetBytesPerFlush = uint64(4194304) StateCommit.FlatKVConfig.StorageStoreConfig.ReservedPrefix = string("_meta/") StateCommit.FlatKVConfig.StorageStoreConfig.FlushSync = bool(false) @@ -128,7 +129,7 @@ StateCommit.FlatKVConfig.MiscStoreConfig.EstimatedOverheadPerEntry = uint64(256) StateCommit.FlatKVConfig.MiscStoreConfig.Name = string("misc") StateCommit.FlatKVConfig.MiscStoreConfig.MetricsEnabled = bool(true) StateCommit.FlatKVConfig.MiscStoreConfig.MetricsScrapeIntervalSeconds = float64(10) -StateCommit.FlatKVConfig.MiscStoreConfig.MaxUnflushedVersions = uint64(4) +StateCommit.FlatKVConfig.MiscStoreConfig.MaxUnflushedVersions = uint64(128) StateCommit.FlatKVConfig.MiscStoreConfig.TargetBytesPerFlush = uint64(4194304) StateCommit.FlatKVConfig.MiscStoreConfig.ReservedPrefix = string("_meta/") StateCommit.FlatKVConfig.MiscStoreConfig.FlushSync = bool(false) diff --git a/sei-cosmos/storev2/rootmulti/flatkv_helpers_test.go b/sei-cosmos/storev2/rootmulti/flatkv_helpers_test.go index 1a0d49f6c5..00a90fcf68 100644 --- a/sei-cosmos/storev2/rootmulti/flatkv_helpers_test.go +++ b/sei-cosmos/storev2/rootmulti/flatkv_helpers_test.go @@ -8,6 +8,7 @@ import ( "encoding/binary" "errors" "testing" + "time" "github.com/sei-protocol/sei-chain/sei-cosmos/store/types" errorutils "github.com/sei-protocol/sei-chain/sei-db/common/errors" @@ -19,9 +20,49 @@ import ( "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/flatkv/vtype" scmemiavl "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/memiavl" sctypes "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/types" + sscomposite "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/composite" "github.com/stretchr/testify/require" ) +// requireStateStoreCaughtUp waits for the state store to reach version. +// +// The state store applies a block asynchronously: ApplyChangesetAsync hands it to a background writer +// whose queue is AsyncWriteBuffer deep, so a commit returning says nothing about the state store having +// applied that block. Any test that asserts a state store version, or that rolls back — rollback +// refuses a state store sitting below the target — has to establish this first. +func requireStateStoreCaughtUp(t *testing.T, store *Store, version int64) { + t.Helper() + require.NotNil(t, store.ssStore) + require.Eventually(t, func() bool { + return store.ssStore.GetLatestVersion() >= version + }, 10*time.Second, 5*time.Millisecond, + "state store never caught up to version %d", version) +} + +// requireStateStoreSnapshotAtOrBelow waits for a state store snapshot at or below target to be on +// disk, which is what rolling the state store back to target requires as its base. +// +// State store snapshots are published asynchronously and their coordinator declines a boundary while +// another snapshot is still in flight, so committing past a boundary does not mean its snapshot exists. +// With a per-block interval only the first boundary is usually accepted, the rest reporting "in_flight". +func requireStateStoreSnapshotAtOrBelow(t *testing.T, homeDir string, target int64) { + t.Helper() + root := utils.GetStateStoreSnapshotsPath(homeDir) + require.Eventually(t, func() bool { + versions, err := sscomposite.ListSnapshotVersions(root) + if err != nil { + return false + } + for _, version := range versions { + if version > 0 && version <= target { + return true + } + } + return false + }, 10*time.Second, 5*time.Millisecond, + "no state store snapshot at or below version %d was ever published under %s", target, root) +} + // --------------------------------------------------------------------------- // Config helpers // --------------------------------------------------------------------------- diff --git a/sei-cosmos/storev2/rootmulti/flatkv_recovery_test.go b/sei-cosmos/storev2/rootmulti/flatkv_recovery_test.go index 35690a9aa4..22834d8d98 100644 --- a/sei-cosmos/storev2/rootmulti/flatkv_recovery_test.go +++ b/sei-cosmos/storev2/rootmulti/flatkv_recovery_test.go @@ -81,6 +81,8 @@ func TestRollbackToVersionRollsBackStateStore(t *testing.T) { for block := 1; block <= 5; block++ { simulateBlock(t, store, storeKeys, block, evmData) } + requireStateStoreCaughtUp(t, store, 5) + requireStateStoreSnapshotAtOrBelow(t, dir, 3) require.NoError(t, store.RollbackToVersion(3)) require.Equal(t, int64(3), store.LastCommitID().Version) @@ -110,6 +112,7 @@ func TestRollbackToVersionWithoutStateStoreSnapshots(t *testing.T) { for block := 1; block <= 5; block++ { simulateBlock(t, store, storeKeys, block, evmData) } + requireStateStoreCaughtUp(t, store, 5) require.NoError(t, store.RollbackToVersion(3)) require.Equal(t, int64(3), store.LastCommitID().Version) @@ -140,6 +143,8 @@ func TestRollbackToVersionProceedsWhenStateStoreCannotFollow(t *testing.T) { for block := 1; block <= 5; block++ { simulateBlock(t, store, storeKeys, block, evmData) } + requireStateStoreCaughtUp(t, store, 5) + rollbackable := store.ssStore store.ssStore = nonRollbackableStateStore{rollbackable} diff --git a/sei-db/controller/prunable_store.go b/sei-db/controller/prunable_store.go index 90bb225061..5c85bb8cce 100644 --- a/sei-db/controller/prunable_store.go +++ b/sei-db/controller/prunable_store.go @@ -20,8 +20,13 @@ type PrunableStore interface { // Only called when ExternalPruning reports true. PruneHistory(blockNumber uint64) error - // PruneSnapshots may drop every snapshot strictly below blockNumber. A store that keeps no - // snapshots returns nil. + // PruneSnapshots may drop every snapshot strictly below blockNumber, and may do so + // asynchronously. A store that keeps no snapshots returns nil. + // + // A store that defers the work may let a later call supersede a pending one, so a given + // blockNumber is not guaranteed to be acted on at all — only that retention converges on the + // most recent one. A nil return therefore means the request was accepted, not that the + // snapshots are gone. // // blockNumber is never 0, and never above the block this store last returned from // GetRollbackFloor. diff --git a/sei-db/db_engine/view/view_manager_config.go b/sei-db/db_engine/view/view_manager_config.go index 35660d7458..9c20d9bc96 100644 --- a/sei-db/db_engine/view/view_manager_config.go +++ b/sei-db/db_engine/view/view_manager_config.go @@ -74,7 +74,7 @@ func DefaultViewManagerConfig(name string, reservedPrefix string) *ViewManagerCo Name: name, MetricsEnabled: true, MetricsScrapeIntervalSeconds: 10, - MaxUnflushedVersions: 4, + MaxUnflushedVersions: 128, TargetBytesPerFlush: unit.MB * 4, ReservedPrefix: reservedPrefix, FlushSync: false, diff --git a/sei-db/state_db/sc/composite/store_auto_test.go b/sei-db/state_db/sc/composite/store_auto_test.go index 00870c556d..0474c4aaf3 100644 --- a/sei-db/state_db/sc/composite/store_auto_test.go +++ b/sei-db/state_db/sc/composite/store_auto_test.go @@ -383,6 +383,18 @@ func TestComposite_Auto_ExportImportRoundTrip(t *testing.T) { runBlocks(t, src, workload, 1) h := src.Version() + // FlatKV writes snapshots off the execution thread, so a commit returning no longer means its + // snapshot churn has finished. Exporting reads a snapshot by copying its directory, and pruning the + // oldest snapshot is the last step of publishing a new one, so without this wait the writer can + // delete the directory the export is part way through copying. + // + // Reached through the concrete store because quiescing the writer is not part of the flatkv.Store + // abstraction: no production caller needs it, and this test only does because it drives commits and + // reads from one goroutine and so has a quiet period to establish. + flatKVStore, ok := src.flatKV.(*flatkv.CommitStore) + require.True(t, ok) + require.NoError(t, flatKVStore.FlushSnapshots()) + exp, err := src.Exporter(h) require.NoError(t, err) items := drainCompositeExporter(t, exp) diff --git a/sei-db/state_db/sc/composite/store_test.go b/sei-db/state_db/sc/composite/store_test.go index 6a5a4733d4..28597f9d40 100644 --- a/sei-db/state_db/sc/composite/store_test.go +++ b/sei-db/state_db/sc/composite/store_test.go @@ -54,7 +54,6 @@ func (f *failingEVMStore) RootHash() ([]byte, int64) { retur func (f *failingEVMStore) Version() int64 { return 0 } func (f *failingEVMStore) PendingVersion() int64 { return 0 } func (f *failingEVMStore) GetLatestVersion() (int64, error) { return 0, nil } -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 } diff --git a/sei-db/state_db/sc/flatkv/api.go b/sei-db/state_db/sc/flatkv/api.go index 85448481db..796f6f2a40 100644 --- a/sei-db/state_db/sc/flatkv/api.go +++ b/sei-db/state_db/sc/flatkv/api.go @@ -152,9 +152,6 @@ type Store interface { // inspect the store's height without taking ownership of it. GetLatestVersion() (int64, error) - // WriteSnapshot writes a complete snapshot to dir. - WriteSnapshot(dir string) error - // Rollback rewinds a store opened with LoadLatest to targetVersion and prunes everything above it: // snapshots, WAL blocks and committed state. It is the only way to move a committable store backwards, // and the result keeps committing from targetVersion+1. An unreachable target is rejected before diff --git a/sei-db/state_db/sc/flatkv/config/config.go b/sei-db/state_db/sc/flatkv/config/config.go index b9da35a528..2aeacbffb9 100644 --- a/sei-db/state_db/sc/flatkv/config/config.go +++ b/sei-db/state_db/sc/flatkv/config/config.go @@ -8,11 +8,6 @@ import ( "github.com/sei-protocol/sei-chain/sei-db/db_engine/view" ) -const ( - DefaultSnapshotInterval uint32 = 10000 - DefaultSnapshotKeepRecent uint32 = 1 -) - // Config defines configuration for the FlatKV (EVM) commit store. type Config struct { // DataDir is the root directory for the FlatKV data files. @@ -22,27 +17,32 @@ type Config struct { // Fsync controls whether every view manager's flush is fsync'd. It overwrites each store // config's FlushSync, so the four databases are always synced alike. The state WAL is // unaffected and always writes NoSync. - // Default: false Fsync bool `mapstructure:"fsync"` // AsyncWriteBuffer defines the size of the async write buffer for data DBs. // Set <= 0 for synchronous writes. - // Default: 0 (synchronous) AsyncWriteBuffer int `mapstructure:"async-write-buffer"` // SnapshotInterval defines how often (in blocks) a PebbleDB checkpoint // snapshot is taken. 0 disables auto-snapshots. // Without periodic snapshots the WAL grows unbounded and every restart // replays the entire history from snapshot-0. - // Default: 10000 SnapshotInterval uint32 `mapstructure:"snapshot-interval"` // SnapshotKeepRecent defines how many old snapshots to keep besides the // latest one. 0 means keep only the current snapshot (no old snapshots). // Ignored entirely when ExternalPruning is set. - // Default: 1 SnapshotKeepRecent uint32 `mapstructure:"snapshot-keep-recent"` + // MaxSnapshotLagBlocks is how many committed blocks may queue up behind a snapshot that is still + // being written before Commit blocks. A value below 1 is treated as 1. + // + // A snapshot being written holds every database pinned at its own height, so no later block can + // reach disk until it completes, and each one is retained in memory meanwhile. This bounds how far + // that can run, trading a pause in block production for the memory the backlog would otherwise + // consume. It bounds blocks rather than bytes, so it mitigates exhaustion rather than preventing it. + MaxSnapshotLagBlocks uint32 `mapstructure:"max-snapshot-lag-blocks"` + // ExternalPruning hands retention to the StorageGarbageCollector: the store stops pruning its // own snapshots (SnapshotKeepRecent) and stops truncating the state WAL. // @@ -51,16 +51,12 @@ type Config struct { // // With it on, snapshots are retained by height rather than by count, so the number kept becomes // RollbackWindow / SnapshotInterval instead of SnapshotKeepRecent + 1. - // - // Default: false ExternalPruning bool `mapstructure:"-"` // EnablePebbleMetrics defines if the Pebble metrics should be enabled. - // Default: true EnablePebbleMetrics bool `mapstructure:"enable-pebble-metrics"` // EnableReadWriteMetrics emits simple estimated read/write counters for FlatKV's Pebble DBs. - // Default: false EnableReadWriteMetrics bool `mapstructure:"enable-read-write-metrics"` // AccountDBConfig defines the PebbleDB configuration for the account database. @@ -133,8 +129,9 @@ func DefaultConfig() *Config { cfg := &Config{ Fsync: false, AsyncWriteBuffer: 0, - SnapshotInterval: DefaultSnapshotInterval, - SnapshotKeepRecent: DefaultSnapshotKeepRecent, + SnapshotInterval: 10000, + SnapshotKeepRecent: 1, + MaxSnapshotLagBlocks: 64, EnablePebbleMetrics: true, AccountDBConfig: pebbledb.DefaultConfig(), AccountStoreConfig: defaultStoreConfig("account"), diff --git a/sei-db/state_db/sc/flatkv/config/flatkv_test_config.go b/sei-db/state_db/sc/flatkv/config/flatkv_test_config.go index 034679bc31..20983577c0 100644 --- a/sei-db/state_db/sc/flatkv/config/flatkv_test_config.go +++ b/sei-db/state_db/sc/flatkv/config/flatkv_test_config.go @@ -28,8 +28,8 @@ func DefaultTestConfig(t *testing.T) *Config { t.Helper() return &Config{ DataDir: filepath.Join(t.TempDir(), "flatkv"), - SnapshotInterval: DefaultSnapshotInterval, - SnapshotKeepRecent: DefaultSnapshotKeepRecent, + SnapshotInterval: 10000, + SnapshotKeepRecent: 1, AccountDBConfig: smallTestPebbleConfig(), AccountStoreConfig: smallTestViewManagerConfig("account"), CodeDBConfig: smallTestPebbleConfig(), diff --git a/sei-db/state_db/sc/flatkv/importer.go b/sei-db/state_db/sc/flatkv/importer.go index dd70e37724..e00d816c61 100644 --- a/sei-db/state_db/sc/flatkv/importer.go +++ b/sei-db/state_db/sc/flatkv/importer.go @@ -303,7 +303,7 @@ func (imp *KVImporter) AddNode(node *types.SnapshotNode) { // Abort tears down the worker pipeline without finalizing the import. // It records reason as the first pipeline error (so any in-flight worker // also bails fast) and then runs Close, which observes the non-nil error -// and skips FinalizeImport / WriteSnapshot. The on-disk FlatKV directory +// and skips FinalizeImport / outOfBandSnapshot. The on-disk FlatKV directory // is left at its pre-import committed version, allowing the operator to // retry without --force. // @@ -325,7 +325,7 @@ func (imp *KVImporter) Abort(reason error) error { // Close on both the success and error paths. // // If the first pipeline error has already been recorded (either by a -// worker or by Abort), Close skips FinalizeImport / WriteSnapshot so the +// worker or by Abort), Close skips FinalizeImport / outOfBandSnapshot so the // store stays at its pre-import version. func (imp *KVImporter) Close() error { imp.finishOnce.Do(func() { @@ -375,7 +375,7 @@ func (imp *KVImporter) Close() error { // Write a snapshot so the imported data survives store reopen / restart. // Import bypasses the WAL, so without a snapshot the next LoadLatest // would clone from the pre-import snapshot and lose all imported data. - if err = imp.store.WriteSnapshot(""); err != nil { + if err = imp.store.outOfBandSnapshot(); err != nil { err = fmt.Errorf("failed to import when writing snapshot: %w", err) return } 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 90983220f2..c9573132dd 100644 --- a/sei-db/state_db/sc/flatkv/lthash_correctness_test.go +++ b/sei-db/state_db/sc/flatkv/lthash_correctness_test.go @@ -1292,7 +1292,7 @@ func TestLtHashSnapshotCatchupFullScan(t *testing.T) { for i := byte(1); i <= 3; i++ { commitMixedState(t, s1, i) } - require.NoError(t, s1.WriteSnapshot("")) + require.NoError(t, s1.outOfBandSnapshot()) // Blocks 4-7: more state (will need WAL catchup on reopen) for i := byte(4); i <= 7; i++ { @@ -1339,7 +1339,7 @@ func TestLtHashRollbackFullScan(t *testing.T) { for i := byte(1); i <= 5; i++ { commitMixedState(t, s, i) } - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) hashAtV5 := rootHash(s) for i := byte(6); i <= 8; i++ { @@ -1418,7 +1418,7 @@ func TestLtHashMultipleRollbacks(t *testing.T) { for i := byte(1); i <= 5; i++ { commitMixedState(t, s, i) } - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) // Original timeline: blocks 6-8 with round byte as-is for i := byte(6); i <= 8; i++ { diff --git a/sei-db/state_db/sc/flatkv/metrics.go b/sei-db/state_db/sc/flatkv/metrics.go index b9eaf859f1..874faf2de8 100644 --- a/sei-db/state_db/sc/flatkv/metrics.go +++ b/sei-db/state_db/sc/flatkv/metrics.go @@ -27,6 +27,7 @@ var ( CatchupLatency metric.Float64Histogram CatchupReplayNumBlocks metric.Int64Counter SnapshotWriteLatency metric.Float64Histogram + SnapshotQueueDepth metric.Int64Gauge SnapshotPruneLatency metric.Float64Histogram SnapshotPruneAttempts metric.Int64Counter CurrentSnapshotHeight metric.Int64Gauge @@ -98,6 +99,12 @@ var ( metric.WithUnit("s"), metric.WithExplicitBucketBoundaries(commonmetrics.LongLatencyBuckets...), )), + SnapshotQueueDepth: must(flatkvMeter.Int64Gauge( + "flatkv_snapshot_queue_depth", + metric.WithDescription( + "Committed blocks queued behind a FlatKV snapshot that is still being written"), + metric.WithUnit("{count}"), + )), SnapshotPruneLatency: must(flatkvMeter.Float64Histogram( "flatkv_snapshot_prune_latency", metric.WithDescription("Time taken to prune FlatKV snapshots"), diff --git a/sei-db/state_db/sc/flatkv/perdb_lthash_test.go b/sei-db/state_db/sc/flatkv/perdb_lthash_test.go index 74fadfcf95..76364bfc98 100644 --- a/sei-db/state_db/sc/flatkv/perdb_lthash_test.go +++ b/sei-db/state_db/sc/flatkv/perdb_lthash_test.go @@ -275,7 +275,7 @@ func TestPerDBLtHashCatchupReplay(t *testing.T) { commitMixedState(t, s1, 1) commitMixedState(t, s1, 2) - require.NoError(t, s1.WriteSnapshot("")) + require.NoError(t, s1.outOfBandSnapshot()) commitMixedState(t, s1, 3) commitMixedState(t, s1, 4) @@ -385,7 +385,7 @@ func TestPerDBLtHashRollback(t *testing.T) { commitMixedState(t, s, 1) commitMixedState(t, s, 2) commitMixedState(t, s, 3) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) commitMixedState(t, s, 4) commitMixedState(t, s, 5) diff --git a/sei-db/state_db/sc/flatkv/snapshot.go b/sei-db/state_db/sc/flatkv/snapshot.go index 654a6d4f0a..5efb8f61a1 100644 --- a/sei-db/state_db/sc/flatkv/snapshot.go +++ b/sei-db/state_db/sc/flatkv/snapshot.go @@ -1,6 +1,7 @@ package flatkv import ( + "context" "errors" "fmt" "io" @@ -9,9 +10,12 @@ import ( "sort" "strconv" "strings" + "sync" "time" + "github.com/sei-protocol/sei-chain/sei-db/common/metrics" "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" + "github.com/sei-protocol/sei-chain/sei-db/db_engine/view" "github.com/sei-protocol/sei-chain/sei-db/state_db/statewal" "go.opentelemetry.io/otel/metric" ) @@ -355,19 +359,17 @@ func (s *CommitStore) resolveSnapshotDir(flatkvDir string) (string, error) { return initDir, nil } -// WriteSnapshot creates a PebbleDB checkpoint of the committed state. -// The snapshot is written into a versioned subdirectory under the flatkv root -// (e.g. flatkv/snapshot-00000000000000000100) and the current symlink is updated. -// The dir parameter is ignored; snapshots are always stored alongside the live data. +// outOfBandSnapshot writes a snapshot of the committed state and does not return until it is on disk, +// whatever snapshot interval is configured. Snapshots are always stored under the flatkv root +// (e.g. flatkv/snapshot-00000000000000000100). // -// Concurrency: this MUST NOT acquire s.mu. Commit calls it while already holding -// the write lock (s.mu is not reentrant), and as a lifecycle operation it is -// otherwise expected to be serialized by the caller. It only reads committed -// state and checkpoints the DBs; it does not touch the pending-writes maps. -func (s *CommitStore) WriteSnapshot(_ string) (err error) { - var pruned int - obs := s.observeOp("snapshot", otelMetrics.SnapshotWriteLatency, - "version", s.committedVersion) +// NOT SAFE on a live store. It reads lastSealed and checkpoints the databases without taking s.mu, and +// it publishes into the same snapshot tree the background writer owns, so a concurrent Commit, +// ApplyChangeSets or read races it. The caller must have quiesced the store. It exists for the two +// bootstrap paths that need a snapshot at a height the cadence would decline — the end of an import, +// and a seeded initial version — and it must not grow a third caller that is merely "convenient". +func (s *CommitStore) outOfBandSnapshot() (err error) { + obs := s.observeOp("snapshot", otelMetrics.SnapshotWriteLatency, "version", s.committedVersion) defer obs.done(&err, func() { otelMetrics.CurrentSnapshotHeight.Record(s.ctx, s.committedVersion) }) @@ -380,78 +382,139 @@ func (s *CommitStore) WriteSnapshot(_ string) (err error) { return fmt.Errorf("cannot snapshot uncommitted store (version %d)", version) } - // Wait until the block we want to checkpoint has actually been flushed down to the pebble instances. - // Since we continue to hold the reservation on that block, later blocks are prevented from being - // flushed down to pebble, thus making the checkpoint operation thread safe. - if err := s.flushLatestVersion(); err != nil { - return fmt.Errorf("await flush before snapshot at version %d: %w", version, err) + // Let the cadence-driven writer finish whatever it has in flight. It writes into the same snapshot + // tree this is about to publish into, and only one writer of that tree may run at a time. + // + // The flush does not cover a retention cut line the collector may hand the writer, which arrives on + // its own channel. That one is safe to overlap: it deletes strictly below the active snapshot while + // this publishes above it, so a publication racing it can only make it delete less. + if s.snapshotWriter != nil { + if err := s.snapshotWriter.Flush(); err != nil { + return fmt.Errorf("await pending snapshot before writing version %d: %w", version, err) + } } - dir := s.flatkvDir() - snapDir := snapshotName(version) - finalPath := filepath.Join(dir, snapDir) - tmpPath := finalPath + tmpSuffix + tmpPath, err := checkpointDatabases( + s.ctx, s.flatkvDir(), version, s.lastSealed, s.checkpointables(), s.phaseTimer) + if err != nil { + return fmt.Errorf("checkpoint databases at version %d: %w", version, err) + } + pruned, err := publishSnapshot( + s.ctx, s.flatkvDir(), s.config.SnapshotKeepRecent, s.config.ExternalPruning, version, tmpPath) + if err != nil { + return fmt.Errorf("publish snapshot at version %d: %w", version, err) + } - _ = os.RemoveAll(tmpPath) + logger.Info("FlatKV snapshot created", + "version", version, "pruned", pruned, "elapsed", obs.elapsed()) + return nil +} - if err := os.MkdirAll(tmpPath, 0750); err != nil { - return fmt.Errorf("create snapshot tmp dir: %w", err) +// checkpointDatabases copies every database at version into a fresh temporary directory and returns +// its path. The directory is removed again if any part of the copy fails. +// +// The caller must hold a reservation on each view passed in, and must keep holding it until this +// returns. That is what stops a later block reaching Pebble mid-copy, and so what makes the result a +// view of exactly this version rather than of no single moment. +// +// phaseTimer reports the two halves of the call separately — waiting for the databases to reach this +// version, then copying them — because the reservation is held across both and they are the same +// duration to a caller measuring only the total. It may be nil. +func checkpointDatabases( + ctx context.Context, + dir string, + version int64, + views map[string]view.View, + dbs map[string]types.Checkpointable, + phaseTimer *metrics.PhaseTimer, +) (_ string, err error) { + // The databases are already flushing this block in the background; this waits for them to finish. + // On return Pebble holds exactly this block, and stays there while the reservations are held. + phaseTimer.SetPhase("snapshot_await_flush") + for name, sealed := range views { + if flushErr := sealed.AwaitFlush(ctx); flushErr != nil { + return "", fmt.Errorf("await flush of %s at version %d: %w", name, version, flushErr) + } } + phaseTimer.SetPhase("snapshot_copy_databases") - success := false + tmpPath := filepath.Join(dir, snapshotName(version)) + tmpSuffix + _ = os.RemoveAll(tmpPath) + if mkErr := os.MkdirAll(tmpPath, 0750); mkErr != nil { + return "", fmt.Errorf("create snapshot tmp dir: %w", mkErr) + } defer func() { - if !success { + if err != nil { _ = os.RemoveAll(tmpPath) } }() - // A checkpoint addresses a database as a file rather than as a key-value store, which is the one thing a - // view manager cannot express — so this is the single place FlatKV reaches past one, and the manager's - // escape hatch names checkpointing as its only sanctioned use. What makes it safe is the flush awaited - // above plus the reservation still held on that block: together they pin the pebble instances at exactly - // the committed version for the duration. - // - // dataDBDirs has a fixed iteration order, so the checkpoint is reproducible. - for _, dir := range dataDBDirs { - manager := s.viewManagerFor(dir) - if manager == nil { - return fmt.Errorf("no view manager for %s", dir) - } - cp, ok := manager.EscapeHatchUnderlyingDB().(types.Checkpointable) - if !ok { - return fmt.Errorf("db %s does not support Checkpoint", dir) - } - if err := cp.Checkpoint(filepath.Join(tmpPath, dir)); err != nil { - return fmt.Errorf("checkpoint %s: %w", dir, err) - } + // Copied concurrently: the pin holds every database at this version for the whole call, so the + // copies describe one moment no matter what order they run in. Serially, the pin — and with it the + // stall on every later block's flush — would last the sum of the four rather than the longest. + errs := make([]error, len(dataDBDirs)) + var wg sync.WaitGroup + for i, name := range dataDBDirs { + idx, dbName := i, name + wg.Add(1) + go func() { + defer wg.Done() + db, ok := dbs[dbName] + if !ok { + errs[idx] = fmt.Errorf("no checkpointable handle for db %s", dbName) + return + } + if cpErr := db.Checkpoint(filepath.Join(tmpPath, dbName)); cpErr != nil { + errs[idx] = fmt.Errorf("checkpoint %s: %w", dbName, cpErr) + } + }() + } + wg.Wait() + if err = errors.Join(errs...); err != nil { + return "", fmt.Errorf("checkpoint databases at version %d: %w", version, err) } + return tmpPath, nil +} + +// publishSnapshot makes a completed checkpoint directory the active snapshot: it takes the versioned +// name, the current symlink comes to point at it, and snapshots beyond the retention count are +// removed. Reports how many were removed. +// +// It touches no database, so a caller holding reservations may hand them back before calling this. +func publishSnapshot( + ctx context.Context, + dir string, + keepRecent uint32, + externalPruning bool, + version int64, + tmpPath string, +) (pruned int, err error) { + defer func() { + if err != nil { + _ = os.RemoveAll(tmpPath) + } + }() + + snapDir := snapshotName(version) + finalPath := filepath.Join(dir, snapDir) _ = atomicRemoveDir(finalPath) // idempotent: stale final may exist - if err := os.Rename(tmpPath, finalPath); err != nil { - return fmt.Errorf("rename snapshot dir: %w", err) + if err = os.Rename(tmpPath, finalPath); err != nil { + return 0, fmt.Errorf("rename snapshot dir: %w", err) } - if err := updateCurrentSymlink(dir, snapDir); err != nil { - return fmt.Errorf("update current symlink: %w", err) + if err = updateCurrentSymlink(dir, snapDir); err != nil { + return 0, fmt.Errorf("update current symlink: %w", err) } // Keep SNAPSHOT_BASE in sync so the next restart reuses the working dir // instead of re-cloning from the snapshot and replaying the full WAL gap. workDir := filepath.Join(dir, workingDirName) - if err := writeSnapshotBase(workDir, snapDir); err != nil { - logger.Error("failed to update SNAPSHOT_BASE", "err", err) + if baseErr := writeSnapshotBase(workDir, snapDir); baseErr != nil { + logger.Error("failed to update SNAPSHOT_BASE", "err", baseErr) } - pruned = s.pruneSnapshotsByCount(dir, version) - - success = true - s.lastSnapshotTime = time.Now() - logger.Info("FlatKV snapshot created", - "version", version, - "dir", finalPath, - "pruned", pruned, - "elapsed", obs.elapsed()) - return nil + return pruneSnapshotsByCount(ctx, dir, keepRecent, externalPruning, version), nil } // pruneSnapshotsByCount removes old snapshots beyond SnapshotKeepRecent, keeping @@ -463,19 +526,25 @@ func (s *CommitStore) WriteSnapshot(_ string) (err error) { // counting one as "old" would spend a keep slot on it and evict a genuinely older snapshot that rollback // still needs as a base. memiavl's pruneSnapshots applies the same guard. // -// Does nothing when config.ExternalPruning is set, which hands retention to the +// Does nothing when externalPruning is set, which hands retention to the // StorageGarbageCollector and its by-block-height PruneSnapshots. -func (s *CommitStore) pruneSnapshotsByCount(dir string, currentVersion int64) int { - if s.config.ExternalPruning { +func pruneSnapshotsByCount( + ctx context.Context, + dir string, + keepRecent uint32, + externalPruning bool, + currentVersion int64, +) int { + if externalPruning { return 0 } start := time.Now() defer func() { - otelMetrics.SnapshotPruneLatency.Record(s.ctx, secondsSince(start)) + otelMetrics.SnapshotPruneLatency.Record(ctx, secondsSince(start)) }() - keep := int(s.config.SnapshotKeepRecent) + keep := int(keepRecent) pruned := 0 var older []int64 @@ -496,7 +565,7 @@ func (s *CommitStore) pruneSnapshotsByCount(dir string, currentVersion int64) in for _, v := range older[keep:] { snapPath := filepath.Join(dir, snapshotName(v)) err := atomicRemoveDir(snapPath) - otelMetrics.SnapshotPruneAttempts.Add(s.ctx, 1, + otelMetrics.SnapshotPruneAttempts.Add(ctx, 1, metric.WithAttributes(successAttr(err))) if err != nil { logger.Error("prune snapshot failed", "version", v, "err", err) @@ -576,7 +645,7 @@ func (s *CommitStore) rollbackBaseVersion(dir string, targetVersion int64) (int6 // closed. Retrying in-process does not work, because establishing reachability reads the WAL's stored range // and that now fails as closed. No block is lost: the un-pruned WAL still holds them, so a restart replays // back to the old tail and the rollback can be retried. The errors from that window say so. Snapshots above -// the target are already gone by then, which costs a cached checkpoint the next WriteSnapshot rebuilds, not +// the target are already gone by then, which costs a cached checkpoint the next snapshot rebuilds, not // history. func (s *CommitStore) Rollback(targetVersion int64) (err error) { obs := s.observeOp("Rollback", otelMetrics.RollbackLatency, diff --git a/sei-db/state_db/sc/flatkv/snapshot_test.go b/sei-db/state_db/sc/flatkv/snapshot_test.go index 8e81e8fbfd..7d1177ac08 100644 --- a/sei-db/state_db/sc/flatkv/snapshot_test.go +++ b/sei-db/state_db/sc/flatkv/snapshot_test.go @@ -45,7 +45,7 @@ func TestSnapshotCreatesDir(t *testing.T) { commitStorageEntry(t, s, ktype.Address{0x01}, ktype.Slot{0x01}, []byte{0xAA}) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) flatkvDir := filepath.Join(dir, flatkvRootDir) @@ -75,8 +75,8 @@ func TestSnapshotIdempotent(t *testing.T) { commitStorageEntry(t, s, ktype.Address{0x02}, ktype.Slot{0x02}, []byte{0xBB}) - require.NoError(t, s.WriteSnapshot("")) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) + require.NoError(t, s.outOfBandSnapshot()) flatkvDir := filepath.Join(dir, flatkvRootDir) target, err := os.Readlink(currentPath(flatkvDir)) @@ -98,7 +98,7 @@ func TestOpenFromSnapshot(t *testing.T) { commitStorageEntry(t, s1, ktype.Address{0x10}, ktype.Slot{0x01}, []byte{0x01}) commitStorageEntry(t, s1, ktype.Address{0x10}, ktype.Slot{0x02}, []byte{0x02}) - require.NoError(t, s1.WriteSnapshot("")) + require.NoError(t, s1.outOfBandSnapshot()) require.Equal(t, int64(2), s1.Version()) commitStorageEntry(t, s1, ktype.Address{0x10}, ktype.Slot{0x03}, []byte{0x03}) @@ -143,7 +143,7 @@ func TestCatchupUpdatesLtHash(t *testing.T) { // Commit 5 versions, snapshot at v2 commitStorageEntry(t, s1, ktype.Address{0x20}, ktype.Slot{0x01}, []byte{0x10}) commitStorageEntry(t, s1, ktype.Address{0x20}, ktype.Slot{0x02}, []byte{0x20}) - require.NoError(t, s1.WriteSnapshot("")) + require.NoError(t, s1.outOfBandSnapshot()) commitStorageEntry(t, s1, ktype.Address{0x20}, ktype.Slot{0x03}, []byte{0x30}) hashAtV3 := rootHash(s1) @@ -179,7 +179,7 @@ func TestRollbackRewindsState(t *testing.T) { commitStorageEntry(t, s, ktype.Address{0x30}, ktype.Slot{0x01}, []byte{0x01}) commitStorageEntry(t, s, ktype.Address{0x30}, ktype.Slot{0x02}, []byte{0x02}) commitStorageEntry(t, s, ktype.Address{0x30}, ktype.Slot{0x03}, []byte{0x03}) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) commitStorageEntry(t, s, ktype.Address{0x30}, ktype.Slot{0x04}, []byte{0x04}) hashAtV4 := rootHash(s) @@ -215,7 +215,7 @@ func TestRollbackToSnapshotExact(t *testing.T) { commitStorageEntry(t, s, ktype.Address{0x40}, ktype.Slot{0x01}, []byte{0x01}) commitStorageEntry(t, s, ktype.Address{0x40}, ktype.Slot{0x02}, []byte{0x02}) hashAtV2 := rootHash(s) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) commitStorageEntry(t, s, ktype.Address{0x40}, ktype.Slot{0x03}, []byte{0x03}) require.Equal(t, int64(3), s.Version()) @@ -239,7 +239,7 @@ func TestPartialSnapshotCleanup(t *testing.T) { commitStorageEntry(t, s, ktype.Address{0x50}, ktype.Slot{0x01}, []byte{0x01}) // Take a valid snapshot first - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) flatkvDir := filepath.Join(dir, flatkvRootDir) prevTarget, err := os.Readlink(currentPath(flatkvDir)) @@ -252,8 +252,8 @@ func TestPartialSnapshotCleanup(t *testing.T) { // and the Close below simply reports the already-closed database. require.NoError(t, s.rawDBFor(codeDBDir).Close()) - err = s.WriteSnapshot("") - require.Error(t, err, "WriteSnapshot should fail when a DB is closed") + err = s.outOfBandSnapshot() + require.Error(t, err, "outOfBandSnapshot should fail when a DB is closed") // Current should still point to the previous snapshot target, err := os.Readlink(currentPath(flatkvDir)) @@ -373,7 +373,7 @@ func TestReadOnlyAtTargetVersion(t *testing.T) { commitStorageEntry(t, s1, ktype.Address{0x70}, ktype.Slot{0x01}, []byte{0x01}) commitStorageEntry(t, s1, ktype.Address{0x70}, ktype.Slot{0x02}, []byte{0x02}) - require.NoError(t, s1.WriteSnapshot("")) + require.NoError(t, s1.outOfBandSnapshot()) commitStorageEntry(t, s1, ktype.Address{0x70}, ktype.Slot{0x03}, []byte{0x03}) hashAtV3 := rootHash(s1) commitStorageEntry(t, s1, ktype.Address{0x70}, ktype.Slot{0x04}, []byte{0x04}) @@ -414,7 +414,7 @@ func TestSnapshotThenCatchupThenVerifyCorrectness(t *testing.T) { commitStorageEntry(t, s1, addr, slot, []byte{0x01}) // v1 commitStorageEntry(t, s1, ktype.Address{0x7A}, ktype.Slot{0x7C}, []byte{0xAA}) // v2 - require.NoError(t, s1.WriteSnapshot("")) + require.NoError(t, s1.outOfBandSnapshot()) // Record baseline value at v2 for the same key. vAtV2, ok := s1.Get(keys.EVMStoreKey, key) @@ -476,7 +476,7 @@ func TestReadOnlyAtIsUnaffectedByLoadLatest(t *testing.T) { commitStorageEntry(t, s, addr, slot, []byte{0x01}) commitStorageEntry(t, s, addr, slot, []byte{0x02}) hashAtV2 := rootHash(s) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) commitStorageEntry(t, s, addr, slot, []byte{0x03}) commitStorageEntry(t, s, addr, slot, []byte{0x04}) @@ -531,7 +531,7 @@ func TestRollbackToSnapshotVersion(t *testing.T) { commitStorageEntry(t, s, ktype.Address{0x90}, ktype.Slot{0x01}, []byte{0x01}) commitStorageEntry(t, s, ktype.Address{0x90}, ktype.Slot{0x02}, []byte{0x02}) hashAtV2 := rootHash(s) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) commitStorageEntry(t, s, ktype.Address{0x90}, ktype.Slot{0x03}, []byte{0x03}) commitStorageEntry(t, s, ktype.Address{0x90}, ktype.Slot{0x04}, []byte{0x04}) @@ -580,7 +580,7 @@ func rollbackFixture(t *testing.T) *CommitStore { for i := byte(1); i <= 5; i++ { commitStorageEntry(t, s, ktype.Address{0x91}, ktype.Slot{i}, []byte{i}) if i == 2 { - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) } } return s @@ -642,7 +642,7 @@ func rollbackFixtureEmptyWALAtV2(t *testing.T) *CommitStore { commitStorageEntry(t, s, ktype.Address{0x92}, ktype.Slot{0x01}, []byte{0x01}) commitStorageEntry(t, s, ktype.Address{0x92}, ktype.Slot{0x02}, []byte{0x02}) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) resetWALForTest(t, s) return s } @@ -958,7 +958,7 @@ func TestPruneSnapshotsKeepsRecent(t *testing.T) { for i := 0; i < 5; i++ { commitStorageEntry(t, s, ktype.Address{byte(i + 1)}, ktype.Slot{byte(i + 1)}, []byte{byte(i + 1)}) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) } var snapshots []int64 @@ -984,7 +984,7 @@ func TestPruneSnapshotsKeepAll(t *testing.T) { for i := 0; i < 3; i++ { commitStorageEntry(t, s, ktype.Address{byte(i + 1)}, ktype.Slot{byte(i + 1)}, []byte{byte(i + 1)}) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) } var count int @@ -1013,7 +1013,8 @@ func TestPruneSnapshotsIgnoresSnapshotsAboveCurrent(t *testing.T) { require.NoError(t, os.MkdirAll(filepath.Join(dir, snapshotName(v)), 0750)) } - require.Equal(t, 0, s.pruneSnapshotsByCount(dir, 30), + require.Equal(t, 0, + pruneSnapshotsByCount(s.ctx, dir, s.config.SnapshotKeepRecent, s.config.ExternalPruning, 30), "only 10 and 20 sit below the current version, and KeepRecent=2 covers both") var remaining []int64 @@ -1116,12 +1117,12 @@ func TestRollbackRemovesPostTargetSnapshots(t *testing.T) { for i := 0; i < 3; i++ { commitStorageEntry(t, s, ktype.Address{byte(i + 1)}, ktype.Slot{byte(i + 1)}, []byte{byte(i + 1)}) } - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) for i := 3; i < 6; i++ { commitStorageEntry(t, s, ktype.Address{byte(i + 1)}, ktype.Slot{byte(i + 1)}, []byte{byte(i + 1)}) } - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) for i := 6; i < 8; i++ { commitStorageEntry(t, s, ktype.Address{byte(i + 1)}, ktype.Slot{byte(i + 1)}, []byte{byte(i + 1)}) @@ -1209,11 +1210,11 @@ func TestRollbackReportsUnremovableSnapshotWithoutRewinding(t *testing.T) { for i := 0; i < 3; i++ { commitStorageEntry(t, s, ktype.Address{byte(i + 1)}, ktype.Slot{byte(i + 1)}, []byte{byte(i + 1)}) } - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) for i := 3; i < 6; i++ { commitStorageEntry(t, s, ktype.Address{byte(i + 1)}, ktype.Slot{byte(i + 1)}, []byte{byte(i + 1)}) } - require.NoError(t, s.WriteSnapshot("")) // snapshot-6, above the rollback target below + require.NoError(t, s.outOfBandSnapshot()) // snapshot-6, above the rollback target below // atomicRemoveDir renames snapshot-6 onto this trash name before unlinking it, so an undeletable // directory already sitting there fails that rename. Restore permissions before t.TempDir's own cleanup, @@ -1299,7 +1300,7 @@ func TestMultipleSnapshotsAndReopen(t *testing.T) { var hashes [][]byte for i := 0; i < 3; i++ { commitStorageEntry(t, s, ktype.Address{byte(i + 1)}, ktype.Slot{byte(i + 1)}, []byte{byte(i + 1)}) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) hashes = append(hashes, rootHash(s)) } require.NoError(t, s.Close()) @@ -1326,7 +1327,7 @@ func TestMultipleSnapshotsAndReopen(t *testing.T) { // Snapshot with all key types // ============================================================================= -func TestWriteSnapshotUpdatesSnapshotBase(t *testing.T) { +func TestOutOfBandSnapshotUpdatesSnapshotBase(t *testing.T) { dir := t.TempDir() cfg := config.DefaultTestConfig(t) cfg.DataDir = filepath.Join(dir, flatkvRootDir) @@ -1337,7 +1338,7 @@ func TestWriteSnapshotUpdatesSnapshotBase(t *testing.T) { commitStorageEntry(t, s, ktype.Address{0xF0}, ktype.Slot{0x01}, []byte{0x01}) commitStorageEntry(t, s, ktype.Address{0xF0}, ktype.Slot{0x02}, []byte{0x02}) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) flatkvDir := filepath.Join(dir, flatkvRootDir) workDir := filepath.Join(flatkvDir, workingDirName) @@ -1392,7 +1393,7 @@ func TestSnapshotPreservesAllKeyTypes(t *testing.T) { require.NoError(t, err) hash := rootHash(s) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) require.NoError(t, s.Close()) cfg = config.DefaultTestConfig(t) @@ -1634,7 +1635,7 @@ func TestSingleDBOpenFailure(t *testing.T) { err = s.LoadLatest() require.NoError(t, err) commitStorageEntry(t, s, ktype.Address{0x01}, ktype.Slot{0x01}, []byte{0xAA}) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) require.NoError(t, s.Close()) workingStorage := filepath.Join(dbDir, "working", storageDBDir) @@ -1674,7 +1675,7 @@ func TestWALDirectoryDeleted(t *testing.T) { commitStorageEntry(t, s, ktype.Address{0x01}, ktype.Slot{0x01}, []byte{0xAA}) commitStorageEntry(t, s, ktype.Address{0x02}, ktype.Slot{0x02}, []byte{0xBB}) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) require.NoError(t, s.Close()) walDir := filepath.Join(dbDir, changelogDir) @@ -1710,7 +1711,7 @@ func TestLocalMetaCorruption(t *testing.T) { err = s.LoadLatest() require.NoError(t, err) commitStorageEntry(t, s, ktype.Address{0x01}, ktype.Slot{0x01}, []byte{0xAA}) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) require.NoError(t, s.Close()) // Corrupt accountDB meta version in working dir: write 3 garbage bytes (expected 8). @@ -2141,7 +2142,7 @@ func TestRollbackPreservesWALContinuity(t *testing.T) { require.Equal(t, hashAfterNewCommits, rootHash(s2)) } -func TestWriteSnapshotOnReadOnlyStore(t *testing.T) { +func TestOutOfBandSnapshotOnReadOnlyStore(t *testing.T) { s := setupTestStore(t) cs := makeChangeSet( @@ -2155,22 +2156,22 @@ func TestWriteSnapshotOnReadOnlyStore(t *testing.T) { require.NoError(t, err) defer ro.Close() - err = ro.WriteSnapshot("") + err = ro.(*CommitStore).outOfBandSnapshot() require.Error(t, err) require.ErrorIs(t, err, errReadOnly) require.NoError(t, s.Close()) } -func TestWriteSnapshotAtVersion0(t *testing.T) { +func TestOutOfBandSnapshotAtVersion0(t *testing.T) { s := setupTestStore(t) defer s.Close() - err := s.WriteSnapshot("") + err := s.outOfBandSnapshot() require.Error(t, err, "snapshot at version 0 should fail") require.Contains(t, err.Error(), "cannot snapshot uncommitted store") } -func TestWriteSnapshotWhileReadOnlyCloneActive(t *testing.T) { +func TestOutOfBandSnapshotWhileReadOnlyCloneActive(t *testing.T) { s := setupTestStore(t) cs := makeChangeSet( @@ -2184,8 +2185,8 @@ func TestWriteSnapshotWhileReadOnlyCloneActive(t *testing.T) { require.NoError(t, err) defer ro.Close() - // WriteSnapshot should succeed even with active RO clone. - require.NoError(t, s.WriteSnapshot("")) + // outOfBandSnapshot should succeed even with active RO clone. + require.NoError(t, s.outOfBandSnapshot()) // RO clone should still work. val, found := ro.Get(keys.EVMStoreKey, keys.BuildEVMKey(keys.EVMKeyStorage, ktype.StorageKey(addrN(0x07), slotN(0x01)))) @@ -2193,23 +2194,3 @@ func TestWriteSnapshotWhileReadOnlyCloneActive(t *testing.T) { require.Equal(t, padLeft32(0x77), val) require.NoError(t, s.Close()) } - -func TestWriteSnapshotDirParameterIgnored(t *testing.T) { - s := setupTestStore(t) - defer s.Close() - - cs := makeChangeSet( - keys.BuildEVMKey(keys.EVMKeyStorage, ktype.StorageKey(addrN(0x08), slotN(0x01))), - padLeft32(0x88), false, - ) - require.NoError(t, s.ApplyChangeSets(s.Version()+1, []*proto.NamedChangeSet{cs})) - commitAndCheck(t, s) - - // Pass a non-empty dir parameter. The implementation should ignore it. - require.NoError(t, s.WriteSnapshot("/tmp/this-should-be-ignored")) - - // Verify snapshot was created in the correct location (not the passed dir). - val, found := s.Get(keys.EVMStoreKey, keys.BuildEVMKey(keys.EVMKeyStorage, ktype.StorageKey(addrN(0x08), slotN(0x01)))) - require.True(t, found) - require.Equal(t, padLeft32(0x88), val) -} diff --git a/sei-db/state_db/sc/flatkv/snapshot_writer.go b/sei-db/state_db/sc/flatkv/snapshot_writer.go new file mode 100644 index 0000000000..d3b4f7bcfa --- /dev/null +++ b/sei-db/state_db/sc/flatkv/snapshot_writer.go @@ -0,0 +1,416 @@ +package flatkv + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "go.opentelemetry.io/otel/metric" + + "github.com/sei-protocol/sei-chain/sei-db/common/metrics" + "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" + "github.com/sei-protocol/sei-chain/sei-db/db_engine/view" +) + +// ErrSnapshotWriterClosed is reported (wrapped) by calls that observe the writer shutting down +// normally rather than failing. Detect it with errors.Is. +var ErrSnapshotWriterClosed = errors.New("snapshot writer closed") + +// snapshotQueueScrapeInterval is how often the writer reports its queue depth. Matches the cadence the +// view managers sample their own gauges at. +const snapshotQueueScrapeInterval = 10 * time.Second + +// SnapshotWriter decides which committed blocks become snapshots and writes them asynchronously, and +// deletes the snapshots a retention cut line has made unnecessary. Its goroutine is the only one that +// mutates the snapshot tree on a live store. +// +// The writer has no recoverable errors. The first internal failure is latched and every subsequent call +// reports it, so a failure that has no caller to fail at the time it happens still stops the node: +// Offer is on the commit path, so the next Commit fails. +type SnapshotWriter struct { + // mu guards fatalErr. + mu sync.Mutex + + // dir is the flatkv root holding the snapshot directories, the current symlink and the working dir. + dir string + + // keepRecent is how many snapshots below the newest to retain. Ignored when externalPruning is set. + keepRecent uint32 + + // externalPruning stands this writer's count-based pruning down in favour of the + // StorageGarbageCollector's by-height retention. + externalPruning bool + + // interval is how many blocks apart snapshots are taken. 0 disables them. + interval uint32 + + // dbs is the handle each database is checkpointed through, keyed by database directory name. + // Captured when the view managers were opened, and valid until they are closed. + dbs map[string]types.Checkpointable + + // ctx is the context checkpoint work runs under. Cancelled by stop, and by the store's own context. + ctx context.Context + + // stop cancels ctx, telling the background goroutine to finish and releasing anyone waiting on it. + stop context.CancelFunc + + // messages is the queue. Its capacity is how many blocks may pile up behind a snapshot before + // offering another one blocks, which is the whole of the writer's backpressure. + messages chan any + + // pruneCutLine holds the retention cut line the StorageGarbageCollector last asked for, until + // the writer acts on it. Capacity 1: a cut line still waiting is replaced by the next rather + // than queued behind it, cut lines only ever rising. + pruneCutLine chan uint64 + + // exited is closed once the background goroutine has returned. + exited chan struct{} + + // phaseTimer breaks down where the writer's goroutine spends its time. Driven only by that + // goroutine, since a PhaseTimer instance is not safe for concurrent use. + phaseTimer *metrics.PhaseTimer + + // fatalErr latches the first failure. Nil until something fails. + fatalErr error +} + +// newSnapshotWriter starts a writer for the given databases. Close stops it. +// +// queueDepth is how many blocks may pile up behind a snapshot before offering another one blocks. A +// value below 1 is treated as 1. +// +// parent is the store's context: cancelling it stops the writer too, which matters because the store +// cancels its own context during teardown. +func newSnapshotWriter( + parent context.Context, + dir string, + keepRecent uint32, + externalPruning bool, + interval uint32, + queueDepth uint32, + dbs map[string]types.Checkpointable, +) *SnapshotWriter { + ctx, stop := context.WithCancel(parent) + w := &SnapshotWriter{ + dir: dir, + keepRecent: keepRecent, + externalPruning: externalPruning, + interval: interval, + dbs: dbs, + ctx: ctx, + stop: stop, + messages: make(chan any, max(queueDepth, 1)), + pruneCutLine: make(chan uint64, 1), + exited: make(chan struct{}), + phaseTimer: metrics.NewPhaseTimer(flatkvMeter, "seidb_snapshot_writer"), + } + go w.run() + go w.reportQueueDepth() + return w +} + +// Offer hands a committed block to the writer, which decides if it should be written to disk. +// +// The writer takes its own reservation on every view for as long as it needs one, and hands it back +// whether it writes a snapshot, declines to, or fails. The caller only has to hold a reservation of its +// own until this returns, and so does not have to know whether the writer keeps the block past the call. +func (w *SnapshotWriter) Offer(version int64, views map[string]view.View) error { + reserved, err := reserveViews(views) + if err != nil { + return fmt.Errorf("reserve version %d for snapshot: %w", version, err) + } + + request := &snapshotRequest{version: version, views: reserved} + if err := w.enqueue(request); err != nil { + return errors.Join( + fmt.Errorf("offer version %d to snapshot writer: %w", version, err), + request.release()) + } + return nil +} + +// PruneBelow hands the writer a retention cut line: every snapshot strictly below it may go. It +// returns as soon as the cut line is recorded, without waiting for the deletion, and reports the +// latched error if the writer has failed. +// +// A cut line still waiting when a newer one arrives is replaced rather than queued behind it. Cut +// lines only ever rise, so the newest is the only one that carries information. +func (w *SnapshotWriter) PruneBelow(cutLine uint64) error { + if err := w.errorIfBricked(); err != nil { + return fmt.Errorf("snapshot writer failed: %w", err) + } + + select { + case <-w.pruneCutLine: // discard the cut line this one supersedes + default: + } + select { + case w.pruneCutLine <- cutLine: + default: + // Only reachable with a second caller, which refilled the cell between the two selects. Its + // cut line supersedes this one in turn, so dropping this one costs nothing. + } + return nil +} + +// Flush blocks until the writer has dealt with every block offered so far, including a snapshot it is +// part way through. It reports the latched error if the writer has failed. +// +// It is not a barrier for PruneBelow, which delivers on its own channel. +func (w *SnapshotWriter) Flush() error { + request := newFlushRequest() + if err := w.enqueue(request); err != nil { + return fmt.Errorf("flush snapshot writer: %w", err) + } + select { + case <-request.responseChan: + if err := w.errorIfBricked(); err != nil { + return fmt.Errorf("flush snapshot writer: %w", err) + } + return nil + case <-w.ctx.Done(): + return fmt.Errorf("flush snapshot writer: %w", w.stoppedError()) + } +} + +// Close stops the writer and waits for its goroutine to exit, which may include finishing blocks that +// are still queued. Reports the latched error if the writer failed. Idempotent. +func (w *SnapshotWriter) Close() error { + w.stop() + // The goroutine closes exited from a deferred call on every exit path, so this cannot strand. + <-w.exited + if err := w.errorIfBricked(); err != nil { + return fmt.Errorf("close snapshot writer: %w", err) + } + return nil +} + +// enqueue puts a message on the queue, blocking while the queue is full, and reports why it could not +// when the writer has stopped instead. Cleaning up after a message it could not deliver belongs to the +// caller, which is the only one that knows whether the message owns anything. +func (w *SnapshotWriter) enqueue(message any) error { + if err := w.errorIfBricked(); err != nil { + return fmt.Errorf("snapshot writer failed: %w", err) + } + + select { + case w.messages <- message: + return nil + case <-w.ctx.Done(): + return fmt.Errorf("enqueue to snapshot writer: %w", w.stoppedError()) + } +} + +// shouldSnapshot reports whether a committed block becomes a snapshot. Snapshots are taken every +// interval blocks; an interval of 0 disables them, at the cost of a WAL that grows without bound and a +// restart that replays the whole history. +func (w *SnapshotWriter) shouldSnapshot(version int64) bool { + if w.interval == 0 || version <= 0 { + return false + } + return version%int64(w.interval) == 0 +} + +// reportQueueDepth samples how many blocks are waiting behind the snapshot being written and updates +// metrics. +func (w *SnapshotWriter) reportQueueDepth() { + ticker := time.NewTicker(snapshotQueueScrapeInterval) + defer ticker.Stop() + for { + select { + case <-w.ctx.Done(): + return + case <-ticker.C: + otelMetrics.SnapshotQueueDepth.Record(w.ctx, int64(len(w.messages))) + } + } +} + +// run acts on the blocks offered to the writer and the cut lines handed to it, until the writer is +// stopped or one of them fails. +func (w *SnapshotWriter) run() { + defer close(w.exited) + // Whatever is still queued is owed a hand-back, so nothing is left holding a reservation that would + // stall its database for good. + defer w.discardQueued() + + for { + // Charged for however long the queue stays empty, so a writer that is never idle is one the + // cadence is outrunning. + w.phaseTimer.SetPhase("idle") + + var err error + select { + case <-w.ctx.Done(): + return + case cutLine := <-w.pruneCutLine: + err = w.handlePruneCutLine(cutLine) + case message := <-w.messages: + err = w.handleMessage(message) + } + if err != nil { + w.brick(err) + return + } + } +} + +// handleMessage acts on one message from the queue. +func (w *SnapshotWriter) handleMessage(message any) error { + switch request := message.(type) { + case *snapshotRequest: + return w.maybeCheckpointBlock(request) + case *flushRequest: + request.responseChan <- struct{}{} + return nil + default: + return fmt.Errorf("unknown snapshot writer message type %T", message) + } +} + +// handlePruneCutLine deletes the snapshots a retention cut line from the StorageGarbageCollector has +// made unnecessary. +// +// A failure here stops the writer, as a failed checkpoint does. A snapshot that will not unlink means +// the storage underneath is misconfigured or broken, and a node that cannot reclaim snapshots is on a +// terminal path to a full disk, so there is nothing to recover to. +func (w *SnapshotWriter) handlePruneCutLine(cutLine uint64) error { + w.phaseTimer.SetPhase("prune_snapshots") + if err := pruneSnapshotsBelow(w.ctx, w.dir, cutLine); err != nil { + return fmt.Errorf("prune snapshots below cut line %d: %w", cutLine, err) + } + return nil +} + +// Possibly checkpoint a block. Releases reservation when finished regardless of choice. +func (w *SnapshotWriter) maybeCheckpointBlock(request *snapshotRequest) (err error) { + // The only hand-back for a block that reached the goroutine, covering written, declined and failed + // alike. A reservation left held stalls its view manager's flushes indefinitely. + defer func() { + if relErr := request.release(); relErr != nil { + err = errors.Join(err, fmt.Errorf( + "hand back reservations for version %d: %w", request.version, relErr)) + } + }() + + if !w.shouldSnapshot(request.version) { + w.phaseTimer.SetPhase("release_declined_block") + return nil + } + if err := w.writeCheckpoint(request); err != nil { + return fmt.Errorf("write snapshot at version %d: %w", request.version, err) + } + return nil +} + +// discardQueued empties the queue, handing back what each snapshot request holds and answering each +// flush so its caller is not left waiting. A message enqueued after this has run is stranded, which +// only happens once the writer has stopped — the view managers are closing by then, and closing one +// releases everything it holds. +func (w *SnapshotWriter) discardQueued() { + for { + select { + case message := <-w.messages: + switch request := message.(type) { + case *snapshotRequest: + if err := request.release(); err != nil { + logger.Error("failed to hand back reservations of a discarded snapshot", + "version", request.version, "err", err) + } + case *flushRequest: + request.responseChan <- struct{}{} + } + default: + return + } + } +} + +// writeCheckpoint writes one snapshot: the databases are copied while they are pinned, and the copy is published +// as the active snapshot. It records how long that took, and reports a failure that by then has no +// caller to return to. +func (w *SnapshotWriter) writeCheckpoint(request *snapshotRequest) (err error) { + start := time.Now() + defer func() { + otelMetrics.SnapshotWriteLatency.Record(w.ctx, secondsSince(start), + metric.WithAttributes(successAttr(err))) + if err != nil { + logger.Error("FlatKV snapshot failed", + "version", request.version, "elapsed", time.Since(start), "err", err) + } + }() + + // Work already under way is not abandoned when the writer is told to stop. w.ctx is cancelled to + // release callers blocked on the queue, but Close is documented to let an in-flight snapshot finish, + // and the databases it is reading are closed only after the drain. Handing it a cancellable context + // would instead abort its AwaitFlush and brick the writer on the way out. + workCtx := context.WithoutCancel(w.ctx) + + tmpPath, err := checkpointDatabases( + workCtx, w.dir, request.version, request.views, w.dbs, w.phaseTimer) + if err != nil { + return fmt.Errorf("snapshot version %d: %w", request.version, err) + } + + w.phaseTimer.SetPhase("publish_snapshot") + pruned, err := publishSnapshot( + workCtx, w.dir, w.keepRecent, w.externalPruning, request.version, tmpPath) + if err != nil { + return fmt.Errorf("publish snapshot at version %d: %w", request.version, err) + } + + otelMetrics.CurrentSnapshotHeight.Record(w.ctx, request.version) + logger.Info("FlatKV snapshot created", + "version", request.version, "pruned", pruned, "elapsed", time.Since(start)) + return nil +} + +// brick latches err as the writer's fatal error and stops the writer. +// +// Stopping is what turns the failure into an error rather than a hang: with the goroutine gone nothing +// drains the queue, so a caller blocked on a full queue or waiting on a flush would wait forever. +// Cancelling the context releases them to read the latched error instead. +func (w *SnapshotWriter) brick(err error) { + w.mu.Lock() + if w.fatalErr == nil { + w.fatalErr = err + } + w.mu.Unlock() + w.stop() +} + +// errorIfBricked reports the latched error, or nil if the writer has not failed. The error is returned +// as latched, for whoever propagates it to describe what they were doing. +func (w *SnapshotWriter) errorIfBricked() error { + w.mu.Lock() + defer w.mu.Unlock() + return w.fatalErr +} + +// stoppedError reports why the writer is no longer running: the latched error if it failed, otherwise +// that it was closed. Never nil. +func (w *SnapshotWriter) stoppedError() error { + if err := w.errorIfBricked(); err != nil { + return fmt.Errorf("snapshot writer failed: %w", err) + } + return ErrSnapshotWriterClosed +} + +// reserveViews takes a reservation on each of the given views, for a consumer that will outlive +// whoever already holds one. Every reservation taken is handed back if any one of them fails, since a +// caller that gets an error takes ownership of nothing. +func reserveViews(views map[string]view.View) (map[string]view.View, error) { + reserved := make(map[string]view.View, len(views)) + for name, sealed := range views { + if err := sealed.Reserve(); err != nil { + for _, taken := range reserved { + _ = taken.Release() + } + return nil, fmt.Errorf("reserve %s view: %w", name, err) + } + reserved[name] = sealed + } + return reserved, nil +} diff --git a/sei-db/state_db/sc/flatkv/snapshot_writer_messages.go b/sei-db/state_db/sc/flatkv/snapshot_writer_messages.go new file mode 100644 index 0000000000..208b6b0024 --- /dev/null +++ b/sei-db/state_db/sc/flatkv/snapshot_writer_messages.go @@ -0,0 +1,51 @@ +package flatkv + +import ( + "errors" + "fmt" + + "github.com/sei-protocol/sei-chain/sei-db/db_engine/view" +) + +// This file contains the messages that can be sent to the snapshot writer's goroutine. + +// snapshotRequest is a committed block offered to the writer, which decides whether it becomes a +// snapshot. +type snapshotRequest struct { + // version is the block height this snapshot would capture. + version int64 + + // views is the block's sealed view for each database, keyed by database directory name. Each + // carries a reservation this request owns and must hand back exactly once — a second Release on a + // view bricks its manager. + views map[string]view.View +} + +// release hands back every reservation this request holds, so the databases can resume writing out +// later blocks. The goroutine owns this for a request it received; Offer owns it only for one it took +// reservations for but could not enqueue. +// +// Every reservation is handed back even if one of them fails, because a reservation left held stalls +// its database's flushes indefinitely. The failures are joined and returned. +func (r *snapshotRequest) release() error { + var errs []error + for name, sealed := range r.views { + if relErr := sealed.Release(); relErr != nil { + errs = append(errs, + fmt.Errorf("release %s view at version %d: %w", name, r.version, relErr)) + } + } + return errors.Join(errs...) +} + +// flushRequest asks the writer to report once it has dealt with everything enqueued ahead of it. +type flushRequest struct { + // responseChan produces a value once every message enqueued ahead of this one has been dealt with. + // Buffered, so the writer answering it cannot block on a caller that has already given up. + responseChan chan struct{} +} + +// newFlushRequest describes a wait for the writer to catch up. +func newFlushRequest() *flushRequest { + return &flushRequest{responseChan: make(chan struct{}, 1)} +} diff --git a/sei-db/state_db/sc/flatkv/snapshot_writer_test.go b/sei-db/state_db/sc/flatkv/snapshot_writer_test.go new file mode 100644 index 0000000000..74ba885fff --- /dev/null +++ b/sei-db/state_db/sc/flatkv/snapshot_writer_test.go @@ -0,0 +1,427 @@ +package flatkv + +import ( + "context" + "errors" + "os" + "path/filepath" + "slices" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" + "github.com/sei-protocol/sei-chain/sei-db/db_engine/view" + "github.com/sei-protocol/sei-chain/sei-db/proto" + "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/flatkv/config" + "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/flatkv/ktype" +) + +// The bulk of this package's suite reaches the writer through commitAndCheck, which flushes it so a +// test can look at the snapshot tree straight after committing. These tests are the ones that exercise +// the writer's own goroutine, so they build a writer directly over stubs rather than going through a +// store. + +var _ view.View = (*fakeView)(nil) + +// fakeView is a view whose flush and hand-back outcomes the test chooses, and which counts both. The +// methods a SnapshotWriter never reaches panic, so a use this stub was not written for is loud rather +// than silently wrong. +type fakeView struct { + // Reported by Name. + name string + + // Returned by AwaitFlush. + awaitFlushErr error + + // Returned by Reserve. A non-nil value also suppresses the reserve count. + reserveErr error + + // Counts successful Reserve calls. + reserves atomic.Int64 + + // Counts Release calls. + releases atomic.Int64 +} + +func (v *fakeView) Name() string { return v.name } + +func (v *fakeView) AwaitFlush(context.Context) error { return v.awaitFlushErr } + +func (v *fakeView) Reserve() error { + if v.reserveErr != nil { + return v.reserveErr + } + v.reserves.Add(1) + return nil +} + +func (v *fakeView) Release() error { + v.releases.Add(1) + return nil +} + +func (v *fakeView) Get([]byte, bool) ([]byte, bool, error) { + panic("fakeView: unexpected Get") +} + +func (v *fakeView) BatchGet([][]byte) (map[string][]byte, error) { + panic("fakeView: unexpected BatchGet") +} + +func (v *fakeView) GetDiff() (map[string][]byte, error) { + panic("fakeView: unexpected GetDiff") +} + +func (v *fakeView) Finalize([]*proto.KVPair) error { + panic("fakeView: unexpected Finalize") +} + +// fakeViews returns one stub per database, as a commit would hand to the writer. +func fakeViews() (map[string]view.View, map[string]*fakeView) { + views := make(map[string]view.View, len(dataDBDirs)) + stubs := make(map[string]*fakeView, len(dataDBDirs)) + for _, name := range dataDBDirs { + stub := &fakeView{name: name} + views[name] = stub + stubs[name] = stub + } + return views, stubs +} + +// requireAllReleased asserts the writer handed back every reservation it took. A reservation left held +// stalls its database's flushes forever, so this is the invariant every path must preserve. +func requireAllReleased(t *testing.T, stubs map[string]*fakeView) { + t.Helper() + for name, stub := range stubs { + require.NotZero(t, stub.reserves.Load(), "%s: the writer must take its own reservation", name) + require.Equal(t, stub.reserves.Load(), stub.releases.Load(), + "%s: the writer must hand back every reservation it took", name) + } +} + +var _ types.Checkpointable = (*fakeCheckpointDB)(nil) + +// fakeCheckpointDB is a Checkpointable whose Checkpoint the test controls: it announces that it has +// started, then blocks until released, then optionally fails. +type fakeCheckpointDB struct { + // Closed the first time Checkpoint is called. + started chan struct{} + + // Checkpoint does not return until this is closed. Nil means do not block. + release chan struct{} + + // Returned by Checkpoint. + err error + + // Guards started against the concurrent checkpoints of the four databases. + once sync.Once +} + +func (d *fakeCheckpointDB) Checkpoint(destDir string) error { + d.once.Do(func() { close(d.started) }) + if d.release != nil { + <-d.release + } + if d.err != nil { + return d.err + } + return os.MkdirAll(destDir, 0o750) +} + +// fakeCheckpointDBs returns the same controllable handle for every database, so a single release +// channel gates the whole checkpoint. +func fakeCheckpointDBs(db *fakeCheckpointDB) map[string]types.Checkpointable { + dbs := make(map[string]types.Checkpointable, len(dataDBDirs)) + for _, name := range dataDBDirs { + dbs[name] = db + } + return dbs +} + +// newTestWriter builds a writer over stubs, writing into a temp dir. +func newTestWriter(t *testing.T, interval uint32, queueDepth uint32, db *fakeCheckpointDB) *SnapshotWriter { + t.Helper() + dir := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(dir, workingDirName), 0o750)) + return newSnapshotWriter(t.Context(), dir, 0, false, interval, queueDepth, fakeCheckpointDBs(db)) +} + +// newCollectorPrunedTestWriter is newTestWriter with retention handed to the StorageGarbageCollector, +// so the only deletions in the writer's snapshot tree are the ones a cut line asks for. The prune +// tests fix snapshots on disk and would otherwise watch the writer's own count-based pruner remove +// them. +func newCollectorPrunedTestWriter( + t *testing.T, + interval uint32, + queueDepth uint32, + db *fakeCheckpointDB, +) *SnapshotWriter { + t.Helper() + dir := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(dir, workingDirName), 0o750)) + return newSnapshotWriter(t.Context(), dir, 0, true, interval, queueDepth, fakeCheckpointDBs(db)) +} + +// requireBlocked asserts a call has not returned yet. +func requireBlocked(t *testing.T, returned <-chan error, what string) { + t.Helper() + select { + case err := <-returned: + t.Fatalf("%s returned early: %v", what, err) + case <-time.After(100 * time.Millisecond): + } +} + +// requireReturns waits for a call to return and yields its error. +func requireReturns(t *testing.T, returned <-chan error, what string) error { + t.Helper() + select { + case err := <-returned: + return err + case <-time.After(5 * time.Second): + t.Fatalf("%s never returned", what) + return nil + } +} + +// The queue's depth is the whole of the writer's backpressure: blocks pile up behind a snapshot that is +// still being written, and once the queue is full offering another one waits. That pause is deliberate — +// a snapshot holds the databases pinned, so the blocks behind it are held in memory. +func TestSnapshotWriterBlocksOnceQueueIsFull(t *testing.T) { + db := &fakeCheckpointDB{started: make(chan struct{}), release: make(chan struct{})} + w := newTestWriter(t, 1, 1, db) + + first, firstStubs := fakeViews() + require.NoError(t, w.Offer(1, first)) + <-db.started // taken off the queue, and will not finish until released + + queued, queuedStubs := fakeViews() + require.NoError(t, w.Offer(2, queued), "the single queue slot is free") + + blocked, blockedStubs := fakeViews() + returned := make(chan error, 1) + go func() { returned <- w.Offer(3, blocked) }() + requireBlocked(t, returned, "Offer with a full queue") + + close(db.release) + require.NoError(t, requireReturns(t, returned, "Offer after the queue drained")) + + require.NoError(t, w.Flush()) + require.NoError(t, w.Close()) + for _, stubs := range []map[string]*fakeView{firstStubs, queuedStubs, blockedStubs} { + requireAllReleased(t, stubs) + } +} + +// A commit blocked on a full queue must be released when the writer is closed underneath it. Nothing +// else wakes it: the snapshot it is queued behind holds the databases pinned, and teardown is what tells +// it that wait will never be satisfied. Getting this wrong deadlocks shutdown against block production +// rather than failing. +func TestSnapshotWriterCloseWakesBlockedOffer(t *testing.T) { + db := &fakeCheckpointDB{started: make(chan struct{}), release: make(chan struct{})} + w := newTestWriter(t, 1, 1, db) + + first, _ := fakeViews() + require.NoError(t, w.Offer(1, first)) + <-db.started + + queued, _ := fakeViews() + require.NoError(t, w.Offer(2, queued)) + + blocked, blockedStubs := fakeViews() + offered := make(chan error, 1) + go func() { offered <- w.Offer(3, blocked) }() + requireBlocked(t, offered, "Offer with a full queue") + + closed := make(chan error, 1) + go func() { closed <- w.Close() }() + + err := requireReturns(t, offered, "Offer after Close") + require.ErrorIs(t, err, ErrSnapshotWriterClosed, + "a blocked commit must be told the writer stopped rather than waiting forever") + requireAllReleased(t, blockedStubs) + + close(db.release) + require.NoError(t, requireReturns(t, closed, "Close")) +} + +// Close waits for an in-flight checkpoint rather than abandoning it, because that checkpoint holds +// handles to databases the caller is about to close. Whatever is still queued behind it is discarded, +// with its reservations handed back. +func TestSnapshotWriterCloseWaitsForCheckpointAndDiscardsQueue(t *testing.T) { + db := &fakeCheckpointDB{started: make(chan struct{}), release: make(chan struct{})} + w := newTestWriter(t, 1, 4, db) + + first, firstStubs := fakeViews() + require.NoError(t, w.Offer(7, first)) + <-db.started + + queued, queuedStubs := fakeViews() + require.NoError(t, w.Offer(8, queued)) + + closed := make(chan error, 1) + go func() { closed <- w.Close() }() + requireBlocked(t, closed, "Close while a checkpoint was reading the databases") + + close(db.release) + require.NoError(t, requireReturns(t, closed, "Close")) + requireAllReleased(t, firstStubs) + requireAllReleased(t, queuedStubs) +} + +// A block the cadence does not select is handed back unwritten, by the goroutine rather than the +// caller. Flush is how a test observes that the goroutine has got that far. +func TestSnapshotWriterReleasesBlocksItDoesNotSnapshot(t *testing.T) { + db := &fakeCheckpointDB{started: make(chan struct{})} + w := newTestWriter(t, 10, 8, db) + defer func() { require.NoError(t, w.Close()) }() + + all := make([]map[string]*fakeView, 0, 3) + for version := int64(1); version <= 3; version++ { + views, stubs := fakeViews() + require.NoError(t, w.Offer(version, views)) + all = append(all, stubs) + } + + require.NoError(t, w.Flush()) + for _, stubs := range all { + requireAllReleased(t, stubs) + } +} + +// The first failure is latched and reported by every later call. Bricking also stops the writer, which +// is what makes the failure an error rather than a hang: with the goroutine gone nothing drains the +// queue, so a caller would otherwise block on it forever. +func TestSnapshotWriterBrickStopsWriterAndReportsToEveryCaller(t *testing.T) { + failure := errors.New("checkpoint exploded") + db := &fakeCheckpointDB{started: make(chan struct{}), err: failure} + w := newTestWriter(t, 1, 1, db) + + views, stubs := fakeViews() + require.NoError(t, w.Offer(1, views)) + + require.Eventually(t, func() bool { + return errors.Is(w.Flush(), failure) + }, 5*time.Second, 5*time.Millisecond, "the failure must be latched and reported") + requireAllReleased(t, stubs) + + later, laterStubs := fakeViews() + require.ErrorIs(t, w.Offer(2, later), failure, + "Offer is on the commit path, so it must surface the failure rather than block on a dead queue") + requireAllReleased(t, laterStubs) + + require.ErrorIs(t, w.Flush(), failure) + require.ErrorIs(t, w.Close(), failure, "Close reports what went wrong rather than hiding it") +} + +// reserveViews takes ownership of nothing when it fails, so a partial success must be undone. Map +// iteration order is unspecified, so the assertion is per-view rather than a total count: whichever +// ones were reserved, those are the ones that must have been released. +func TestReserveViewsUnwindsPartialSuccess(t *testing.T) { + ok := &fakeView{name: accountDBDir} + bad := &fakeView{name: codeDBDir, reserveErr: errors.New("manager is bricked")} + + reserved, err := reserveViews(map[string]view.View{ + accountDBDir: ok, + codeDBDir: bad, + }) + require.Error(t, err) + require.ErrorContains(t, err, "manager is bricked") + require.Nil(t, reserved, "a failed reserve must not hand back a partial set") + + require.Equal(t, ok.reserves.Load(), ok.releases.Load(), + "a reservation taken before the failure must be handed back, not stranded") + require.Zero(t, bad.releases.Load(), "a reservation that was never taken must not be released") +} + +// The collector's cut line is acted on by the writer's goroutine, not the collector's. With a +// checkpoint in flight the writer cannot reach it, so PruneBelow returns with the snapshots still on +// disk; they go once the writer is free. This is what makes the writer the only mutator of the +// snapshot tree, which is the whole point of routing deletion through it. +func TestSnapshotWriterPruneWaitsForTheWritersGoroutine(t *testing.T) { + db := &fakeCheckpointDB{started: make(chan struct{}), release: make(chan struct{})} + w := newCollectorPrunedTestWriter(t, 1, 1, db) + mkSnapshots(t, w.dir, 5, 10, 20) + + views, stubs := fakeViews() + require.NoError(t, w.Offer(100, views)) + <-db.started // the goroutine is inside the checkpoint and cannot service a cut line + + require.NoError(t, w.PruneBelow(20)) + require.Equal(t, []int64{5, 10, 20}, snapshotVersions(t, w.dir), + "nothing may be deleted on the collector's goroutine") + + close(db.release) + require.Eventually(t, func() bool { + return slices.Equal([]int64{20, 100}, snapshotVersions(t, w.dir)) + }, 5*time.Second, 10*time.Millisecond, "the writer deletes once it is free") + + require.NoError(t, w.Close()) + requireAllReleased(t, stubs) +} + +// Cut lines only ever rise, so one still waiting carries nothing the newer one does not. The cell +// holds exactly one, and it is the newest. +func TestSnapshotWriterPruneCutLineIsSuperseded(t *testing.T) { + db := &fakeCheckpointDB{started: make(chan struct{}), release: make(chan struct{})} + w := newCollectorPrunedTestWriter(t, 1, 1, db) + + views, stubs := fakeViews() + require.NoError(t, w.Offer(100, views)) + <-db.started // nothing drains the cell while the checkpoint runs + + require.NoError(t, w.PruneBelow(6)) + require.NoError(t, w.PruneBelow(11)) + require.Len(t, w.pruneCutLine, 1, "a waiting cut line is replaced, not queued behind") + require.Equal(t, uint64(11), <-w.pruneCutLine, "the newest cut line is the one kept") + + close(db.release) + require.NoError(t, w.Close()) + requireAllReleased(t, stubs) +} + +// A deletion that fails stops the writer, as a failed checkpoint does. Here "current" is missing, so +// there is nothing to bound the cut line by and the prune refuses rather than running blind. +func TestSnapshotWriterPruneFailureBricks(t *testing.T) { + w := newCollectorPrunedTestWriter(t, 0, 1, &fakeCheckpointDB{started: make(chan struct{})}) + mkSnapshots(t, w.dir, 5, 10) + require.NoError(t, os.Remove(currentPath(w.dir))) + + require.NoError(t, w.PruneBelow(10), "the refusal happens on the writer, not at the hand-off") + require.Eventually(t, func() bool { + return w.errorIfBricked() != nil + }, 5*time.Second, 10*time.Millisecond) + + require.ErrorContains(t, w.Close(), "resolve active snapshot") + require.Equal(t, []int64{5, 10}, snapshotVersions(t, w.dir), "a refused prune deletes nothing") +} + +// End to end through a store with the writer running asynchronously: the snapshot appears once +// FlushSnapshots says the writer has caught up, and names the height that was committed. +func TestStoreWritesSnapshotAsynchronously(t *testing.T) { + cfg := config.DefaultTestConfig(t) + cfg.SnapshotInterval = 2 + cfg.MaxSnapshotLagBlocks = 1000 // asynchronous, unlike the rest of the suite + s := setupTestStoreWithConfig(t, cfg) + defer func() { _ = s.Close() }() + + commitStorageEntry(t, s, ktype.Address{0x01}, ktype.Slot{0x01}, []byte{0xaa}) + commitStorageEntry(t, s, ktype.Address{0x02}, ktype.Slot{0x02}, []byte{0xbb}) + + require.NoError(t, s.FlushSnapshots()) + + dir := s.flatkvDir() + for _, sub := range dataDBDirs { + info, err := os.Stat(filepath.Join(dir, snapshotName(2), sub)) + require.NoError(t, err, "%s should exist in the asynchronously written snapshot", sub) + require.True(t, info.IsDir()) + } + + target, err := os.Readlink(currentPath(dir)) + require.NoError(t, err) + require.Equal(t, snapshotName(2), target, "current must point at the snapshot the writer published") +} diff --git a/sei-db/state_db/sc/flatkv/store.go b/sei-db/state_db/sc/flatkv/store.go index b2b78bed34..9a020e5d05 100644 --- a/sei-db/state_db/sc/flatkv/store.go +++ b/sei-db/state_db/sc/flatkv/store.go @@ -9,7 +9,6 @@ import ( "runtime" "strings" "sync" - "time" "github.com/zbiljic/go-filelock" "go.opentelemetry.io/otel/attribute" @@ -158,7 +157,10 @@ type CommitStore struct { // nor Commit validates its version against this field. pendingBlockHeight int64 - lastSnapshotTime time.Time + // Writes snapshots off the execution thread. Built by openStores once the view managers exist and + // torn down by closeStores, so its lifetime is exactly the window in which the databases it + // checkpoints are open. Nil on a read-only store, which never commits. + snapshotWriter *SnapshotWriter // File lock prevents multiple processes from opening the same DB. fileLock filelock.TryLockerSafe @@ -869,9 +871,40 @@ func (s *CommitStore) openStores(dbs rawDBs) (retErr error) { return err } + if !s.readOnly { + // Built last, and only here: it checkpoints the databases the view managers above own, so it must + // not outlive them. closeStores drains it before those managers go away. + s.snapshotWriter = newSnapshotWriter( + s.ctx, + s.flatkvDir(), + s.config.SnapshotKeepRecent, + s.config.ExternalPruning, + s.config.SnapshotInterval, + s.config.MaxSnapshotLagBlocks, + s.checkpointables(), + ) + } + return nil } +// checkpointables returns the handle each database is checkpointed through, keyed by database +// directory name. Captured once while the view managers exist, so a snapshot being written off-thread +// never has to reach back into the store for a handle that teardown may have cleared. +// +// A checkpoint addresses a database as a file rather than as a key-value store, which is the one thing +// a view manager cannot express — so this is the single place FlatKV reaches past one, and the +// manager's escape hatch names checkpointing as its only sanctioned use. +func (s *CommitStore) checkpointables() map[string]seidbtypes.Checkpointable { + dbs := make(map[string]seidbtypes.Checkpointable, len(dataDBDirs)) + for _, name := range dataDBDirs { + if db, ok := s.rawDBFor(name).(seidbtypes.Checkpointable); ok { + dbs[name] = db + } + } + return dbs +} + // viewManagerFor returns the view manager mediating the named database, or nil before the managers exist. // // It answers with the manager, not the database beneath it, so that every access through it is an access the @@ -913,6 +946,17 @@ func (s *CommitStore) rawDBFor(name string) seidbtypes.KeyValueDB { func (s *CommitStore) closeStores() error { var errs []error + // The writer must stop before anything below runs: closing a view manager closes the database it + // owns, and a checkpoint in progress would then be reading a closed handle. This is the choke point + // every teardown path reaches — Close directly, Rollback and resetForImport through closeDBsOnly — + // so the guard lives here rather than at each of them. + if s.snapshotWriter != nil { + if err := s.snapshotWriter.Close(); err != nil { + errs = append(errs, fmt.Errorf("close snapshot writer: %w", err)) + } + s.snapshotWriter = nil + } + // Hand back the reservations on the last sealed block and forget the handles. They belong to the // stores being torn down here, so keeping them would leave a reopened store (rollback, restore) // awaiting a flush on views whose store is already gone. diff --git a/sei-db/state_db/sc/flatkv/store_apply.go b/sei-db/state_db/sc/flatkv/store_apply.go index 446e769f7b..361e1a9f4a 100644 --- a/sei-db/state_db/sc/flatkv/store_apply.go +++ b/sei-db/state_db/sc/flatkv/store_apply.go @@ -18,7 +18,7 @@ import ( func (s *CommitStore) ApplyChangeSets(version int64, changeSets []*proto.NamedChangeSet) error { // The read-only refusal belongs here rather than in applyChangeSets, which a read-only store reaches // legitimately: building a view at a past height replays the primary's WAL through the same apply path. - // Commit and WriteSnapshot place their refusals at the same boundary, and readOnly is fixed for a store's + // Commit and outOfBandSnapshot place their refusals at the same boundary, and readOnly is fixed for a store's // lifetime, so reading it outside the lock is safe. if s.readOnly { return errReadOnly diff --git a/sei-db/state_db/sc/flatkv/store_gc.go b/sei-db/state_db/sc/flatkv/store_gc.go index 5a93de5e40..ea131f53db 100644 --- a/sei-db/state_db/sc/flatkv/store_gc.go +++ b/sei-db/state_db/sc/flatkv/store_gc.go @@ -1,10 +1,14 @@ package flatkv import ( + "context" "errors" "fmt" "os" "path/filepath" + "time" + + "go.opentelemetry.io/otel/metric" "github.com/sei-protocol/sei-chain/sei-db/controller" ) @@ -32,33 +36,70 @@ func (s *CommitStore) PruneHistory(uint64) error { return nil } -// PruneSnapshots deletes every snapshot strictly below blockNumber, never the active snapshot. A -// snapshot that fails to delete is reported while the rest are still attempted; one that is already -// gone is not an error. +// PruneSnapshots hands blockNumber to the snapshot writer as a retention cut line and returns +// without waiting for the deletion. It reports only that the cut line was refused, which happens +// when the writer has already failed. +// +// Deletion runs on the writer's goroutine because that goroutine is the only one allowed to mutate +// the snapshot tree on a live store: it is also what publishes snapshots into that tree. Doing it +// here instead would race a publication that is renaming a directory in and moving "current". func (s *CommitStore) PruneSnapshots(blockNumber uint64) error { if blockNumber == 0 { return nil } - blocks, err := s.snapshotBlocks() + writer := s.currentSnapshotWriter() + if writer == nil { + return nil + } + if err := writer.PruneBelow(blockNumber); err != nil { + return fmt.Errorf("hand cut line %d to the snapshot writer: %w", blockNumber, err) + } + return nil +} + +// currentSnapshotWriter returns the writer this store currently holds, or nil when it has none: a +// read-only store, a closed one, or one between closeStores and openStores. A cycle that finds nil +// is skipped, the next one carrying a cut line at or above the one it dropped. +func (s *CommitStore) currentSnapshotWriter() *SnapshotWriter { + s.mu.RLock() + defer s.mu.RUnlock() + return s.snapshotWriter +} + +// pruneSnapshotsBelow deletes every snapshot under dir strictly below cutLine, never the active +// snapshot. A snapshot that fails to delete is reported while the rest are still attempted; one +// that is already gone is not an error. +// +// ctx carries only the metric recording; the deletion itself is not cancellable. +func pruneSnapshotsBelow(ctx context.Context, dir string, cutLine uint64) error { + start := time.Now() + defer func() { + otelMetrics.SnapshotPruneLatency.Record(ctx, secondsSince(start)) + }() + + blocks, err := snapshotBlocks(dir) if err != nil { return fmt.Errorf("scan snapshots: %w", err) } if len(blocks) == 0 { return nil } - active, err := s.activeSnapshotHeight() + active, err := activeSnapshotHeight(dir) if err != nil { - return fmt.Errorf("prune snapshots below %d: %w", blockNumber, err) + return fmt.Errorf("bound the cut line by the active snapshot: %w", err) } - cutLine := min(blockNumber, active) + // The active snapshot is what the next open resolves through, so the cut line stops there + // however deep it was asked to go. + bounded := min(cutLine, active) var errs error pruned := 0 for _, block := range blocks { - if block >= cutLine { + if block >= bounded { break // ascending, so nothing further is a candidate } - removed, err := s.deleteSnapshot(block) + removed, err := deleteSnapshot(dir, block) + otelMetrics.SnapshotPruneAttempts.Add(ctx, 1, metric.WithAttributes(successAttr(err))) if err != nil { errs = errors.Join(errs, err) continue @@ -69,16 +110,16 @@ func (s *CommitStore) PruneSnapshots(blockNumber uint64) error { } if pruned > 0 { - logger.Info("pruned snapshots below the rollback cut line", "count", pruned, "cutLine", cutLine) + logger.Info("pruned snapshots below the rollback cut line", "count", pruned, "cutLine", bounded) } return errs } -// activeSnapshotHeight returns the height of the snapshot "current" points at — the one the next open -// clones and replays the state WAL forward from. It is usually the newest snapshot on disk, but a -// crash during WriteSnapshot or a partial Rollback can leave it lower. -func (s *CommitStore) activeSnapshotHeight() (uint64, error) { - _, version, err := currentSnapshotDir(s.flatkvDir()) +// activeSnapshotHeight returns the height of the snapshot "current" points at under dir — the one +// the next open clones and replays the state WAL forward from. It is usually the newest snapshot on +// disk, but a crash while a snapshot was being published, or a partial Rollback, can leave it lower. +func activeSnapshotHeight(dir string) (uint64, error) { + _, version, err := currentSnapshotDir(dir) if err != nil { return 0, fmt.Errorf("resolve active snapshot: %w", err) } @@ -124,11 +165,11 @@ func snapshotFloor(blocks []uint64, head uint64, rollbackWindow uint64) uint64 { return oldest } -// snapshotBlocks returns the block number of every snapshot on disk, ascending. A missing snapshot +// snapshotBlocks returns the block number of every snapshot under dir, ascending. A missing snapshot // directory yields no blocks rather than an error. -func (s *CommitStore) snapshotBlocks() ([]uint64, error) { +func snapshotBlocks(dir string) ([]uint64, error) { var blocks []uint64 - err := traverseSnapshots(s.flatkvDir(), true, func(version int64) (bool, error) { + err := traverseSnapshots(dir, true, func(version int64) (bool, error) { if version >= 0 { blocks = append(blocks, uint64(version)) } @@ -140,10 +181,10 @@ func (s *CommitStore) snapshotBlocks() ([]uint64, error) { return blocks, nil } -// deleteSnapshot removes the snapshot directory for block, reporting whether this call is the one that -// removed it. An already-gone snapshot is not an error and reports false. -func (s *CommitStore) deleteSnapshot(block uint64) (bool, error) { - path := filepath.Join(s.flatkvDir(), snapshotName(int64(block))) //nolint:gosec // block numbers are bounded well below 2^63 +// deleteSnapshot removes the snapshot directory for block under dir, reporting whether this call is +// the one that removed it. An already-gone snapshot is not an error and reports false. +func deleteSnapshot(dir string, block uint64) (bool, error) { + path := filepath.Join(dir, snapshotName(int64(block))) //nolint:gosec // block numbers are bounded well below 2^63 if err := atomicRemoveDir(path); err != nil { if os.IsNotExist(err) { return false, nil @@ -160,7 +201,7 @@ func (s *CommitStore) deleteSnapshot(block uint64) (bool, error) { // It returns 0 — nothing here is eligible for pruning — when there is no snapshot to name, when the // window is deeper than the head, or when the snapshot layout cannot be read. func (s *CommitStore) GetRollbackFloor(rollbackWindow uint64) uint64 { - blocks, err := s.snapshotBlocks() + blocks, err := snapshotBlocks(s.flatkvDir()) if err != nil { logger.Error("failed to scan snapshots for the rollback floor; holding it at 0", "rollbackWindow", rollbackWindow, "err", err) @@ -176,7 +217,7 @@ func (s *CommitStore) GetRollbackFloor(rollbackWindow uint64) uint64 { if floor == 0 { return 0 } - active, err := s.activeSnapshotHeight() + active, err := activeSnapshotHeight(s.flatkvDir()) if err != nil { logger.Error("failed to resolve the active snapshot for the rollback floor; holding it at 0", "rollbackWindow", rollbackWindow, "err", err) diff --git a/sei-db/state_db/sc/flatkv/store_gc_test.go b/sei-db/state_db/sc/flatkv/store_gc_test.go index 1d8cce5fa3..e579f50e1e 100644 --- a/sei-db/state_db/sc/flatkv/store_gc_test.go +++ b/sei-db/state_db/sc/flatkv/store_gc_test.go @@ -3,7 +3,9 @@ package flatkv import ( "os" "path/filepath" + "slices" "testing" + "time" "github.com/sei-protocol/sei-chain/sei-db/controller" @@ -15,7 +17,7 @@ import ( // The GC surface reads the snapshot directory and two plain fields, so most of it can be exercised // without opening the five PebbleDBs a real store carries. Only the tests that assert against -// snapshots WriteSnapshot produced, or against a concurrent committer, pay for a live store. +// snapshots the writer produced, or against a concurrent committer, pay for a live store. func gcStore(t *testing.T, dir string) (*CommitStore, controller.PrunableStore) { t.Helper() s := &CommitStore{config: config.Config{DataDir: dir}} @@ -32,7 +34,7 @@ func gcStoreAtHead(t *testing.T, dir string, head int64) (*CommitStore, controll } // mkSnapshots creates snapshot directories and points "current" at the newest one on disk, which is -// the shape production leaves behind: WriteSnapshot repoints the symlink at each snapshot it writes. +// the shape production leaves behind: publishing repoints the symlink at each snapshot written. // The GC surface is bounded by the active snapshot, so a test that left "current" unset would be // measuring a state a live store never reaches. The tests that want a stale or missing symlink say so // after calling this. @@ -211,20 +213,20 @@ func TestGCRollbackFloorScanFailureHoldsHistory(t *testing.T) { require.Equal(t, uint64(0), store.GetRollbackFloor(10)) } -// PruneSnapshots drops everything strictly below the height it is given, and the height need not -// have a snapshot on it. +// The retention policy drops everything strictly below the height it is given, and the height need +// not have a snapshot on it. Exercised directly rather than through PruneSnapshots, which only hands +// the cut line to the snapshot writer — see TestGCPruneSnapshotsDefersToTheWriter. func TestGCPruneSnapshotsDeletesBelowTheCutLine(t *testing.T) { - s, store := gcStoreAtHead(t, t.TempDir(), 30) - dir := s.flatkvDir() + dir := t.TempDir() mkSnapshots(t, dir, 0, 5, 10, 20, 30) require.NoError(t, updateCurrentSymlink(dir, snapshotName(30))) // A cut line of 25 has no snapshot on it; 20 is the newest below it and goes with the rest. - require.NoError(t, store.PruneSnapshots(25)) + require.NoError(t, pruneSnapshotsBelow(t.Context(), dir, 25)) require.Equal(t, []int64{30}, snapshotVersions(t, dir)) // Idempotent: the same cut line twice deletes nothing more. - require.NoError(t, store.PruneSnapshots(25)) + require.NoError(t, pruneSnapshotsBelow(t.Context(), dir, 25)) require.Equal(t, []int64{30}, snapshotVersions(t, dir)) } @@ -241,7 +243,9 @@ func TestGCPruneSnapshotsKeepsTheSnapshotItReported(t *testing.T) { for rollbackWindow := uint64(0); rollbackWindow <= 40; rollbackWindow++ { floor := store.GetRollbackFloor(rollbackWindow) before := snapshotVersions(t, dir) - require.NoError(t, store.PruneSnapshots(floor)) + if floor > 0 { + require.NoError(t, pruneSnapshotsBelow(t.Context(), dir, floor)) + } remaining := snapshotVersions(t, dir) if floor == 0 { @@ -258,12 +262,11 @@ func TestGCPruneSnapshotsKeepsTheSnapshotItReported(t *testing.T) { // A cut line at or below the oldest snapshot leaves the lot. It must not be read as "delete // everything below the newest". func TestGCPruneSnapshotsBelowTheOldestSnapshotIsNoOp(t *testing.T) { - s, store := gcStoreAtHead(t, t.TempDir(), 1_000) - dir := s.flatkvDir() + dir := t.TempDir() mkSnapshots(t, dir, 500, 900) require.NoError(t, updateCurrentSymlink(dir, snapshotName(900))) - require.NoError(t, store.PruneSnapshots(500)) + require.NoError(t, pruneSnapshotsBelow(t.Context(), dir, 500)) require.Equal(t, []int64{500, 900}, snapshotVersions(t, dir)) } @@ -279,41 +282,54 @@ func TestGCPruneSnapshotsAtZeroIsNoOp(t *testing.T) { require.Equal(t, []int64{5, 10}, snapshotVersions(t, dir)) } +// A store with no snapshot writer has no goroutine to do the deletion on, so the cycle is skipped +// rather than performed here. The next cycle carries a cut line at or above this one, so nothing is +// lost by declining. This is the shape of a read-only store, a closed one, and one reopening. +func TestGCPruneSnapshotsWithoutAWriterIsSkipped(t *testing.T) { + s, store := gcStoreAtHead(t, t.TempDir(), 30) + dir := s.flatkvDir() + mkSnapshots(t, dir, 5, 10, 20) + require.NoError(t, updateCurrentSymlink(dir, snapshotName(20))) + + require.Nil(t, s.snapshotWriter, "the fixture builds a store that never opened") + require.NoError(t, store.PruneSnapshots(20)) + require.Equal(t, []int64{5, 10, 20}, snapshotVersions(t, dir)) +} + // A store with no snapshots has nothing to prune, which is not a failure — and notably not an error // about the missing active symlink, since there is no deletion to protect. func TestGCPruneSnapshotsWithoutSnapshots(t *testing.T) { - s, store := gcStoreAtHead(t, t.TempDir(), 100) - require.NoError(t, store.PruneSnapshots(10)) + dir := t.TempDir() + require.NoError(t, pruneSnapshotsBelow(t.Context(), dir, 10)) - require.NoError(t, os.RemoveAll(s.flatkvDir())) - require.NoError(t, store.PruneSnapshots(10)) + require.NoError(t, os.RemoveAll(dir)) + require.NoError(t, pruneSnapshotsBelow(t.Context(), dir, 10)) } // The active snapshot is what the next open resolves to, so the cut line stops there even when it is -// asked to go deeper. This is the shape a crash between WriteSnapshot's rename and its symlink update +// asked to go deeper. This is the shape a crash between publishing's rename and its symlink update // leaves behind: snapshot 10 is on disk while "current" is still 5. Deleting 5 would leave a dangling // symlink, which os.Readlink resolves happily, so the store would open against a directory that is // not there. func TestGCPruneSnapshotsKeepsActiveSnapshotBelowTheCutLine(t *testing.T) { - s, store := gcStoreAtHead(t, t.TempDir(), 10) - dir := s.flatkvDir() + dir := t.TempDir() mkSnapshots(t, dir, 3, 5, 10) require.NoError(t, updateCurrentSymlink(dir, snapshotName(5))) - require.NoError(t, store.PruneSnapshots(10)) + require.NoError(t, pruneSnapshotsBelow(t.Context(), dir, 10)) require.Equal(t, []int64{5, 10}, snapshotVersions(t, dir), "the cut line stops at the active snapshot, so 3 goes and 5 stays") } // Without a resolvable active snapshot there is nothing to bound the deletion by, so the prune is -// refused rather than run blind. +// refused rather than run blind. The refusal reaches the snapshot writer, which stops on it — see +// TestSnapshotWriterPruneFailureBricks. func TestGCPruneSnapshotsRefusesWithoutActiveSnapshot(t *testing.T) { - s, store := gcStoreAtHead(t, t.TempDir(), 10) - dir := s.flatkvDir() + dir := t.TempDir() mkSnapshots(t, dir, 5, 10) require.NoError(t, os.Remove(currentPath(dir))) - require.Error(t, store.PruneSnapshots(10)) + require.Error(t, pruneSnapshotsBelow(t.Context(), dir, 10)) require.Equal(t, []int64{5, 10}, snapshotVersions(t, dir)) } @@ -330,7 +346,7 @@ func TestGCRollbackFloorHoldsHistoryWithoutActiveSnapshot(t *testing.T) { require.Equal(t, uint64(0), store.GetRollbackFloor(80)) } -// A snapshot newer than "current" is what a crash between WriteSnapshot's rename and its symlink +// A snapshot newer than "current" is what a crash between publishing's rename and its symlink // update leaves behind, and the next open takes the symlink rather than adopting the orphan. So the // floor stops at the active snapshot: reporting the orphan would hold the WAL only from there, and // the replay that starts at the active snapshot needs the blocks below it. @@ -370,7 +386,7 @@ func TestGCExternalPruningStandsDownSnapshotPruner(t *testing.T) { }, } mkSnapshots(t, dir, 5, 10, 15) - s.pruneSnapshotsByCount(dir, 15) + pruneSnapshotsByCount(s.ctx, dir, s.config.SnapshotKeepRecent, s.config.ExternalPruning, 15) return snapshotVersions(t, dir) } @@ -423,7 +439,7 @@ func TestGCExternalPruningStandsDownWALTruncation(t *testing.T) { require.False(t, prune(true).pruned, "under ExternalPruning tryTruncateWAL must not touch the WAL") } -// End to end against snapshots WriteSnapshot actually produced, including that the store still opens +// End to end against snapshots the writer actually produced, including that the store still opens // afterwards — the prune must not disturb what the next open resolves through. func TestGCPrunesRealSnapshotsAndStoreStillOpens(t *testing.T) { dir := t.TempDir() @@ -455,8 +471,11 @@ func TestGCPrunesRealSnapshotsAndStoreStillOpens(t *testing.T) { floor := store.GetRollbackFloor(1) require.Equal(t, uint64(4), floor, "newest snapshot at or below head - 1") + // The deletion runs on the writer's goroutine, so it is not done when PruneSnapshots returns. require.NoError(t, store.PruneSnapshots(floor)) - require.Equal(t, []int64{4, 6}, snapshotVersions(t, cfg.DataDir)) + require.Eventually(t, func() bool { + return slices.Equal([]int64{4, 6}, snapshotVersions(t, cfg.DataDir)) + }, 5*time.Second, 10*time.Millisecond) require.NoError(t, s.Close()) @@ -471,7 +490,7 @@ func TestGCPrunesRealSnapshotsAndStoreStillOpens(t *testing.T) { // The collector runs on its own goroutine while the store commits on another. Only the race detector // can judge this: CommitStore keeps its committed version in a plain field that Commit advances under -// the write lock, and WriteSnapshot rewrites the snapshot directory the other two methods scan. +// the write lock, and publishing rewrites the snapshot directory the other two methods scan. func TestGCConcurrentWithCommitter(t *testing.T) { dir := t.TempDir() cfg := config.DefaultTestConfig(t) diff --git a/sei-db/state_db/sc/flatkv/store_iteration_stability_test.go b/sei-db/state_db/sc/flatkv/store_iteration_stability_test.go index ac9417f236..a38831d5bb 100644 --- a/sei-db/state_db/sc/flatkv/store_iteration_stability_test.go +++ b/sei-db/state_db/sc/flatkv/store_iteration_stability_test.go @@ -153,9 +153,12 @@ func TestEvmIteratorSurvivesAutoSnapshot(t *testing.T) { return s.Iterator(keys.EVMStoreKey, nil, nil, true) }) - // Block 2 trips the snapshot interval, which forces a flush of everything committed so far. + // Block 2 trips the snapshot interval, which forces a flush of everything committed so far. The + // snapshot is written off the execution thread, so wait for it: without that this test drains the + // iterator before the checkpoint has touched the databases, and stops testing anything. applyAndCommitBlock(t, s, touchEveryLane(0x01, 99, 0xbb)) require.Equal(t, int64(2), s.Version()) + require.NoError(t, s.FlushSnapshots()) require.Equal(t, before, collectIterEntries(t, iter)) require.NoError(t, iter.Close()) diff --git a/sei-db/state_db/sc/flatkv/store_meta.go b/sei-db/state_db/sc/flatkv/store_meta.go index 661077a0a9..913713b07d 100644 --- a/sei-db/state_db/sc/flatkv/store_meta.go +++ b/sei-db/state_db/sc/flatkv/store_meta.go @@ -382,7 +382,7 @@ func (s *CommitStore) SetInitialVersion(initialVersion int64) error { } if seededVersion > 0 { - if err := s.WriteSnapshot(""); err != nil { + if err := s.outOfBandSnapshot(); err != nil { return fmt.Errorf("flatkv: SetInitialVersion: write seeded snapshot: %w", err) } } diff --git a/sei-db/state_db/sc/flatkv/store_replay_test.go b/sei-db/state_db/sc/flatkv/store_replay_test.go index deebb81575..09ee983a36 100644 --- a/sei-db/state_db/sc/flatkv/store_replay_test.go +++ b/sei-db/state_db/sc/flatkv/store_replay_test.go @@ -198,6 +198,10 @@ func TestReadOnlySurfacesReplayGap(t *testing.T) { commit(v, byte(v)) } + // CommitBlock offers snapshots to the writer without waiting, so wait here: the snapshots this test + // falls back to have to be on disk before the WAL is wiped. + require.NoError(t, s.FlushSnapshots()) + // Wipe the WAL and resume, so it no longer reaches back to the snapshot at version 2. resetWALForTest(t, s) commit(5, 0x99) diff --git a/sei-db/state_db/sc/flatkv/store_test.go b/sei-db/state_db/sc/flatkv/store_test.go index 062f48b885..826fefe215 100644 --- a/sei-db/state_db/sc/flatkv/store_test.go +++ b/sei-db/state_db/sc/flatkv/store_test.go @@ -446,15 +446,15 @@ func TestStoreRootHashStableAfterCommit(t *testing.T) { } // ============================================================================= -// Lifecycle (WriteSnapshot, Rollback) +// Lifecycle (outOfBandSnapshot, Rollback) // ============================================================================= -func TestStoreWriteSnapshotRequiresCommit(t *testing.T) { +func TestStoreOutOfBandSnapshotRequiresCommit(t *testing.T) { s := setupTestStore(t) defer s.Close() // Cannot snapshot at version 0 (nothing committed) - err := s.WriteSnapshot("") + err := s.outOfBandSnapshot() require.Error(t, err) require.Contains(t, err.Error(), "uncommitted") } @@ -534,7 +534,7 @@ func TestReadOnlyAtBeyondWALFails(t *testing.T) { commitStorageEntry(t, s1, ktype.Address{0x01}, ktype.Slot{0x01}, []byte{0x01}) commitStorageEntry(t, s1, ktype.Address{0x01}, ktype.Slot{0x02}, []byte{0x02}) - require.NoError(t, s1.WriteSnapshot("")) + require.NoError(t, s1.outOfBandSnapshot()) require.NoError(t, s1.Close()) cfg = config.DefaultTestConfig(t) @@ -563,7 +563,7 @@ func TestReopenReusesWorkingDir(t *testing.T) { require.NoError(t, err) commitStorageEntry(t, s, ktype.Address{0x01}, ktype.Slot{0x01}, []byte{0x01}) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) require.NoError(t, s.Close()) workDir := filepath.Join(dir, flatkvRootDir, workingDirName) @@ -600,7 +600,7 @@ func TestCatchupFromSpecificVersion(t *testing.T) { } hashAtV10 := rootHash(s1) - require.NoError(t, s1.WriteSnapshot("")) + require.NoError(t, s1.outOfBandSnapshot()) require.NoError(t, s1.Close()) cfg = config.DefaultTestConfig(t) @@ -826,7 +826,7 @@ func TestReadOnlyWriteGuards(t *testing.T) { require.ErrorIs(t, ro.ApplyChangeSets(ro.Version()+1, nil), errReadOnly) _, err = ro.Commit(ro.Version() + 1) require.ErrorIs(t, err, errReadOnly) - require.ErrorIs(t, ro.WriteSnapshot(""), errReadOnly) + require.ErrorIs(t, ro.(*CommitStore).outOfBandSnapshot(), errReadOnly) require.ErrorIs(t, ro.Rollback(1), errReadOnly) _, err = ro.(*CommitStore).Importer(1) require.ErrorIs(t, err, errReadOnly) @@ -1395,6 +1395,9 @@ func TestCrashRecoveryWALReplayLargeGap(t *testing.T) { require.NoError(t, err) } expectedHash := rootHash(s) + // Close discards whatever the snapshot writer still has queued, so wait for it here: the gap this + // test is about only exists once the snapshots are on disk. + require.NoError(t, s.FlushSnapshots()) require.NoError(t, s.Close()) // Reopen normally -- large WAL gap between snapshot and HEAD. @@ -1434,7 +1437,7 @@ func TestCrashRecoveryEmptyWALAfterSnapshot(t *testing.T) { _, err = s.Commit(s.Version() + 1) require.NoError(t, err) - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) expectedHash := rootHash(s) expectedVersion := s.Version() diff --git a/sei-db/state_db/sc/flatkv/store_version_probe_test.go b/sei-db/state_db/sc/flatkv/store_version_probe_test.go index 19d587c80e..5c032a75d3 100644 --- a/sei-db/state_db/sc/flatkv/store_version_probe_test.go +++ b/sei-db/state_db/sc/flatkv/store_version_probe_test.go @@ -114,7 +114,7 @@ func TestGetLatestVersionSnapshotBehindWAL(t *testing.T) { for i := int64(1); i <= 2; i++ { require.NoError(t, s.CommitBlock(i, []*proto.NamedChangeSet{bankPair([]byte("k"), []byte{byte(i)})})) } - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) for i := int64(3); i <= 5; i++ { require.NoError(t, s.CommitBlock(i, []*proto.NamedChangeSet{bankPair([]byte("k"), []byte{byte(i)})})) } @@ -132,7 +132,7 @@ func TestGetLatestVersionWALWipedAfterSnapshot(t *testing.T) { for i := int64(1); i <= 2; i++ { require.NoError(t, s.CommitBlock(i, []*proto.NamedChangeSet{bankPair([]byte("k"), []byte{byte(i)})})) } - require.NoError(t, s.WriteSnapshot("")) + require.NoError(t, s.outOfBandSnapshot()) resetWALForTest(t, s) require.Equal(t, int64(2), requireProbeMatchesOpen(t, s, cfg)) diff --git a/sei-db/state_db/sc/flatkv/store_write.go b/sei-db/state_db/sc/flatkv/store_write.go index 6a72605cba..5e6db5a0d9 100644 --- a/sei-db/state_db/sc/flatkv/store_write.go +++ b/sei-db/state_db/sc/flatkv/store_write.go @@ -114,13 +114,12 @@ func (s *CommitStore) Commit(version int64) (committed int64, err error) { // Step 4: Clear per-block bookkeeping s.clearPendingBlock() - // Periodic snapshot so WAL stays bounded and restarts are fast. A failure fails the commit: the - // flush wait inside WriteSnapshot is where a dead store surfaces, and a block whose data will never - // reach disk must not be reported as committed. The block is already durable in the WAL, so replay - // reconciles whatever the caller's halt leaves behind. - if s.config.SnapshotInterval > 0 && version%int64(s.config.SnapshotInterval) == 0 { - s.phaseTimer.SetPhase("commit_write_snapshot") - if err := s.WriteSnapshot(""); err != nil { + // Step 5: Offer the block to the snapshot writer, which decides whether it becomes a snapshot and, + // if so, writes it on its own goroutine. Periodic snapshots are what keep the WAL bounded and + // restarts fast. + if s.snapshotWriter != nil { + s.phaseTimer.SetPhase("commit_offer_snapshot") + if err := s.snapshotWriter.Offer(version, s.lastSealed); err != nil { return version, fmt.Errorf("auto snapshot at version %d: %w", version, err) } } @@ -139,6 +138,16 @@ func (s *CommitStore) Commit(version int64) (committed int64, err error) { return version, nil } +// FlushSnapshots blocks until no snapshot is being written. It is a synchronization point for callers +// that need the snapshot tree on disk to have caught up with the blocks committed so far; block +// commit does not need it. +func (s *CommitStore) FlushSnapshots() error { + if s.snapshotWriter == nil { + return nil + } + return s.snapshotWriter.Flush() +} + // clearPendingBlock resets the per-block bookkeeping that Commit consumed. func (s *CommitStore) clearPendingBlock() { s.pendingChangeSets = make([]*proto.NamedChangeSet, 0, len(s.pendingChangeSets)) 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 bf016b7e6c..7e5c9507be 100644 --- a/sei-db/state_db/sc/flatkv/store_write_test.go +++ b/sei-db/state_db/sc/flatkv/store_write_test.go @@ -7,7 +7,6 @@ import ( "fmt" "os" "testing" - "time" "github.com/stretchr/testify/require" @@ -582,10 +581,11 @@ func TestStoreFsyncConfig(t *testing.T) { // Auto-snapshot triggered by SnapshotInterval // ============================================================================= -// A failed periodic snapshot must fail the commit rather than being logged and discarded. The flush -// wait at the front of WriteSnapshot is where a dead store surfaces, so swallowing an error there would -// report a block as committed whose data will never reach disk — and the caller, which is required to -// halt on the first error, would never learn it had one. +// A failed periodic snapshot must fail a commit rather than being logged and discarded. The writer +// latches its first failure and reports it from every later call, so a checkpoint that failed with no +// caller to return to still stops the node. Swallowing it would report blocks as committed whose data +// will never reach disk, and the caller, which is required to halt on the first error, would never +// learn it had one. // // The failure is forced with directory permissions: the snapshot cannot create its temporary directory // under the flatkv root. The WAL and the databases live in subdirectories that already exist, so they @@ -616,8 +616,24 @@ func TestCommitFailsWhenPeriodicSnapshotFails(t *testing.T) { Changeset: proto.ChangeSet{Pairs: []*proto.KVPair{{Key: key, Value: make([]byte, 32)}}}, }})) + // The commit that trips the interval only hands the block to the writer, so it succeeds. _, err = s.Commit(s.Version() + 1) - require.Error(t, err, "a failed periodic snapshot must fail the commit") + require.NoError(t, err) + + // Waiting for the writer surfaces the failure it latched. + err = s.FlushSnapshots() + require.Error(t, err, "a failed snapshot must be reported, not swallowed") + require.ErrorContains(t, err, "create snapshot tmp dir", + "the error must name what actually failed") + + // And the node halts: every later commit reports the same failure. + key = keys.BuildEVMKey(keys.EVMKeyStorage, ktype.StorageKey(ktype.Address{0x03}, ktype.Slot{0x03})) + require.NoError(t, s.ApplyChangeSets(s.Version()+1, []*proto.NamedChangeSet{{ + Name: "evm", + Changeset: proto.ChangeSet{Pairs: []*proto.KVPair{{Key: key, Value: make([]byte, 32)}}}, + }})) + _, err = s.Commit(s.Version() + 1) + require.Error(t, err, "a bricked snapshot writer must fail every later commit") require.ErrorContains(t, err, "auto snapshot", "the error must name the snapshot as the cause rather than being swallowed") } @@ -1007,27 +1023,6 @@ func TestStoreFsyncEnabled(t *testing.T) { require.Equal(t, padLeft32(0x01), v) } -// ============================================================================= -// lastSnapshotTime is set after WriteSnapshot -// ============================================================================= - -func TestLastSnapshotTimeUpdated(t *testing.T) { - cfg := config.DefaultTestConfig(t) - s, err := newCommitStoreWithWAL(t.Context(), cfg) - require.NoError(t, err) - err = s.LoadLatest() - require.NoError(t, err) - defer s.Close() - - require.True(t, s.lastSnapshotTime.IsZero()) - - commitStorageEntry(t, s, ktype.Address{0x01}, ktype.Slot{0x01}, []byte{0x01}) - require.NoError(t, s.WriteSnapshot("")) - - require.False(t, s.lastSnapshotTime.IsZero()) - require.True(t, time.Since(s.lastSnapshotTime) < time.Second) -} - // ============================================================================= // WAL records all changesets // ============================================================================= diff --git a/sei-db/state_db/sc/flatkv/testutil_test.go b/sei-db/state_db/sc/flatkv/testutil_test.go index 07d2a5795a..6869b6e45f 100644 --- a/sei-db/state_db/sc/flatkv/testutil_test.go +++ b/sei-db/state_db/sc/flatkv/testutil_test.go @@ -164,11 +164,16 @@ func setupTestStoreWithConfig(t *testing.T, cfg *config.Config) *CommitStore { // without it a test that commits and then reads a database directly is looking at a disk that lags the // commit. It also matches how the Cosmos-era node drives the store, which forces a flush every block. // A test specifically about asynchronous flushing should call s.Commit directly instead. +// +// Snapshots are written off the execution thread for the same reason, so the wait covers them too: a +// test that commits past SnapshotInterval and then looks at the snapshot tree would otherwise be +// racing the writer. func commitAndCheck(t *testing.T, s *CommitStore) int64 { t.Helper() v, err := s.Commit(s.Version() + 1) require.NoError(t, err) requireFlushedToDisk(t, s) + require.NoError(t, s.FlushSnapshots()) return v } diff --git a/sei-db/tools/cmd/seidb/operations/flatkv_open_test.go b/sei-db/tools/cmd/seidb/operations/flatkv_open_test.go index ecd04238ee..c811377d84 100644 --- a/sei-db/tools/cmd/seidb/operations/flatkv_open_test.go +++ b/sei-db/tools/cmd/seidb/operations/flatkv_open_test.go @@ -148,7 +148,7 @@ func TestPrepareFlatKVToolingCloneRetriesENOENT(t *testing.T) { // so it shares the source snapshots' mounted filesystem even when dbDir is a // dedicated mount point. func TestPrepareFlatKVToolingClonePlacesTempDirInsideDBDir(t *testing.T) { - store, dbDir := newDiskBackedFlatKVStore(t) + store, dbDir := newDiskBackedFlatKVStore(t, 1) require.NoError(t, store.ApplyChangeSets(store.Version()+1, []*proto.NamedChangeSet{{ Name: keys.EVMStoreKey, Changeset: proto.ChangeSet{Pairs: []*proto.KVPair{ @@ -157,7 +157,7 @@ func TestPrepareFlatKVToolingClonePlacesTempDirInsideDBDir(t *testing.T) { }})) _, err := store.Commit(store.Version() + 1) require.NoError(t, err) - require.NoError(t, store.WriteSnapshot("")) + require.NoError(t, store.FlushSnapshots()) require.NoError(t, store.Close()) cloneDir, err := prepareFlatKVToolingClone(dbDir, 0) @@ -177,19 +177,11 @@ func TestPrepareFlatKVToolingClonePlacesTempDirInsideDBDir(t *testing.T) { // The cloned WAL would skip versions during catchup; the clone path must // detect the gap and surface it as a retryable errSourceChurning. func TestPrepareFlatKVToolingCloneDetectsWALTruncationRace(t *testing.T) { - store, dbDir := newDiskBackedFlatKVStore(t) + // Interval 5 lands exactly one snapshot, at v5: the next boundary is v10, which this test never + // reaches. The selected snapshot is therefore v5 for the whole test. + store, dbDir := newDiskBackedFlatKVStore(t, 5) - require.NoError(t, store.ApplyChangeSets(store.Version()+1, []*proto.NamedChangeSet{{ - Name: keys.EVMStoreKey, - Changeset: proto.ChangeSet{Pairs: []*proto.KVPair{ - noncePair(addrN(0xA1), 1), - }}, - }})) - _, err := store.Commit(store.Version() + 1) - require.NoError(t, err) - require.NoError(t, store.WriteSnapshot("")) - - for i := byte(2); i <= 5; i++ { + for i := byte(1); i <= 5; i++ { require.NoError(t, store.ApplyChangeSets(store.Version()+1, []*proto.NamedChangeSet{{ Name: keys.EVMStoreKey, Changeset: proto.ChangeSet{Pairs: []*proto.KVPair{ @@ -199,18 +191,19 @@ func TestPrepareFlatKVToolingCloneDetectsWALTruncationRace(t *testing.T) { _, err := store.Commit(store.Version() + 1) require.NoError(t, err) } + require.NoError(t, store.FlushSnapshots()) require.NoError(t, store.Close()) - // Simulate tryTruncateWAL having dropped v1..v3 from the WAL after a newer snapshot rolled — i.e. the - // WAL now starts at v4 while the selected snapshot is still v1. Rebuild the source changelog as a fresh - // state WAL that starts at v4 (the state WAL permits any first block number), which is the exact shape a - // front-truncation past the snapshot leaves behind. + // Simulate tryTruncateWAL having dropped everything up to v6 from the WAL after a newer snapshot + // rolled — i.e. the WAL now starts at v7 while the selected snapshot is still v5. Rebuild the source + // changelog as a fresh state WAL that starts at v7 (the state WAL permits any first block number), + // which is the exact shape a front-truncation past the snapshot leaves behind. walDir := filepath.Join(dbDir, "changelog") walCfg := statewal.DefaultConfig(walDir, "flatkv") require.NoError(t, os.RemoveAll(walDir)) // the store is closed above, so nothing holds the directory w, err := statewal.New(walCfg) require.NoError(t, err) - for v := uint64(4); v <= 5; v++ { + for v := uint64(7); v <= 8; v++ { require.NoError(t, w.Write(v, []*proto.NamedChangeSet{{ Name: keys.EVMStoreKey, Changeset: proto.ChangeSet{Pairs: []*proto.KVPair{noncePair(addrN(byte(v)), v)}}, @@ -227,7 +220,9 @@ func TestPrepareFlatKVToolingCloneDetectsWALTruncationRace(t *testing.T) { } func TestOpenFlatKVReadOnlyLatestAndHistoricalHeight(t *testing.T) { - store, dbDir := newDiskBackedFlatKVStore(t) + // Interval 1 snapshots both blocks, which is what opening at height 1 needs: a snapshot sitting on + // that height rather than one below it. + store, dbDir := newDiskBackedFlatKVStore(t, 1) addrA := addrN(0xA1) addrB := addrN(0xB2) @@ -239,7 +234,7 @@ func TestOpenFlatKVReadOnlyLatestAndHistoricalHeight(t *testing.T) { }})) _, err := store.Commit(store.Version() + 1) require.NoError(t, err) - require.NoError(t, store.WriteSnapshot("")) + require.NoError(t, store.FlushSnapshots()) require.NoError(t, store.ApplyChangeSets(store.Version()+1, []*proto.NamedChangeSet{{ Name: keys.EVMStoreKey, @@ -249,7 +244,7 @@ func TestOpenFlatKVReadOnlyLatestAndHistoricalHeight(t *testing.T) { }})) _, err = store.Commit(store.Version() + 1) require.NoError(t, err) - require.NoError(t, store.WriteSnapshot("")) + require.NoError(t, store.FlushSnapshots()) require.NoError(t, store.Close()) latest, err := openFlatKVReadOnly(dbDir, 0) @@ -270,7 +265,8 @@ func TestOpenFlatKVReadOnlyLatestAndHistoricalHeight(t *testing.T) { } func TestOpenFlatKVReadOnlyAfterSetInitialVersion(t *testing.T) { - store, dbDir := newDiskBackedFlatKVStore(t) + // No cadence snapshots: SetInitialVersion writes the one this test relies on. + store, dbDir := newDiskBackedFlatKVStore(t, 0) addr := addrN(0xC3) require.NoError(t, store.SetInitialVersion(100)) @@ -293,10 +289,18 @@ func TestOpenFlatKVReadOnlyAfterSetInitialVersion(t *testing.T) { require.NoError(t, latest.Close()) } -func newDiskBackedFlatKVStore(t *testing.T) (*flatkv.CommitStore, string) { +// newDiskBackedFlatKVStore opens a store whose snapshot cadence the caller chooses, so a test can place +// snapshots at the heights it needs by committing to a boundary and waiting with FlushSnapshots. An +// interval of 0 disables them. +// +// Retention is set well above anything these tests commit: they assert against specific snapshot +// heights, and count-based pruning would otherwise delete the one under test. +func newDiskBackedFlatKVStore(t *testing.T, snapshotInterval uint32) (*flatkv.CommitStore, string) { t.Helper() cfg := flatkvconfig.DefaultTestConfig(t) + cfg.SnapshotInterval = snapshotInterval + cfg.SnapshotKeepRecent = 100 stateWAL, err := flatkv.OpenStateWAL(cfg) require.NoError(t, err) store, err := flatkv.NewCommitStore(context.Background(), cfg, stateWAL)