diff --git a/sei-db/common/utils/path.go b/sei-db/common/utils/path.go index d98751d3b0..5148ec59c3 100644 --- a/sei-db/common/utils/path.go +++ b/sei-db/common/utils/path.go @@ -74,6 +74,13 @@ func GetReceiptStorePath(homePath string, backend string) string { return filepath.Join(homePath, "data", "ledger", "receipt", backend) } +// GetBlockStorePath returns the path for the ledger block store, which sits alongside the +// receipt store under data/ledger. There is no legacy fallback: the block store has only +// ever lived at this path. +func GetBlockStorePath(homePath string) string { + return filepath.Join(homePath, "data", "ledger", "block") +} + func GetChangelogPath(dbPath string) string { return filepath.Join(dbPath, "changelog") } diff --git a/sei-db/common/utils/path_test.go b/sei-db/common/utils/path_test.go index a5b6b009db..3a5a90657a 100644 --- a/sei-db/common/utils/path_test.go +++ b/sei-db/common/utils/path_test.go @@ -135,6 +135,14 @@ func TestGetReceiptStorePath_LegacyExists(t *testing.T) { assert.Equal(t, legacy, got) } +// --- GetBlockStorePath --- + +func TestGetBlockStorePath_NewNode(t *testing.T) { + home := t.TempDir() + got := GetBlockStorePath(home) + assert.Equal(t, filepath.Join(home, "data", "ledger", "block"), got) +} + // --- GetChangelogPath (unchanged, but verify) --- func TestGetChangelogPath(t *testing.T) { diff --git a/sei-db/config/giga_config.go b/sei-db/config/giga_config.go new file mode 100644 index 0000000000..84044e3689 --- /dev/null +++ b/sei-db/config/giga_config.go @@ -0,0 +1,67 @@ +package config + +import ( + "fmt" + + "github.com/sei-protocol/sei-chain/sei-db/common/utils" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/block/littblock" + "github.com/sei-protocol/sei-chain/sei-db/management/gc" + flatkvConfig "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/flatkv/config" +) + +// GigaStorageConfig is an in-process composition of store configs for Giga storage. +// It is intentionally not read from app.toml (no mapstructure tags, no TOML section): +// callers build it via DefaultGigaStorageConfig. Nested knobs live on the store and +// collector configs they already own (StateStoreConfig, ReceiptStoreConfig, +// gc.StorageGarbageCollectorConfig) rather than being redeclared here — in particular +// RollbackWindow has a single source of truth in +// gc.DefaultStorageGarbageCollectorConfig; per-store extras live on each +// PrunableStore via GetRetentionWindow. +type GigaStorageConfig struct { + HomePath string + FlatKVConfig *flatkvConfig.Config + SSConfig StateStoreConfig + ReceiptDBConfig ReceiptStoreConfig + BlockDBConfig *littblock.LittBlockConfig + PruningConfig *gc.StorageGarbageCollectorConfig +} + +// DefaultGigaStorageConfig returns a GigaStorageConfig whose store directories match the +// layout below, and whose pruning knobs are gc.DefaultStorageGarbageCollectorConfig(): +// +// data/state_commit/flatkv +// data/state_store/evm/{backend} (sole SS; no Cosmos SS in Giga) +// data/ledger/receipt/{backend} +// data/ledger/block +// +// Giga opens SS via evm.NewEVMStateStore(dir, ssConfig) directly — not through +// composite.NewCompositeStateStore — so the EVM path lives on EVMDBDirectory (the +// argument callers pass as dir). DBDirectory and EVMSplit are left at their defaults: +// they only matter for the composite path, which Giga does not use. +// +// Unlike the other Default*Config helpers in this package, this can fail: the block-store +// default wraps littdb.DefaultConfig, which validates the directory path. +func DefaultGigaStorageConfig(homePath string) (GigaStorageConfig, error) { + blockDBConfig, err := littblock.DefaultConfig(utils.GetBlockStorePath(homePath)) + if err != nil { + return GigaStorageConfig{}, fmt.Errorf("failed to build block db config: %w", err) + } + + flatKV := flatkvConfig.DefaultConfig() + flatKV.DataDir = utils.GetFlatKVPath(homePath) + + ssConfig := DefaultStateStoreConfig() + ssConfig.EVMDBDirectory = utils.GetEVMStateStorePath(homePath, ssConfig.Backend) + + receiptConfig := DefaultReceiptStoreConfig() + receiptConfig.DBDirectory = utils.GetReceiptStorePath(homePath, receiptConfig.Backend) + + return GigaStorageConfig{ + HomePath: homePath, + FlatKVConfig: flatKV, + SSConfig: ssConfig, + ReceiptDBConfig: receiptConfig, + BlockDBConfig: blockDBConfig, + PruningConfig: gc.DefaultStorageGarbageCollectorConfig(), + }, nil +} diff --git a/sei-db/config/giga_config_test.go b/sei-db/config/giga_config_test.go new file mode 100644 index 0000000000..e590c95762 --- /dev/null +++ b/sei-db/config/giga_config_test.go @@ -0,0 +1,34 @@ +package config + +import ( + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/sei-protocol/sei-chain/sei-db/common/utils" +) + +// TestDefaultGigaStorageConfigDirectories pins Giga's EVM-only SS layout: the path +// for evm.NewEVMStateStore lives on EVMDBDirectory. Giga does not use composite, so +// DBDirectory stays empty and EVMSplit stays at its default. +func TestDefaultGigaStorageConfigDirectories(t *testing.T) { + home := t.TempDir() + cfg, err := DefaultGigaStorageConfig(home) + require.NoError(t, err) + + require.Equal(t, home, cfg.HomePath) + require.False(t, cfg.SSConfig.EVMSplit) + require.Empty(t, cfg.SSConfig.DBDirectory) + require.Equal(t, utils.GetEVMStateStorePath(home, cfg.SSConfig.Backend), cfg.SSConfig.EVMDBDirectory) + + require.Equal(t, utils.GetFlatKVPath(home), cfg.FlatKVConfig.DataDir) + require.Equal(t, utils.GetReceiptStorePath(home, cfg.ReceiptDBConfig.Backend), cfg.ReceiptDBConfig.DBDirectory) + require.NotNil(t, cfg.BlockDBConfig.Litt) + require.Equal(t, []string{utils.GetBlockStorePath(home)}, cfg.BlockDBConfig.Litt.Paths) + + require.Equal(t, + filepath.Join(home, "data", "state_store", "evm", cfg.SSConfig.Backend), + cfg.SSConfig.EVMDBDirectory, + ) +} diff --git a/sei-db/ledger_db/event/placeholder.go b/sei-db/ledger_db/event/placeholder.go deleted file mode 100644 index 0e4b82e85c..0000000000 --- a/sei-db/ledger_db/event/placeholder.go +++ /dev/null @@ -1 +0,0 @@ -package event diff --git a/sei-db/ledger_db/transaction/placeholder.go b/sei-db/ledger_db/transaction/placeholder.go deleted file mode 100644 index 0619207533..0000000000 --- a/sei-db/ledger_db/transaction/placeholder.go +++ /dev/null @@ -1 +0,0 @@ -package transaction diff --git a/sei-db/management/gc/api.go b/sei-db/management/gc/api.go new file mode 100644 index 0000000000..86ea4a933a --- /dev/null +++ b/sei-db/management/gc/api.go @@ -0,0 +1,64 @@ +package gc + +// InfiniteRetentionWindow is the conventional GetRetentionWindow value meaning "never +// prune this store." Any negative value is treated the same: the store does not vote +// and is never passed to PruneBelow. +// +// Note: elsewhere in this repo (KeepRecent / MinRetainBlocks) 0 often means "keep +// forever." Here 0 means "no extra beyond the shared RollbackWindow," so infinite +// retention needs a negative sentinel. +const InfiniteRetentionWindow int64 = -1 + +// PrunableStore is a store whose old data may be dropped by the StorageGarbageCollector. +type PrunableStore interface { + // Name identifies the store (logging / errors). Duplicate names are allowed; + // the collector keys answers by index, not by name. + Name() string + + // PruneBelow may drop data for all blocks below blockNumber. + // The store may perform the deletion asynchronously. + PruneBelow(blockNumber uint64) error + + // GetRetentionWindow is how many extra blocks beyond the shared RollbackWindow + // this store needs to keep servable. Cut line (head = min non-zero GetLatestBlock): + // + // retention < 0 → store ignored (cutLine treated as 0; conventionally -1) + // retention >= 0 → cutLine = head - RollbackWindow - retention + // + // RollbackWindow is shared so every store stays consistent under rollback. + // Retention is optional history on top — kept so the store can still serve queries + // after a rollback that consumes the entire rollback window. + // + // Because the collector prunes every participating store to the *lowest* + // GetPruningBoundary, a large retention on one store effectively deepens pruning + // for the others (see StorageGarbageCollector). Read this as an input to that + // shared min, not as an independently applied per-store policy. + // + // Contract by store kind: + // - Contiguous (blockDB, receiptDB, state WAL): -1 / 0 / positive as configured. + // - SC: always 0 (only needs the shared rollback window for snapshots). + // - SS: always 0 (SS prunes its own version history via KeepRecent; the collector + // only drives SS snapshot pruning for the shared rollback window). + GetRetentionWindow() int64 + + // GetPruningBoundary returns the oldest block this store must keep to remain able to + // serve cutLine, or 0 to opt out of this cycle (not participate in calculating min prune height). + // + // Contiguous stores can restore to any height they hold → return cutLine. + // + // Snapshot stores can restore only at snapshot heights → return the newest completed + // snapshot ≤ cutLine, or the oldest completed snapshot if every snapshot is above + // cutLine (none can be dropped yet). No completed snapshot → 0. + // + // Assumption: snapshot creation finishes quickly enough that an in-flight write need + // not be reserved; modeling long-running snapshot writes is out of scope. + GetPruningBoundary(cutLine uint64) uint64 + + // GetLatestBlock returns the highest block this store has ingested. + // + // 0 means "nothing ingested yet" and is ignored when computing the global head, so a store + // still filling does not stall pruning for the rest of the fleet. It does not mean + // "disabled": a store disabled for this node is not instantiated and never reaches the + // collector. See StorageGarbageCollector for why skipping a 0 head is safe. + GetLatestBlock() (uint64, error) +} diff --git a/sei-db/management/gc/storage_garbage_collector.go b/sei-db/management/gc/storage_garbage_collector.go new file mode 100644 index 0000000000..74463dc738 --- /dev/null +++ b/sei-db/management/gc/storage_garbage_collector.go @@ -0,0 +1,225 @@ +package gc + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/sei-protocol/seilog" +) + +var logger = seilog.NewLogger("db", "gc") + +// StorageGarbageCollector periodically prunes a set of PrunableStores. +// +// All stores share config.RollbackWindow. Each store may request extra history via +// GetRetentionWindow (see that method). SC/SS return 0 — the collector only drives +// their snapshot pruning for the shared window; SS version-history pruning is separate. +// +// Each cycle: +// 1. head = min non-zero GetLatestBlock across stores +// 2. per store: cutLine = head - RollbackWindow - GetRetentionWindow +// (skipped when cutLine == 0: infinite retention, or head still inside the window) +// 3. ask GetPruningBoundary(cutLine); 0 = opt out for this cycle +// 4. pruneHeight = min of non-zero answers; PruneBelow(pruneHeight) on every store +// that answered non-zero +// +// Step 4 uses the shared min (not each store's own boundary) so a retained snapshot +// stays restorable: contiguous stores must still hold the blocks that follow it. +// A side effect is that one store's deep GetRetentionWindow is imposed on every +// participating store — e.g. receiptDB retention 100_000 also keeps SC/SS snapshots +// that far back even though those stores report retention 0. That uniform prune is +// an intentional tradeoff for Giga (one effective retention across stores), not +// independent per-store retention. +// +// PruneBelow failures are joined and do not skip later stores — permission-to-drop +// is not transactional, and one unhealthy store must not block pruning of others. +// +// Precondition on stores: every store passed in is live. A store disabled for this node is +// not instantiated and so never reaches the collector, so the set holds no permanently-empty +// member. A store that is present but has ingested nothing reports head 0; that head is +// ignored when computing the global head and the store opts out via GetPruningBoundary, so +// one store still filling does not stall pruning fleet-wide. +// +// That skip is only safe while no store bootstraps from another store's history: today each +// replays its own changelog and owns that changelog's retention. If a store is ever fed from +// another store's WAL, it could sit at head 0 while that WAL is pruned past what it needs, +// and this rule has to change (distinguish "empty and still bootstrapping" from "not +// participating" rather than skipping both). +// +// Invariant: if RollbackWindow blocks of rollback were possible before a prune, that +// prune does not take the ability away. Full headroom is not promised after a rollback +// has already consumed part of the window. +// +// The type is a ticker around prune; decision logic lives in prune for unit testing. +// Not safe for concurrent use (Close must be called exactly once). +type StorageGarbageCollector struct { + config *StorageGarbageCollectorConfig + stores []PrunableStore + ctx context.Context + stopCh chan struct{} + wg sync.WaitGroup +} + +func NewStorageGarbageCollector( + ctx context.Context, + config *StorageGarbageCollectorConfig, + stores []PrunableStore, +) (*StorageGarbageCollector, error) { + if err := config.Validate(); err != nil { + return nil, fmt.Errorf("invalid storage garbage collector config: %w", err) + } + + s := &StorageGarbageCollector{ + config: config, + stores: stores, + ctx: ctx, + stopCh: make(chan struct{}), + } + s.wg.Add(1) + go s.run() + return s, nil +} + +// Close stops the run loop and waits for it to exit. Must be called exactly once. +func (s *StorageGarbageCollector) Close() error { + close(s.stopCh) + s.wg.Wait() + return nil +} + +func (s *StorageGarbageCollector) run() { + defer s.wg.Done() + + ticker := time.NewTicker(s.config.PruneInterval) + defer ticker.Stop() + + for { + select { + case <-s.stopCh: + return + case <-s.ctx.Done(): + return + case <-ticker.C: + if err := prune(s.config, s.stores); err != nil { + logger.Error("prune cycle failed", "err", err) + } + } + } +} + +// prune runs one prune cycle. See StorageGarbageCollector for the decision rules. +func prune(config *StorageGarbageCollectorConfig, stores []PrunableStore) error { + if len(stores) == 0 { + return nil + } + + globalLatestBlock, err := getGlobalLatestBlock(stores) + if err != nil { + return err + } + if globalLatestBlock == 0 { + logger.Info("skipping pruning: no store has a latest block") + return nil + } + + // Boundaries are positional so duplicate Name() values cannot mis-attribute an opt-out. + pruningBoundaries := make([]uint64, len(stores)) + var pruneHeight uint64 + for i, store := range stores { + retention := store.GetRetentionWindow() + cutLine := getCutLine(globalLatestBlock, config.RollbackWindow, retention) + if cutLine == 0 { + // Infinite retention, or head still inside this store's retain window. + continue + } + + pruningBoundaries[i] = store.GetPruningBoundary(cutLine) + if pruningBoundaries[i] == 0 { + // Opt-out (e.g. no completed snapshot yet). + continue + } + if pruneHeight == 0 || pruningBoundaries[i] < pruneHeight { + pruneHeight = pruningBoundaries[i] + } + } + if pruneHeight == 0 { + logger.Info("skipping pruning: no store reported a pruning boundary", + "globalLatestBlock", globalLatestBlock, + "pruningBoundaryByStore", describeAnswers(stores, pruningBoundaries), + ) + return nil + } + + logger.Info("pruning stores", + "globalLatestBlock", globalLatestBlock, + "pruneHeight", pruneHeight, + "pruningBoundaryByStore", describeAnswers(stores, pruningBoundaries), + ) + var pruneErrs error + for i, store := range stores { + if pruningBoundaries[i] == 0 { + continue + } + if err := store.PruneBelow(pruneHeight); err != nil { + pruneErrs = errors.Join(pruneErrs, fmt.Errorf("failed to prune %s below %d: %w", store.Name(), pruneHeight, err)) + } + } + return pruneErrs +} + +func describeAnswers(stores []PrunableStore, pruningBoundaries []uint64) string { + var sb strings.Builder + for i, store := range stores { + if i > 0 { + sb.WriteString(" ") + } + fmt.Fprintf(&sb, "%s=%d", store.Name(), pruningBoundaries[i]) + } + return sb.String() +} + +// getCutLine returns head - RollbackWindow - retention for retention >= 0. +// Returns 0 when retention < 0, when the combined window overflows uint64, or when +// head is still inside that window (unsigned subtraction must not wrap). +func getCutLine(globalLatestBlock uint64, rollbackWindow uint64, retention int64) uint64 { + if retention < 0 { + return 0 + } + totalRetainWindow := rollbackWindow + uint64(retention) + if totalRetainWindow < rollbackWindow { + // Addition wrapped; treat as an unsatisfiable retain window (skip pruning). + return 0 + } + if globalLatestBlock <= totalRetainWindow { + return 0 + } + return globalLatestBlock - totalRetainWindow +} + +// getGlobalLatestBlock returns the smallest non-zero GetLatestBlock among stores. +// Using the min keeps a lagging store from being pruned past blocks it still needs. +// Returns 0 when no store has data. +// +// Heads of 0 are skipped rather than treated as "unknown". Note this excludes from the min +// exactly the store that holds nothing, so the min-head argument does not protect that +// store; what protects it is the store-set precondition on StorageGarbageCollector. +func getGlobalLatestBlock(stores []PrunableStore) (uint64, error) { + var blockNum uint64 + for _, store := range stores { + storeHeight, err := store.GetLatestBlock() + if err != nil { + return 0, fmt.Errorf("failed to read latest block from %s: %w", store.Name(), err) + } + if storeHeight == 0 { + continue + } + if blockNum == 0 || storeHeight < blockNum { + blockNum = storeHeight + } + } + return blockNum, nil +} diff --git a/sei-db/management/gc/storage_garbage_collector_config.go b/sei-db/management/gc/storage_garbage_collector_config.go new file mode 100644 index 0000000000..eb0a2bac51 --- /dev/null +++ b/sei-db/management/gc/storage_garbage_collector_config.go @@ -0,0 +1,46 @@ +package gc + +import ( + "fmt" + "time" +) + +// StorageGarbageCollectorConfig configures a StorageGarbageCollector. +type StorageGarbageCollectorConfig struct { + // RollbackWindow is how many blocks behind head the system must remain able to + // roll back to. Shared by every managed store so they stay mutually consistent; + // per-store extras beyond this window use PrunableStore.GetRetentionWindow. + // + // 0 is allowed: cutLine then equals head for any store with retention 0. That is + // safe only when every participating store still leaves enough history on its own + // (e.g. positive GetRetentionWindow, or snapshot votes that keep contiguous stores + // above genesis). With retention-0 contiguous stores and a snapshot store that + // answers 0 (no snapshot yet), pruneHeight can reach head and drop everything + // below it — a snapshot landing near head then cannot be replayed forward. + // + // After a rollback that consumes part of the window, full headroom is not promised + // (e.g. window 10_000, roll back 5_000 → only ~5_000 of headroom remain). + RollbackWindow uint64 + + // PruneInterval is how often the collector runs a prune cycle. Must be > 0. + PruneInterval time.Duration +} + +// DefaultStorageGarbageCollectorConfig returns the default collector config. +func DefaultStorageGarbageCollectorConfig() *StorageGarbageCollectorConfig { + return &StorageGarbageCollectorConfig{ + RollbackWindow: 1000, + PruneInterval: 10 * time.Minute, + } +} + +// Validate checks that required fields are set to usable values. +func (c *StorageGarbageCollectorConfig) Validate() error { + if c == nil { + return fmt.Errorf("config is required") + } + if c.PruneInterval <= 0 { + return fmt.Errorf("prune interval must be greater than 0") + } + return nil +} diff --git a/sei-db/management/gc/storage_garbage_collector_test.go b/sei-db/management/gc/storage_garbage_collector_test.go new file mode 100644 index 0000000000..8bfe42d327 --- /dev/null +++ b/sei-db/management/gc/storage_garbage_collector_test.go @@ -0,0 +1,593 @@ +package gc + +import ( + "context" + "errors" + "math" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// mockStore is a PrunableStore with canned answers. Build via snapshotStore or contiguousStore. +type mockStore struct { + name string + latestHeight uint64 + // retentionWindow is returned by GetRetentionWindow (0 = shared RollbackWindow only). + retentionWindow int64 + pruningBoundary func(cutLine uint64) uint64 + getErr error + pruneErr error + + pruneBelowCalled atomic.Bool + prunedBelow atomic.Uint64 + pruneBelowCalls atomic.Uint64 +} + +// snapshotStore models SC/SS (retention 0). GetPruningBoundary returns the newest snapshot +// ≤ cutLine, or the oldest snapshot if all are above cutLine. Empty snapshots → 0. +// Snapshots must be ascending. +func snapshotStore(name string, latestHeight uint64, snapshots ...uint64) *mockStore { + return &mockStore{ + name: name, + latestHeight: latestHeight, + retentionWindow: 0, + pruningBoundary: func(cutLine uint64) uint64 { + if len(snapshots) == 0 { + return 0 + } + oldest := snapshots[0] + for _, snapshot := range snapshots { + if snapshot > cutLine { + break + } + oldest = snapshot + } + return oldest + }, + } +} + +// contiguousStore models blockDB / receiptDB / WAL (retention 0 by default). +// hasData=false → GetPruningBoundary returns 0 (opt out). Otherwise returns cutLine. +// Use withRetentionWindow for extras or InfiniteRetentionWindow. +func contiguousStore(name string, latestHeight uint64, hasData bool) *mockStore { + return &mockStore{ + name: name, + latestHeight: latestHeight, + retentionWindow: 0, + pruningBoundary: func(cutLine uint64) uint64 { + if !hasData { + return 0 + } + return cutLine + }, + } +} + +func withRetentionWindow(store *mockStore, retention int64) *mockStore { + store.retentionWindow = retention + return store +} + +func (m *mockStore) Name() string { + return m.name +} + +func (m *mockStore) GetRetentionWindow() int64 { + return m.retentionWindow +} + +func (m *mockStore) GetLatestBlock() (uint64, error) { + return m.latestHeight, m.getErr +} + +func (m *mockStore) GetPruningBoundary(cutLine uint64) uint64 { + return m.pruningBoundary(cutLine) +} + +func (m *mockStore) PruneBelow(blockNumber uint64) error { + m.pruneBelowCalled.Store(true) + m.prunedBelow.Store(blockNumber) + m.pruneBelowCalls.Add(1) + return m.pruneErr +} + +func prunableStores(list ...*mockStore) []PrunableStore { + result := make([]PrunableStore, len(list)) + for i, store := range list { + result[i] = store + } + return result +} + +func testConfig(t *testing.T, rollbackWindow uint64) *StorageGarbageCollectorConfig { + t.Helper() + config := &StorageGarbageCollectorConfig{ + RollbackWindow: rollbackWindow, + PruneInterval: time.Minute, + } + require.NoError(t, config.Validate()) + return config +} + +// TestPruneDecisions covers one prune cycle. wantPruneBelow == nil means no store is pruned. +// +// wantPruned is hardcoded per store rather than derived from getCutLine / GetPruningBoundary: deriving it from the +// helpers under test would let a sign error or off-by-one shift the expectation in lockstep with the bug. +func TestPruneDecisions(t *testing.T) { + cases := []struct { + name string + rollbackWindow uint64 + stores []*mockStore + wantPruneBelow *uint64 + wantPruned []bool + }{ + { + name: "SC and WAL both retention 0: min boundary wins", + rollbackWindow: 10_000, + stores: []*mockStore{ + snapshotStore("sc", 100_000, 80_000, 85_000, 92_000), + contiguousStore("stateWAL", 100_000, true), + }, + // cutLine 90_000; sc keeps snapshot 85_000; WAL answers 90_000 → min 85_000. + wantPruneBelow: ptr(85_000), + wantPruned: []bool{true, true}, + }, + { + name: "lowest boundary across many stores wins", + rollbackWindow: 10_000, + stores: []*mockStore{ + snapshotStore("sc", 100_000, 80_000, 85_000, 92_000), + snapshotStore("ss", 100_000, 70_000, 88_000, 95_000), + snapshotStore("flatKV", 100_000, 50_000, 90_000, 99_000), + contiguousStore("stateWAL", 100_000, true), + }, + // sc 85_000, ss 88_000, flatKV 90_000, WAL 90_000 → min 85_000. + wantPruneBelow: ptr(85_000), + wantPruned: []bool{true, true, true, true}, + }, + { + name: "RollbackWindow of 1 still leaves one block of margin", + rollbackWindow: 1, + stores: []*mockStore{ + snapshotStore("sc", 100_000, 80_000, 99_999), + contiguousStore("stateWAL", 100_000, true), + }, + wantPruneBelow: ptr(99_999), + wantPruned: []bool{true, true}, + }, + { + name: "RollbackWindow 0: cutLine equals head", + rollbackWindow: 0, + stores: []*mockStore{ + snapshotStore("sc", 100_000, 80_000, 90_000), + contiguousStore("stateWAL", 100_000, true), + }, + // cutLine 100_000; sc 90_000; WAL 100_000 → min 90_000. + wantPruneBelow: ptr(90_000), + wantPruned: []bool{true, true}, + }, + { + name: "RollbackWindow 0 with no snapshot vote prunes contiguous to head", + rollbackWindow: 0, + stores: []*mockStore{ + snapshotStore("sc", 100_000), + contiguousStore("stateWAL", 100_000, true), + }, + wantPruneBelow: ptr(100_000), + wantPruned: []bool{false, true}, + }, + { + name: "positive contiguous retention deepens that store's cut line", + rollbackWindow: 10_000, + stores: []*mockStore{ + snapshotStore("sc", 100_000, 80_000, 90_000), + withRetentionWindow(contiguousStore("stateWAL", 100_000, true), 5_000), + }, + // sc cutLine 90_000 → 90_000; WAL cutLine 85_000 → 85_000 → min 85_000. + wantPruneBelow: ptr(85_000), + wantPruned: []bool{true, true}, + }, + { + name: "SS retention 0 behaves like SC", + rollbackWindow: 10_000, + stores: []*mockStore{ + snapshotStore("ss", 100_000, 80_000, 90_000), + contiguousStore("stateWAL", 100_000, true), + }, + wantPruneBelow: ptr(90_000), + wantPruned: []bool{true, true}, + }, + { + name: "infinite retention on every store skips pruning", + rollbackWindow: 10_000, + stores: []*mockStore{ + withRetentionWindow(contiguousStore("blockDB", 100_000, true), InfiniteRetentionWindow), + withRetentionWindow(contiguousStore("receiptDB", 100_000, true), InfiniteRetentionWindow), + }, + wantPruneBelow: nil, + wantPruned: []bool{false, false}, + }, + { + name: "infinite retention on one store leaves others free to prune", + rollbackWindow: 10_000, + stores: []*mockStore{ + withRetentionWindow(contiguousStore("archiveWAL", 100_000, true), InfiniteRetentionWindow), + snapshotStore("sc", 100_000, 80_000), + contiguousStore("stateWAL", 100_000, true), + }, + // archiveWAL skipped; sc 80_000; stateWAL 90_000 → min 80_000. + wantPruneBelow: ptr(80_000), + wantPruned: []bool{false, true, true}, + }, + { + name: "snapshot exactly at the cut line is kept", + rollbackWindow: 10_000, + stores: []*mockStore{ + snapshotStore("sc", 100_000, 50_000, 90_000), + contiguousStore("stateWAL", 100_000, true), + }, + wantPruneBelow: ptr(90_000), + wantPruned: []bool{true, true}, + }, + { + name: "all snapshots above cut line: store still votes its oldest", + rollbackWindow: 10_000, + stores: []*mockStore{ + snapshotStore("sc", 100_000, 95_000, 97_000), + contiguousStore("stateWAL", 100_000, true), + }, + // sc answers 95_000; WAL answers 90_000 → shared prune 90_000 (sc snapshots untouched). + wantPruneBelow: ptr(90_000), + wantPruned: []bool{true, true}, + }, + { + name: "lagging store lowers the global head", + rollbackWindow: 10_000, + stores: []*mockStore{ + snapshotStore("lagging", 50_000, 30_000, 50_000), + contiguousStore("stateWAL", 100_000, true), + }, + // head 50_000 → cutLine 40_000; lagging answers 30_000; WAL 40_000 → min 30_000. + wantPruneBelow: ptr(30_000), + wantPruned: []bool{true, true}, + }, + { + name: "store answering 0 is ignored", + rollbackWindow: 10_000, + stores: []*mockStore{ + contiguousStore("optOut", 100_000, false), + snapshotStore("sc", 100_000, 80_000), + contiguousStore("stateWAL", 100_000, true), + }, + wantPruneBelow: ptr(80_000), + wantPruned: []bool{false, true, true}, + }, + { + name: "store with no snapshot yet is ignored", + rollbackWindow: 10_000, + stores: []*mockStore{ + snapshotStore("sc", 100_000), + contiguousStore("stateWAL", 100_000, true), + }, + wantPruneBelow: ptr(90_000), + wantPruned: []bool{false, true}, + }, + { + name: "zero head ignored for global head; store still votes", + rollbackWindow: 10_000, + stores: []*mockStore{ + snapshotStore("stalled", 0, 50_000), + contiguousStore("stateWAL", 100_000, true), + }, + // head from WAL 100_000; stalled still answers 50_000 → min 50_000. + wantPruneBelow: ptr(50_000), + wantPruned: []bool{true, true}, + }, + { + name: "head inside retain window: no prune", + rollbackWindow: 10_000, + stores: []*mockStore{ + snapshotStore("sc", 5_000, 1_000, 2_000), + contiguousStore("stateWAL", 5_000, true), + }, + wantPruneBelow: nil, + wantPruned: []bool{false, false}, + }, + { + name: "head inside one store's window skips only that store", + rollbackWindow: 60_000, + stores: []*mockStore{ + snapshotStore("sc", 100_000, 500, 1_000), + withRetentionWindow(contiguousStore("stateWAL", 100_000, true), 40_000), + }, + // WAL cutLine 0 (skipped); sc cutLine 40_000 → 1_000. + wantPruneBelow: ptr(1_000), + wantPruned: []bool{true, false}, + }, + { + name: "no store has a latest block", + rollbackWindow: 10_000, + stores: []*mockStore{ + snapshotStore("sc", 0), + contiguousStore("stateWAL", 0, false), + }, + wantPruneBelow: nil, + wantPruned: []bool{false, false}, + }, + { + name: "only an opt-out store has data", + rollbackWindow: 10_000, + stores: []*mockStore{ + contiguousStore("optOut", 100_000, false), + }, + wantPruneBelow: nil, + wantPruned: []bool{false}, + }, + { + name: "no stores at all", + rollbackWindow: 10_000, + stores: nil, + wantPruneBelow: nil, + wantPruned: nil, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Len(t, tc.wantPruned, len(tc.stores), "wantPruned must have one entry per store") + + require.NoError(t, prune(testConfig(t, tc.rollbackWindow), prunableStores(tc.stores...))) + + for i, store := range tc.stores { + if !tc.wantPruned[i] { + require.Falsef(t, store.pruneBelowCalled.Load(), "%s should not be pruned", store.name) + continue + } + require.NotNil(t, tc.wantPruneBelow, "wantPruned expects a prune height") + require.Truef(t, store.pruneBelowCalled.Load(), "%s should be pruned", store.name) + require.Equalf(t, *tc.wantPruneBelow, store.prunedBelow.Load(), "%s prune height", store.name) + } + }) + } +} + +func TestPruneOnYoungChainWithDeepContiguousRetention(t *testing.T) { + // head 100 << WAL retain window (rollback 1_000 + retention 100_000) → both cutLines 0. + sc := snapshotStore("sc", 100, 50, 100) + wal := withRetentionWindow(contiguousStore("stateWAL", 100, true), 100_000) + + require.NoError(t, prune(testConfig(t, 1_000), prunableStores(sc, wal))) + + require.False(t, sc.pruneBelowCalled.Load()) + require.False(t, wal.pruneBelowCalled.Load()) +} + +func TestPruneKeepsOptedOutStoreWithDuplicateName(t *testing.T) { + // Answers are keyed by index, so a duplicate Name() must not prune the opt-out store. + optOut := contiguousStore("ss", 100_000, false) + participating := snapshotStore("ss", 100_000, 80_000) + wal := contiguousStore("stateWAL", 100_000, true) + + require.NoError(t, prune(testConfig(t, 10_000), prunableStores(optOut, participating, wal))) + + require.False(t, optOut.pruneBelowCalled.Load(), "store answering 0 must not be pruned") + require.True(t, participating.pruneBelowCalled.Load()) + require.Equal(t, uint64(80_000), participating.prunedBelow.Load()) + require.Equal(t, uint64(80_000), wal.prunedBelow.Load()) +} + +func TestPruneGetLatestBlockError(t *testing.T) { + sentinel := errors.New("boom") + sc := snapshotStore("sc", 100_000, 80_000) + broken := contiguousStore("brokenStore", 100_000, true) + broken.getErr = sentinel + + err := prune(testConfig(t, 10_000), prunableStores(sc, broken)) + require.ErrorIs(t, err, sentinel) + require.ErrorContains(t, err, "brokenStore") + require.False(t, sc.pruneBelowCalled.Load()) + require.False(t, broken.pruneBelowCalled.Load()) +} + +func TestPruneBelowErrorContinuesRemainingStores(t *testing.T) { + firstErr := errors.New("boom1") + secondErr := errors.New("boom2") + first := snapshotStore("first", 100_000, 80_000) + first.pruneErr = firstErr + second := snapshotStore("second", 100_000, 80_000) + second.pruneErr = secondErr + wal := contiguousStore("stateWAL", 100_000, true) + + err := prune(testConfig(t, 10_000), prunableStores(first, second, wal)) + require.ErrorIs(t, err, firstErr) + require.ErrorIs(t, err, secondErr) + require.ErrorContains(t, err, "first") + require.ErrorContains(t, err, "second") + require.ErrorContains(t, err, "80000") + require.True(t, first.pruneBelowCalled.Load()) + require.True(t, second.pruneBelowCalled.Load()) + require.True(t, wal.pruneBelowCalled.Load()) + require.Equal(t, uint64(80_000), wal.prunedBelow.Load()) +} + +func TestGetCutLine(t *testing.T) { + cases := []struct { + name string + head uint64 + rollbackWindow uint64 + retention int64 + want uint64 + }{ + {name: "rollback window only", head: 100_000, rollbackWindow: 10_000, want: 90_000}, + {name: "zero rollback window", head: 100_000, rollbackWindow: 0, want: 100_000}, + {name: "zero rollback with retention", head: 100_000, rollbackWindow: 0, retention: 10_000, want: 90_000}, + {name: "retention adds to rollback window", head: 100_000, rollbackWindow: 1, retention: 10_000, want: 89_999}, + {name: "the two windows add", head: 100_000, rollbackWindow: 10_000, retention: 5_000, want: 85_000}, + {name: "zero retention", head: 100_000, rollbackWindow: 10_000, want: 90_000}, + {name: "infinite retention", head: 100_000, rollbackWindow: 10_000, retention: InfiniteRetentionWindow, want: 0}, + {name: "any negative retention is infinite", head: 100_000, rollbackWindow: 10_000, retention: -99, want: 0}, + {name: "head one above the window", head: 10_001, rollbackWindow: 10_000, want: 1}, + {name: "head exactly at the window", head: 10_000, rollbackWindow: 10_000, want: 0}, + {name: "head one below the window", head: 9_999, rollbackWindow: 10_000, want: 0}, + {name: "head far below the window", head: 100, rollbackWindow: 1_000, retention: 100_000, want: 0}, + {name: "head at genesis", head: 0, rollbackWindow: 10_000, want: 0}, + {name: "rollback plus retention overflows uint64", head: math.MaxUint64, rollbackWindow: math.MaxUint64, retention: 1, want: 0}, + {name: "max rollback alone does not overflow", head: math.MaxUint64, rollbackWindow: math.MaxUint64, want: 0}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, getCutLine(tc.head, tc.rollbackWindow, tc.retention)) + }) + } +} + +func TestGetGlobalLatestBlock(t *testing.T) { + storesWithHeads := func(heads ...uint64) []PrunableStore { + list := make([]*mockStore, len(heads)) + for i, head := range heads { + list[i] = contiguousStore("store", head, true) + } + return prunableStores(list...) + } + + cases := []struct { + name string + heads []uint64 + want uint64 + }{ + {name: "single store", heads: []uint64{100}, want: 100}, + {name: "all agree", heads: []uint64{100, 100, 100}, want: 100}, + {name: "smallest first", heads: []uint64{80, 100}, want: 80}, + {name: "smallest last", heads: []uint64{100, 80}, want: 80}, + {name: "smallest in the middle", heads: []uint64{100, 50, 90}, want: 50}, + {name: "leading zero ignored", heads: []uint64{0, 100}, want: 100}, + {name: "trailing zero ignored", heads: []uint64{100, 0}, want: 100}, + {name: "zero among many ignored", heads: []uint64{100, 0, 80}, want: 80}, + {name: "every store reports zero", heads: []uint64{0, 0}, want: 0}, + {name: "no stores", heads: nil, want: 0}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + head, err := getGlobalLatestBlock(storesWithHeads(tc.heads...)) + require.NoError(t, err) + require.Equal(t, tc.want, head) + }) + } +} + +func TestDefaultStorageGarbageCollectorConfig(t *testing.T) { + cfg := DefaultStorageGarbageCollectorConfig() + require.Equal(t, uint64(1_000), cfg.RollbackWindow) + require.Equal(t, 10*time.Minute, cfg.PruneInterval) + require.NoError(t, cfg.Validate()) +} + +func TestValidate(t *testing.T) { + require.ErrorContains(t, (*StorageGarbageCollectorConfig)(nil).Validate(), "config is required") + + require.NoError(t, (&StorageGarbageCollectorConfig{ + RollbackWindow: 0, + PruneInterval: time.Minute, + }).Validate()) + + require.NoError(t, (&StorageGarbageCollectorConfig{ + RollbackWindow: 1, + PruneInterval: time.Minute, + }).Validate()) + + require.ErrorContains(t, (&StorageGarbageCollectorConfig{ + RollbackWindow: 1, + PruneInterval: 0, + }).Validate(), "prune interval") + require.ErrorContains(t, (&StorageGarbageCollectorConfig{ + RollbackWindow: 1, + PruneInterval: -time.Second, + }).Validate(), "prune interval") +} + +func TestNewStorageGarbageCollectorInvalidConfig(t *testing.T) { + sm, err := NewStorageGarbageCollector( + context.Background(), + &StorageGarbageCollectorConfig{PruneInterval: 0}, + prunableStores(snapshotStore("sc", 100)), + ) + require.Error(t, err) + require.Nil(t, sm) +} + +func TestNewStorageGarbageCollectorNilConfig(t *testing.T) { + sm, err := NewStorageGarbageCollector(context.Background(), nil, nil) + require.ErrorContains(t, err, "config is required") + require.Nil(t, sm) +} + +func TestNewStorageGarbageCollectorConstructAndClose(t *testing.T) { + sm, err := NewStorageGarbageCollector( + context.Background(), + DefaultStorageGarbageCollectorConfig(), + prunableStores(snapshotStore("sc", 100, 100), contiguousStore("stateWAL", 100, true)), + ) + require.NoError(t, err) + require.NotNil(t, sm) + + require.NoError(t, sm.Close()) +} + +func TestCloseAfterContextCancelled(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + sm, err := NewStorageGarbageCollector(ctx, DefaultStorageGarbageCollectorConfig(), nil) + require.NoError(t, err) + + cancel() + require.NoError(t, sm.Close()) +} + +// startCollector runs a real collector on a short PruneInterval so the ticker branch in run() is exercised. +func startCollector(t *testing.T, interval time.Duration, stores ...*mockStore) { + t.Helper() + + config := &StorageGarbageCollectorConfig{RollbackWindow: 10_000, PruneInterval: interval} + require.NoError(t, config.Validate()) + + sm, err := NewStorageGarbageCollector(context.Background(), config, prunableStores(stores...)) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, sm.Close()) }) +} + +// TestRunTickerDrivesPruneCycles covers run()'s ticker branch: a cycle fires on its own and prunes to the expected +// height. +func TestRunTickerDrivesPruneCycles(t *testing.T) { + sc := snapshotStore("sc", 100_000, 80_000) + wal := contiguousStore("stateWAL", 100_000, true) + + startCollector(t, 10*time.Millisecond, sc, wal) + + require.Eventually(t, func() bool { + return sc.pruneBelowCalled.Load() && wal.pruneBelowCalled.Load() + }, 2*time.Second, 5*time.Millisecond, "ticker should drive a prune cycle") + require.Equal(t, uint64(80_000), sc.prunedBelow.Load()) + require.Equal(t, uint64(80_000), wal.prunedBelow.Load()) +} + +// TestRunSurvivesPruneError covers run()'s logger.Error branch: a PruneBelow failure is logged and the loop keeps +// ticking rather than exiting after the first error. +func TestRunSurvivesPruneError(t *testing.T) { + broken := snapshotStore("broken", 100_000, 80_000) + broken.pruneErr = errors.New("boom") + + startCollector(t, 10*time.Millisecond, broken) + + require.Eventually(t, func() bool { + return broken.pruneBelowCalls.Load() > 1 + }, 2*time.Second, 5*time.Millisecond, "a failed prune must not kill the run loop") +} + +func ptr(v uint64) *uint64 { + return &v +} diff --git a/sei-db/management/storage_garbage_collector.go b/sei-db/management/storage_garbage_collector.go deleted file mode 100644 index 20ad25f2e2..0000000000 --- a/sei-db/management/storage_garbage_collector.go +++ /dev/null @@ -1,223 +0,0 @@ -package management - -import ( - "context" - "fmt" - "sync" - "time" - - "github.com/sei-protocol/seilog" -) - -var logger = seilog.NewLogger("db", "management") - -// StorageGarbageCollector manages deletion of stored data across a set of state stores and the StateWAL. -// -// The StorageGarbageCollector performs state deletion while maintaining the following invariant: -// "If it's possible to roll back at least config.RollbackWindow blocks, then any state deletion operation -// will retain the system's ability to roll back at least config.RollbackWindow blocks." That is to say, -// while the StorageGarbageCollector cannot unilaterally guarantee the ability to roll back config.RollbackWindow -// blocks by itself, it guarantees that the act of deleting old state does not prevent rollback within this -// rollback window. -// -// StorageGarbageCollector is not thread safe. -type StorageGarbageCollector struct { - - // Configuration for this storage garbage collector. - config *StorageGarbageCollectorConfig - - // The stores whose data can be rebuilt by replaying the state WAL. - stateStores []SnapshotStore - - // The WAL containing changesets for each block (i.e. the key-value pairs that change each block). - stateWAL StreamStore - - // Cancelled to signal the run loop to stop. - ctx context.Context - - // Closed by Close to signal the run loop to stop. - stopCh chan struct{} - - // Tracks the run loop goroutine so Close can wait for it to exit. - wg sync.WaitGroup -} - -func NewStorageGarbageCollector( - ctx context.Context, - config *StorageGarbageCollectorConfig, - stateStores []SnapshotStore, - stateWAL StreamStore, -) (*StorageGarbageCollector, error) { - - if err := config.Validate(); err != nil { - return nil, fmt.Errorf("invalid storage garbage collector config: %w", err) - } - - s := &StorageGarbageCollector{ - config: config, - stateStores: stateStores, - stateWAL: stateWAL, - ctx: ctx, - stopCh: make(chan struct{}), - } - - s.wg.Add(1) - go s.run() - - return s, nil -} - -// Close stops the storage garbage collector and waits for its run loop to exit. It must be called exactly once. -func (s *StorageGarbageCollector) Close() error { - close(s.stopCh) - s.wg.Wait() - return nil -} - -// run periodically drives prune cycles until the manager is stopped. All decision logic lives in prune so it can -// be unit tested without threading. -func (s *StorageGarbageCollector) run() { - defer s.wg.Done() - - //nolint:gosec // G115 - Config.Validate rejects PruneIntervalSeconds large enough to overflow this conversion. - ticker := time.NewTicker(time.Duration(s.config.PruneIntervalSeconds) * time.Second) - defer ticker.Stop() - - for { - select { - case <-s.stopCh: - return - case <-s.ctx.Done(): - return - case <-ticker.C: - if err := prune(s.config.RollbackWindow, s.stateStores, s.stateWAL); err != nil { - logger.Error("prune cycle failed", "err", err) - } - } - } -} - -// prune performs a single prune cycle: it observes the blocks retained by each managed store, computes how far each -// store may prune while preserving the rollback window, and issues the prune commands. -func prune( - rollbackWindow uint64, - stateStores []SnapshotStore, - stateWAL StreamStore, -) error { - stateWALStart, stateWALEnd, stateWALHasData, err := stateWAL.GetStoredBlocks() - if err != nil { - return fmt.Errorf("failed to read stored blocks from %s: %w", stateWAL.Name(), err) - } - - stateStoreBlocks := make([][]uint64, len(stateStores)) - anyStoreEmpty := false - for i, store := range stateStores { - blocks, err := store.GetStoredBlocks() - if err != nil { - return fmt.Errorf("failed to read stored blocks from %s: %w", store.Name(), err) - } - stateStoreBlocks[i] = blocks - if len(blocks) == 0 { - anyStoreEmpty = true - } - } - - if !stateWALHasData || anyStoreEmpty { - // We only prune if the WAL and every provided store has at least some data. An empty store is treated as - // unknown rather than empty (a snapshot may be mid-write and take hours), and pruning against it would risk - // breaking the primary invariant (i.e. breaking the ability to roll back by the act of deletion). - logArgs := []any{"stateWAL", blockRange(stateWALStart, stateWALEnd), "stateWALHasData", stateWALHasData} - for i, store := range stateStores { - logArgs = append(logArgs, store.Name(), stateStoreBlocks[i]) - } - logger.Info("skipping pruning, not all stores have data", logArgs...) - return nil - } - - // The latest committed block is defined by the head of the state WAL. - latestBlock := stateWALEnd - - // The oldest block we must remain able to roll back to. - var oldestBlockNeeded uint64 - if latestBlock > rollbackWindow { - oldestBlockNeeded = latestBlock - rollbackWindow - } - - floors := make([]uint64, len(stateStores)) - for i, blocks := range stateStoreBlocks { - floors[i] = snapshotPruningFloor(blocks, oldestBlockNeeded) - } - - // The WAL retains from the oldest block any state store still needs. With no state stores, nothing depends on the - // WAL for rollback, so it retains only the rollback window. - stateWALFloor := oldestBlockNeeded - for i, floor := range floors { - if i == 0 || floor < stateWALFloor { - stateWALFloor = floor - } - } - - logArgs := []any{"latestBlock", latestBlock, "oldestBlockNeeded", oldestBlockNeeded} - for i, store := range stateStores { - logArgs = append(logArgs, - store.Name()+"Initial", stateStoreBlocks[i], - store.Name()+"Final", blocksAtOrAbove(stateStoreBlocks[i], floors[i]), - ) - } - logArgs = append(logArgs, - "stateWALInitial", blockRange(stateWALStart, stateWALEnd), - "stateWALFinal", blockRange(stateWALFloor, stateWALEnd), - ) - logger.Info("pruning storage", logArgs...) - - for i, store := range stateStores { - if err := store.PruneBelow(floors[i]); err != nil { - return fmt.Errorf("failed to prune %s below %d: %w", store.Name(), floors[i], err) - } - } - if err := stateWAL.PruneBelow(stateWALFloor); err != nil { - return fmt.Errorf("failed to prune %s below %d: %w", stateWAL.Name(), stateWALFloor, err) - } - - return nil -} - -// Given a list of snapshot block numbers, determine the lowest snapshot we need to keep in order to be able -// to roll back to the target block number. -// -// Returns the highest numbered block from snapshotBlocks that is less than or equal to the rollbackTarget. If every -// snapshot is greater than rollbackTarget, the lowest snapshot is returned, since none can be safely pruned. -// snapshotBlocks must be non-empty and sorted in ascending order, per the SnapshotStore contract. -func snapshotPruningFloor( - // Blocks we have snapshots for, in ascending order. Must be non-empty. - snapshotBlocks []uint64, - // The target block that the system needs to be able to roll back to. - rollbackTarget uint64, -) (floor uint64) { - - floor = snapshotBlocks[0] // guaranteed non-empty - for _, b := range snapshotBlocks { - if b > rollbackTarget { - break - } - floor = b - } - return floor -} - -// blocksAtOrAbove returns the subset of blocks that are >= floor, preserving order. This is the set a snapshot store is -// expected to hold after PruneBelow(floor) completes. -func blocksAtOrAbove(blocks []uint64, floor uint64) []uint64 { - result := make([]uint64, 0, len(blocks)) - for _, b := range blocks { - if b >= floor { - result = append(result, b) - } - } - return result -} - -// blockRange renders an inclusive block range for logging. -func blockRange(start uint64, end uint64) string { - return fmt.Sprintf("[%d, %d]", start, end) -} diff --git a/sei-db/management/storage_garbage_collector_config.go b/sei-db/management/storage_garbage_collector_config.go deleted file mode 100644 index 41ba06af13..0000000000 --- a/sei-db/management/storage_garbage_collector_config.go +++ /dev/null @@ -1,45 +0,0 @@ -package management - -import ( - "fmt" - "math" - "time" -) - -// Configures a StorageGarbageCollector. -type StorageGarbageCollectorConfig struct { - - // The maximum number of blocks the system should be able to roll back at any point in time. - // Storage garbage collector ensures that enough data is kept on disk such that the system can always - // roll back this many blocks. - // - // Note that the "always able to rollback" invariant may be broken after a rollback. For example, if we normally - // require a rollback window of 10,000 blocks and then we rollback 5,000 blocks, we can then only rollback an - // additional 5,000 blocks after that first rollback. - RollbackWindow uint64 - - // The frequency of prune operations, in seconds. - PruneIntervalSeconds uint64 -} - -// Construct a default storage garbage collector config. -func DefaultStorageGarbageCollectorConfig() *StorageGarbageCollectorConfig { - return &StorageGarbageCollectorConfig{ - RollbackWindow: 10_000, - PruneIntervalSeconds: 60, - } -} - -// Validate the storage garbage collector's config. -func (c *StorageGarbageCollectorConfig) Validate() error { - // A zero rollback window is legal: it means the system prunes as aggressively as possible. - if c.PruneIntervalSeconds == 0 { - return fmt.Errorf("prune interval must be greater than 0 seconds") - } - // The prune interval is converted to a time.Duration (int64 nanoseconds) when the ticker is created. Reject values - // that would overflow that conversion so the ticker can never be handed a zero/negative duration. - if c.PruneIntervalSeconds > math.MaxInt64/uint64(time.Second) { - return fmt.Errorf("prune interval must be at most %d seconds", math.MaxInt64/uint64(time.Second)) - } - return nil -} diff --git a/sei-db/management/storage_garbage_collector_test.go b/sei-db/management/storage_garbage_collector_test.go deleted file mode 100644 index 8e8d0997e9..0000000000 --- a/sei-db/management/storage_garbage_collector_test.go +++ /dev/null @@ -1,415 +0,0 @@ -package management - -import ( - "context" - "errors" - "fmt" - "math" - "math/rand" - "testing" - "time" - - "github.com/stretchr/testify/require" -) - -// mockSnapshotStore is a hand-written SnapshotStore for tests. It records the last PruneBelow argument and returns -// canned GetStoredBlocks data or errors. -type mockSnapshotStore struct { - name string - blocks []uint64 - getErr error - - pruneBelowCalled bool - prunedBelow uint64 - pruneErr error -} - -func (m *mockSnapshotStore) GetStoredBlocks() ([]uint64, error) { - return m.blocks, m.getErr -} - -func (m *mockSnapshotStore) PruneBelow(blockNumber uint64) error { - m.pruneBelowCalled = true - m.prunedBelow = blockNumber - return m.pruneErr -} - -func (m *mockSnapshotStore) Name() string { - return m.name -} - -// mockStreamStore is a hand-written StreamStore for tests. -type mockStreamStore struct { - name string - start uint64 - end uint64 - hasData bool - getErr error - - pruneBelowCalled bool - prunedBelow uint64 - pruneErr error -} - -func (m *mockStreamStore) GetStoredBlocks() (uint64, uint64, bool, error) { - return m.start, m.end, m.hasData, m.getErr -} - -func (m *mockStreamStore) PruneBelow(blockNumber uint64) error { - m.pruneBelowCalled = true - m.prunedBelow = blockNumber - return m.pruneErr -} - -func (m *mockStreamStore) Name() string { - return m.name -} - -func toStores(s []*mockSnapshotStore) []SnapshotStore { - result := make([]SnapshotStore, len(s)) - for i, store := range s { - result[i] = store - } - return result -} - -// TestPruneDecisions covers the full decision matrix: which floor each store is pruned to, what the WAL is pruned to, -// and when a cycle is skipped entirely. want == nil means "PruneBelow must not be called". -func TestPruneDecisions(t *testing.T) { - cases := []struct { - name string - rollbackWindow uint64 - wal *mockStreamStore - storeBlocks [][]uint64 - wantStores []*uint64 // expected prunedBelow per store; nil => not pruned - wantWAL *uint64 // expected WAL prunedBelow; nil => not pruned - }{ - { - name: "single store (flatKV) normal", - rollbackWindow: 10_000, - wal: &mockStreamStore{start: 1, end: 100_000, hasData: true}, - storeBlocks: [][]uint64{{80_000, 85_000, 92_000}}, - wantStores: []*uint64{ptr(85_000)}, - wantWAL: ptr(85_000), - }, - { - name: "three stores, WAL floor is the min", - rollbackWindow: 10_000, - wal: &mockStreamStore{start: 1, end: 100_000, hasData: true}, - storeBlocks: [][]uint64{{80_000, 85_000, 92_000}, {70_000, 88_000, 95_000}, {50_000, 90_000, 99_000}}, - wantStores: []*uint64{ptr(85_000), ptr(88_000), ptr(90_000)}, - wantWAL: ptr(85_000), - }, - { - name: "rollback window zero => oldestNeeded == latest", - rollbackWindow: 0, - wal: &mockStreamStore{start: 1, end: 100_000, hasData: true}, - storeBlocks: [][]uint64{{80_000, 100_000}, {90_000, 100_000}}, - wantStores: []*uint64{ptr(100_000), ptr(100_000)}, - wantWAL: ptr(100_000), - }, - { - name: "latest == rollback window => oldestNeeded == 0", - rollbackWindow: 100_000, - wal: &mockStreamStore{start: 1, end: 100_000, hasData: true}, - storeBlocks: [][]uint64{{500, 1_000}}, - // oldestNeeded 0, no snapshot <= 0, so floor is the lowest snapshot. - wantStores: []*uint64{ptr(500)}, - wantWAL: ptr(500), - }, - { - name: "latest below rollback window => underflow clamped to 0", - rollbackWindow: 10_000, - wal: &mockStreamStore{start: 1, end: 5_000, hasData: true}, - storeBlocks: [][]uint64{{1_000, 2_000}, {1_500}}, - wantStores: []*uint64{ptr(1_000), ptr(1_500)}, - wantWAL: ptr(1_000), - }, - { - name: "snapshot exactly at oldestNeeded is retained", - rollbackWindow: 10_000, - wal: &mockStreamStore{start: 1, end: 100_000, hasData: true}, - storeBlocks: [][]uint64{{50_000, 90_000}, {90_000}}, - wantStores: []*uint64{ptr(90_000), ptr(90_000)}, - wantWAL: ptr(90_000), - }, - { - name: "all snapshots newer than oldestNeeded => floor is oldest snapshot", - rollbackWindow: 10_000, - wal: &mockStreamStore{start: 1, end: 100_000, hasData: true}, - // oldestNeeded 90,000; every snapshot is above it. - storeBlocks: [][]uint64{{95_000, 97_000}, {92_000, 99_000}}, - wantStores: []*uint64{ptr(95_000), ptr(92_000)}, - wantWAL: ptr(92_000), - }, - { - name: "mixed: one store at/below oldest, one all-above => WAL is the min", - rollbackWindow: 10_000, - wal: &mockStreamStore{start: 1, end: 100_000, hasData: true}, - storeBlocks: [][]uint64{{85_000, 92_000}, {95_000, 99_000}}, - wantStores: []*uint64{ptr(85_000), ptr(95_000)}, - wantWAL: ptr(85_000), - }, - { - name: "no stores + WAL has data => prune WAL to the rollback window", - rollbackWindow: 10_000, - wal: &mockStreamStore{start: 1, end: 100_000, hasData: true}, - storeBlocks: nil, - wantStores: nil, - wantWAL: ptr(90_000), - }, - { - name: "no stores + empty WAL => skip", - rollbackWindow: 10_000, - wal: &mockStreamStore{hasData: false}, - storeBlocks: nil, - wantStores: nil, - wantWAL: nil, - }, - { - name: "empty WAL => skip everything", - rollbackWindow: 10_000, - wal: &mockStreamStore{hasData: false}, - storeBlocks: [][]uint64{{80_000}, {80_000}}, - wantStores: []*uint64{nil, nil}, - wantWAL: nil, - }, - { - name: "one store empty among many => skip everything", - rollbackWindow: 10_000, - wal: &mockStreamStore{start: 1, end: 100_000, hasData: true}, - storeBlocks: [][]uint64{{80_000}, nil, {80_000}}, - wantStores: []*uint64{nil, nil, nil}, - wantWAL: nil, - }, - { - name: "all stores empty => skip everything", - rollbackWindow: 10_000, - wal: &mockStreamStore{start: 1, end: 100_000, hasData: true}, - storeBlocks: [][]uint64{nil, nil}, - wantStores: []*uint64{nil, nil}, - wantWAL: nil, - }, - } - - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - mocks := make([]*mockSnapshotStore, len(tc.storeBlocks)) - for i, blocks := range tc.storeBlocks { - mocks[i] = &mockSnapshotStore{name: fmt.Sprintf("store%d", i), blocks: blocks} - } - tc.wal.name = "stateWAL" - - require.NoError(t, prune(tc.rollbackWindow, toStores(mocks), tc.wal)) - - for i, want := range tc.wantStores { - if want == nil { - require.Falsef(t, mocks[i].pruneBelowCalled, "store%d should not be pruned", i) - } else { - require.Truef(t, mocks[i].pruneBelowCalled, "store%d should be pruned", i) - require.Equalf(t, *want, mocks[i].prunedBelow, "store%d prune floor", i) - } - } - if tc.wantWAL == nil { - require.False(t, tc.wal.pruneBelowCalled, "WAL should not be pruned") - } else { - require.True(t, tc.wal.pruneBelowCalled, "WAL should be pruned") - require.Equal(t, *tc.wantWAL, tc.wal.prunedBelow, "WAL prune floor") - } - }) - } -} - -func TestPruneWALGetError(t *testing.T) { - sentinel := errors.New("boom") - a := &mockSnapshotStore{name: "a", blocks: []uint64{80_000}} - wal := &mockStreamStore{name: "stateWAL", getErr: sentinel} - - err := prune(10_000, toStores([]*mockSnapshotStore{a}), wal) - require.ErrorIs(t, err, sentinel) - require.ErrorContains(t, err, "stateWAL") - require.False(t, a.pruneBelowCalled) - require.False(t, wal.pruneBelowCalled) -} - -func TestPruneStoreGetError(t *testing.T) { - sentinel := errors.New("boom") - a := &mockSnapshotStore{name: "commitStore", getErr: sentinel} - wal := &mockStreamStore{name: "stateWAL", start: 1, end: 100_000, hasData: true} - - err := prune(10_000, toStores([]*mockSnapshotStore{a}), wal) - require.ErrorIs(t, err, sentinel) - require.ErrorContains(t, err, "commitStore") - require.False(t, wal.pruneBelowCalled) -} - -func TestPruneStorePruneErrorStopsBeforeLaterStoresAndWAL(t *testing.T) { - sentinel := errors.New("boom") - // The first store fails to prune; later stores and the WAL must be left untouched. - a := &mockSnapshotStore{name: "a", blocks: []uint64{80_000}, pruneErr: sentinel} - b := &mockSnapshotStore{name: "b", blocks: []uint64{80_000}} - wal := &mockStreamStore{name: "stateWAL", start: 1, end: 100_000, hasData: true} - - err := prune(10_000, toStores([]*mockSnapshotStore{a, b}), wal) - require.ErrorIs(t, err, sentinel) - require.ErrorContains(t, err, "a") - require.True(t, a.pruneBelowCalled) - require.False(t, b.pruneBelowCalled) - require.False(t, wal.pruneBelowCalled) -} - -func TestPruneWALPruneError(t *testing.T) { - sentinel := errors.New("boom") - // All stores prune successfully, then the WAL prune fails. - a := &mockSnapshotStore{name: "a", blocks: []uint64{80_000}} - b := &mockSnapshotStore{name: "b", blocks: []uint64{85_000}} - wal := &mockStreamStore{name: "stateWAL", start: 1, end: 100_000, hasData: true, pruneErr: sentinel} - - err := prune(10_000, toStores([]*mockSnapshotStore{a, b}), wal) - require.ErrorIs(t, err, sentinel) - require.ErrorContains(t, err, "stateWAL") - require.ErrorContains(t, err, "80000") // WAL floor = min(store floors) - require.True(t, a.pruneBelowCalled) - require.True(t, b.pruneBelowCalled) - require.True(t, wal.pruneBelowCalled) -} - -func TestSnapshotPruningFloor(t *testing.T) { - cases := []struct { - name string - blocks []uint64 - target uint64 - wantFloor uint64 - }{ - {name: "single below", blocks: []uint64{50}, target: 100, wantFloor: 50}, - {name: "single above", blocks: []uint64{150}, target: 100, wantFloor: 150}, - {name: "single exact", blocks: []uint64{100}, target: 100, wantFloor: 100}, - {name: "all below picks max", blocks: []uint64{10, 20, 30}, target: 100, wantFloor: 30}, - {name: "all above picks min", blocks: []uint64{150, 200, 300}, target: 100, wantFloor: 150}, - {name: "exact match preferred", blocks: []uint64{50, 100, 150}, target: 100, wantFloor: 100}, - {name: "mixed", blocks: []uint64{10, 40, 90, 150, 200}, target: 100, wantFloor: 90}, - {name: "target zero all above", blocks: []uint64{5, 10}, target: 0, wantFloor: 5}, - {name: "duplicate blocks", blocks: []uint64{50, 50, 100, 100}, target: 100, wantFloor: 100}, - {name: "brackets target", blocks: []uint64{99, 101}, target: 100, wantFloor: 99}, - } - - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - require.Equal(t, tc.wantFloor, snapshotPruningFloor(tc.blocks, tc.target)) - }) - } -} - -// TestSnapshotPruningFloorProperty checks snapshotPruningFloor against a brute-force oracle over random sorted input: -// the result must be the highest block <= target, or the lowest block when none qualify. -func TestSnapshotPruningFloorProperty(t *testing.T) { - rng := rand.New(rand.NewSource(1)) - for iter := 0; iter < 2000; iter++ { - n := 1 + rng.Intn(8) - blocks := make([]uint64, n) - var next uint64 - for i := range blocks { - next += uint64(rng.Intn(5)) // ascending, allows duplicates - blocks[i] = next - } - target := uint64(rng.Intn(int(next) + 5)) - - // Oracle: highest block <= target, or the lowest block (blocks[0]) when none qualify. - want := blocks[0] - for _, b := range blocks { - if b <= target { - want = b - } - } - - require.Equalf(t, want, snapshotPruningFloor(blocks, target), "blocks=%v target=%d", blocks, target) - } -} - -func TestBlocksAtOrAbove(t *testing.T) { - require.Empty(t, blocksAtOrAbove(nil, 100)) - require.Equal(t, []uint64{100, 150}, blocksAtOrAbove([]uint64{50, 100, 150}, 100)) - require.Equal(t, []uint64{10, 20, 30}, blocksAtOrAbove([]uint64{10, 20, 30}, 0)) - require.Empty(t, blocksAtOrAbove([]uint64{10, 20}, 100)) - // Floor strictly between elements. - require.Equal(t, []uint64{60, 70}, blocksAtOrAbove([]uint64{50, 60, 70}, 55)) -} - -// TestBlocksAtOrAboveProperty checks blocksAtOrAbove equals the filter b >= floor, preserving order. -func TestBlocksAtOrAboveProperty(t *testing.T) { - rng := rand.New(rand.NewSource(2)) - for iter := 0; iter < 2000; iter++ { - n := rng.Intn(8) - blocks := make([]uint64, n) - for i := range blocks { - blocks[i] = uint64(rng.Intn(50)) - } - floor := uint64(rng.Intn(50)) - - want := []uint64{} - for _, b := range blocks { - if b >= floor { - want = append(want, b) - } - } - got := blocksAtOrAbove(blocks, floor) - require.Equalf(t, want, got, "blocks=%v floor=%d", blocks, floor) - } -} - -func TestBlockRange(t *testing.T) { - require.Equal(t, "[1, 100]", blockRange(1, 100)) - require.Equal(t, "[0, 0]", blockRange(0, 0)) -} - -func TestDefaultStorageGarbageCollectorConfig(t *testing.T) { - cfg := DefaultStorageGarbageCollectorConfig() - require.Equal(t, uint64(10_000), cfg.RollbackWindow) - require.Equal(t, uint64(60), cfg.PruneIntervalSeconds) -} - -func TestValidate(t *testing.T) { - require.NoError(t, DefaultStorageGarbageCollectorConfig().Validate()) - - // A zero rollback window is legal. - require.NoError(t, (&StorageGarbageCollectorConfig{RollbackWindow: 0, PruneIntervalSeconds: 60}).Validate()) - - err := (&StorageGarbageCollectorConfig{PruneIntervalSeconds: 0}).Validate() - require.ErrorContains(t, err, "prune interval") - - // The largest interval that does not overflow time.Duration is accepted; one larger is rejected. - maxInterval := uint64(math.MaxInt64) / uint64(time.Second) - require.NoError(t, (&StorageGarbageCollectorConfig{PruneIntervalSeconds: maxInterval}).Validate()) - require.ErrorContains(t, (&StorageGarbageCollectorConfig{PruneIntervalSeconds: maxInterval + 1}).Validate(), "at most") -} - -func TestNewStorageGarbageCollectorInvalidConfig(t *testing.T) { - sm, err := NewStorageGarbageCollector( - context.Background(), - &StorageGarbageCollectorConfig{PruneIntervalSeconds: 0}, - toStores([]*mockSnapshotStore{{name: "a"}}), - &mockStreamStore{name: "stateWAL"}, - ) - require.Error(t, err) - require.Nil(t, sm) -} - -func TestNewStorageGarbageCollectorConstructAndClose(t *testing.T) { - a := &mockSnapshotStore{name: "a", blocks: []uint64{100}} - wal := &mockStreamStore{name: "stateWAL", start: 1, end: 100, hasData: true} - sm, err := NewStorageGarbageCollector( - context.Background(), - &StorageGarbageCollectorConfig{RollbackWindow: 10, PruneIntervalSeconds: 60}, - toStores([]*mockSnapshotStore{a}), - wal, - ) - require.NoError(t, err) - require.NotNil(t, sm) - - // Close must return without hanging. - require.NoError(t, sm.Close()) -} - -func ptr(v uint64) *uint64 { - return &v -} diff --git a/sei-db/management/store_types.go b/sei-db/management/store_types.go deleted file mode 100644 index d53c157870..0000000000 --- a/sei-db/management/store_types.go +++ /dev/null @@ -1,34 +0,0 @@ -package management - -// A store that is persisted via snapshots (e.g. SC, SS). -type SnapshotStore interface { - - // Fetch a list of all block numbers that this store has snapshots for, in ascending sorted order. - GetStoredBlocks() ([]uint64, error) - - // Instruct the store that it may drop snapshots for all blocks below a specified number. - // Store may drop data asynchronously. - PruneBelow(blockNumber uint64) error - - // Return the name of the snapshot store. - Name() string -} - -// A store that is made up of a stream of data (e.g. a WAL). -type StreamStore interface { - - // Fetch the range of blocks stored within this stream. - GetStoredBlocks() ( - start uint64, // inclusive; meaningful only when hasData is true - end uint64, // inclusive; meaningful only when hasData is true - hasData bool, // true if the stream contains at least one block - err error, - ) - - // Instruct the store that it may drop data for all blocks below a specified number. - // Store may drop data asynchronously. - PruneBelow(blockNumber uint64) error - - // Return the name of the stream store. - Name() string -}