From 4c72a931323cd86a0a3fd994ecc35476e1c90c77 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Wed, 26 Aug 2026 13:13:35 -0700 Subject: [PATCH 01/11] Add global checkpoint scheduler --- sei-db/controller/checkpoint_scheduler.go | 261 +++++++++--- .../controller/checkpoint_scheduler_test.go | 401 ++++++++++++++++++ sei-db/controller/checkpointable_store.go | 22 + sei-db/db_engine/pebbledb/mvcc/db_test.go | 10 +- sei-db/db_engine/types/types.go | 5 +- sei-db/state_db/sc/flatkv/checkpoint.go | 79 ++++ sei-db/state_db/sc/flatkv/checkpoint_test.go | 114 +++++ sei-db/state_db/sc/flatkv/store.go | 9 + sei-db/state_db/sc/flatkv/store_write.go | 7 +- sei-db/state_db/ss/composite/checkpoint.go | 32 ++ sei-db/state_db/ss/composite/snapshot.go | 88 +++- sei-db/state_db/ss/composite/snapshot_test.go | 17 +- sei-db/state_db/ss/cosmos/store.go | 9 +- sei-db/state_db/ss/evm/store.go | 15 +- sei-db/state_db/ss/snapshot/checkpoint.go | 92 ++++ sei-db/state_db/ss/snapshot/manager.go | 19 +- sei-db/state_db/ss/snapshot/manager_test.go | 15 +- 17 files changed, 1072 insertions(+), 123 deletions(-) create mode 100644 sei-db/controller/checkpoint_scheduler_test.go create mode 100644 sei-db/controller/checkpointable_store.go create mode 100644 sei-db/state_db/sc/flatkv/checkpoint.go create mode 100644 sei-db/state_db/sc/flatkv/checkpoint_test.go create mode 100644 sei-db/state_db/ss/composite/checkpoint.go create mode 100644 sei-db/state_db/ss/snapshot/checkpoint.go diff --git a/sei-db/controller/checkpoint_scheduler.go b/sei-db/controller/checkpoint_scheduler.go index bfc3dcdaa4..d57e1e1988 100644 --- a/sei-db/controller/checkpoint_scheduler.go +++ b/sei-db/controller/checkpoint_scheduler.go @@ -4,91 +4,228 @@ package controller import ( + "context" "errors" "fmt" + "maps" + "slices" + "strings" "sync" + "time" - "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" + "github.com/sei-protocol/seilog" ) -// CheckpointScheduler coordinates checkpoints for stores with in-flight writes. +var checkpointLogger = seilog.NewLogger("db", "checkpoint") + +// checkpointPollInterval is how often the scheduler looks for a boundary to dispatch. It has to stay +// well inside the time the stores take to cover an interval's worth of versions, or boundaries pass +// undispatched. +const checkpointPollInterval = 10 * time.Second + +// CheckpointConfig is the cadence a CheckpointScheduler holds every registered store to. +type CheckpointConfig struct { + // CheckpointInterval is how many blocks apart checkpoints are taken. 0 turns checkpointing off. + CheckpointInterval int64 + + // MinTimeBetweenCheckpoints is the shortest wall-clock gap allowed between one checkpoint + // finishing and the next being scheduled, which bounds how fast a node replaying blocks + // checkpoints. 0 leaves CheckpointInterval as the only pacing. + MinTimeBetweenCheckpoints time.Duration +} + +// Validate reports whether this config describes a cadence that can be scheduled. +func (c CheckpointConfig) Validate() error { + if c.CheckpointInterval < 0 { + return fmt.Errorf("checkpoint interval must not be negative, got %d", c.CheckpointInterval) + } + if c.MinTimeBetweenCheckpoints < 0 { + return fmt.Errorf("minimum time between checkpoints must not be negative, got %s", + c.MinTimeBetweenCheckpoints) + } + return nil +} + +// CheckpointScheduler drives one checkpoint cadence across every store registered with it, so that a +// node's stores hold checkpoints of the same versions rather than of whatever version each happened to +// be at. Each cycle picks the next interval boundary above every store's committed version and hands +// that one version to all of them, to checkpoint when their own write paths reach it; one target is +// outstanding at a time. // -// The engine-side capabilities this builds on — types.Checkpointable, -// types.DrainBarrier and types.CheckpointVersionSetter — stay with the engines -// that implement them. What lives here is the decision of when a checkpoint runs -// and what version it is labeled with. -type CheckpointScheduler interface { - SupportsCheckpoint() bool - ScheduleCheckpoint(destDir string, shouldRun func() bool, done func(error)) - SetCheckpointVersion(destDir string, version int64) error +// Start and Close are the owner's to call, from one goroutine: nothing here is guarded. +type CheckpointScheduler struct { + config CheckpointConfig + ctx context.Context + stopCh chan struct{} + wg sync.WaitGroup + + stores map[string]CheckpointableStore + started bool + closed bool + + // Only the run loop touches these, through scheduleNextCheckpoint. + scheduledVersion int64 + lastCheckpointAt time.Time } -// ErrCheckpointCanceled reports that a queued checkpoint was canceled before -// it started. -var ErrCheckpointCanceled = errors.New("state store checkpoint canceled") - -// SupportsCheckpoint reports whether db carries every engine capability a scheduled checkpoint needs. -// A store built from several engines answers for the set: one engine short of the capabilities makes -// the whole snapshot unpublishable. -func SupportsCheckpoint(db types.StateStore) bool { - _, checkpointable := db.(types.Checkpointable) - _, barrier := db.(types.DrainBarrier) - _, versionSetter := db.(types.CheckpointVersionSetter) - return checkpointable && barrier && versionSetter +// NewCheckpointScheduler returns a scheduler that holds every store in stores to config, or that +// schedules nothing when config turns checkpointing off. Call Start to begin. +func NewCheckpointScheduler( + ctx context.Context, + config CheckpointConfig, + stores map[string]CheckpointableStore, +) (*CheckpointScheduler, error) { + if ctx == nil { + return nil, errors.New("context is required") + } + if err := config.Validate(); err != nil { + return nil, err + } + copied := make(map[string]CheckpointableStore, len(stores)) + for name, store := range stores { + if name == "" { + return nil, errors.New("checkpoint store name is required") + } + if store == nil { + return nil, fmt.Errorf("checkpoint store %q is nil", name) + } + copied[name] = store + } + return &CheckpointScheduler{ + config: config, + ctx: ctx, + stopCh: make(chan struct{}), + stores: copied, + }, nil } -// FanIn returns a report callback for n parallel branches. Each branch calls it once, and the last -// call passes done the first error any branch reported, or nil. -func FanIn(n int, done func(error)) func(error) { - var ( - mu sync.Mutex - remaining = n - firstErr error +// Start begins dispatching targets until Close is called or ctx is cancelled. Starting twice is an +// error. +func (s *CheckpointScheduler) Start() error { + if s.closed { + return errors.New("cannot start a closed checkpoint scheduler") + } + if s.started { + return errors.New("checkpoint scheduler already started") + } + s.started = true + checkpointLogger.Info("checkpoint scheduler started", + "interval", s.config.CheckpointInterval, + "minTimeBetweenCheckpoints", s.config.MinTimeBetweenCheckpoints, + "stores", strings.Join(slices.Sorted(maps.Keys(s.stores)), ","), ) - return func(err error) { - mu.Lock() - if err != nil && firstErr == nil { - firstErr = err + s.wg.Add(1) + go s.run() + return nil +} + +// Close stops dispatching and waits for the run loop to exit. +func (s *CheckpointScheduler) Close() error { + if s.closed { + return nil + } + s.closed = true + close(s.stopCh) + s.wg.Wait() + return nil +} + +// CheckpointInProgress reports whether any registered store is still writing a checkpoint. +func (s *CheckpointScheduler) CheckpointInProgress() bool { + for _, store := range s.stores { + if store.CheckpointInProgress() { + return true } - remaining-- - isLast := remaining == 0 - // Read under the lock: a peer branch may report between the unlock and the call to done. - outcome := firstErr - mu.Unlock() - if isLast { - done(outcome) + } + return false +} + +func (s *CheckpointScheduler) run() { + defer s.wg.Done() + + if s.config.CheckpointInterval == 0 || len(s.stores) == 0 { + return + } + + ticker := time.NewTicker(checkpointPollInterval) + defer ticker.Stop() + + for { + // Ahead of the first wait, not after it: there is already a boundary to announce by the time + // Start returns, and waiting out a poll interval only delays the first checkpoint. + s.scheduleNextCheckpoint() + + select { + case <-s.stopCh: + return + case <-s.ctx.Done(): + return + case <-ticker.C: } } } -// ScheduleCheckpoint checkpoints an engine after all writes already enqueued -// on it have been applied. -func ScheduleCheckpoint(db types.StateStore, destDir string, shouldRun func() bool, done func(error)) { - cp, ok := db.(types.Checkpointable) - if !ok { - done(fmt.Errorf("state store backend %T does not support checkpoints", db)) +// scheduleNextCheckpoint runs one cycle: it hands the next boundary to every store, or does nothing +// when a store is still writing the last checkpoint, has not reached its scheduled version, or the +// last one finished too recently. +func (s *CheckpointScheduler) scheduleNextCheckpoint() { + if s.CheckpointInProgress() { return } - barrier, ok := db.(types.DrainBarrier) - if !ok { - done(fmt.Errorf("state store backend %T does not support ordered checkpoint barriers", db)) + if s.scheduledVersion != 0 && !s.allReached(s.scheduledVersion) { return } - barrier.ScheduleAtDrain(func() { - if shouldRun != nil && !shouldRun() { - done(ErrCheckpointCanceled) - return + s.noteCheckpointFinished() + if s.withinMinTime() { + return + } + + targetVersion := nextCheckpointVersion(s.stores, s.config.CheckpointInterval) + for _, store := range s.stores { + store.ScheduleCheckpoint(targetVersion) + } + s.scheduledVersion = targetVersion + checkpointLogger.Info("checkpoint scheduled", + "targetVersion", targetVersion, "stores", strings.Join(slices.Sorted(maps.Keys(s.stores)), ",")) +} + +// allReached reports whether every store has committed at least version. +func (s *CheckpointScheduler) allReached(version int64) bool { + for _, store := range s.stores { + if store.LatestVersion() < version { + return false } - done(cp.Checkpoint(destDir)) - }) + } + return true +} + +// noteCheckpointFinished records that the last scheduled checkpoint completed, starting the +// minimum-time gate. scheduledVersion is what marks this as the completion: later cycles also find +// no store writing, and treating those as completions too would push the gate forward every poll. +func (s *CheckpointScheduler) noteCheckpointFinished() { + if s.scheduledVersion == 0 { + return + } + s.scheduledVersion = 0 + s.lastCheckpointAt = time.Now() +} + +// withinMinTime reports whether the last checkpoint finished too recently for another to be scheduled. +// +// Timed from the checkpoint finishing rather than from its dispatch: dispatch runs an interval of +// blocks ahead, so timing from there would leave the gap short by however long the stores took. +func (s *CheckpointScheduler) withinMinTime() bool { + if s.config.MinTimeBetweenCheckpoints == 0 || s.lastCheckpointAt.IsZero() { + return false + } + return time.Since(s.lastCheckpointAt) < s.config.MinTimeBetweenCheckpoints } -// SetCheckpointVersion makes a completed checkpoint self-describing without -// changing the live database. -func SetCheckpointVersion(db types.StateStore, destDir string, version int64) error { - setter, ok := db.(types.CheckpointVersionSetter) - if !ok { - return fmt.Errorf("state store backend %T cannot set checkpoint version", db) +// nextCheckpointVersion returns the next interval-aligned height strictly above every store's latest version. +func nextCheckpointVersion(stores map[string]CheckpointableStore, interval int64) int64 { + var latest int64 + for _, store := range stores { + latest = max(latest, store.LatestVersion()) } - return setter.SetCheckpointVersion(destDir, version) + return (latest/interval + 1) * interval } diff --git a/sei-db/controller/checkpoint_scheduler_test.go b/sei-db/controller/checkpoint_scheduler_test.go new file mode 100644 index 0000000000..467f3b2b7c --- /dev/null +++ b/sei-db/controller/checkpoint_scheduler_test.go @@ -0,0 +1,401 @@ +package controller + +import ( + "context" + "slices" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// Which target the scheduler picks, and whether it picks one at all, is decided synchronously in +// scheduleNextCheckpoint. Those tests drive it directly and assert exactly; only the tests about the run +// loop itself — that it dispatches, that it stops — start a goroutine. + +// fakeStore stands in for a store the scheduler drives. It records every target it is offered so a +// test can assert what the stores were asked for, which is the scheduler's whole output. +type fakeStore struct { + mu sync.Mutex + version int64 + pending int64 + running bool + offered []int64 +} + +func newFakeStore(version int64) *fakeStore { + return &fakeStore{version: version} +} + +func (f *fakeStore) ScheduleCheckpoint(targetVersion int64) { + f.mu.Lock() + defer f.mu.Unlock() + f.offered = append(f.offered, targetVersion) + f.pending = targetVersion +} + +func (f *fakeStore) LatestVersion() int64 { + f.mu.Lock() + defer f.mu.Unlock() + return f.version +} + +func (f *fakeStore) CheckpointInProgress() bool { + f.mu.Lock() + defer f.mu.Unlock() + return f.running +} + +func (f *fakeStore) setRunning(running bool) { + f.mu.Lock() + defer f.mu.Unlock() + f.running = running +} + +// commitTo advances the store to version, performing an accepted checkpoint on the way past it. +func (f *fakeStore) commitTo(version int64) { + f.mu.Lock() + defer f.mu.Unlock() + f.version = version + if f.pending != 0 && version >= f.pending { + f.pending = 0 + } +} + +func (f *fakeStore) offeredTargets() []int64 { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.offered) +} + +// newScheduler returns an unstarted scheduler over stores. +func newScheduler(t *testing.T, interval int64, stores map[string]*fakeStore) *CheckpointScheduler { + t.Helper() + return newConfiguredScheduler(t, CheckpointConfig{CheckpointInterval: interval}, stores) +} + +// newConfiguredScheduler is newScheduler for the tests that care about more of the cadence than the +// block interval. +func newConfiguredScheduler( + t *testing.T, + config CheckpointConfig, + stores map[string]*fakeStore, +) *CheckpointScheduler { + t.Helper() + copied := make(map[string]CheckpointableStore, len(stores)) + for name, store := range stores { + copied[name] = store + } + scheduler, err := NewCheckpointScheduler(context.Background(), config, copied) + require.NoError(t, err) + return scheduler +} + +// requireLoopStopped waits, with a bound, for the run loop to have exited — or to never have started. +// The wait is bounded rather than a bare wg.Wait because the failure it looks for is a loop that keeps +// running, and that failure has to be a test failure rather than a test that hangs. +func requireLoopStopped(t *testing.T, scheduler *CheckpointScheduler) { + t.Helper() + stopped := make(chan struct{}) + go func() { + scheduler.wg.Wait() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(time.Second): + t.Fatal("checkpoint scheduler run loop is still running") + } +} + +// --------------------------------------------------------------------------- +// Which target a cycle picks +// --------------------------------------------------------------------------- + +func TestNextCheckpointVersionAlignsToInterval(t *testing.T) { + for _, tc := range []struct { + latest, interval, want int64 + }{ + {latest: 0, interval: 1000, want: 1000}, + {latest: 1, interval: 1000, want: 1000}, + {latest: 999, interval: 1000, want: 1000}, + {latest: 1000, interval: 1000, want: 2000}, + {latest: 2100, interval: 1000, want: 3000}, + {latest: 3700, interval: 1000, want: 4000}, + {latest: 4000, interval: 1000, want: 5000}, + } { + stores := map[string]CheckpointableStore{"only": newFakeStore(tc.latest)} + require.Equal(t, tc.want, nextCheckpointVersion(stores, tc.interval), + "latest %d interval %d", tc.latest, tc.interval) + } +} + +// The point of the scheduler: every store is asked for the same version, even when they are at +// different versions when the target is chosen. +func TestCycleOffersOneTargetToEveryStore(t *testing.T) { + fast, slow := newFakeStore(95), newFakeStore(80) + scheduler := newScheduler(t, 100, map[string]*fakeStore{"fast": fast, "slow": slow}) + + scheduler.scheduleNextCheckpoint() + + require.Equal(t, []int64{100}, fast.offeredTargets()) + require.Equal(t, []int64{100}, slow.offeredTargets()) +} + +// The target has to be above every store's version, not above the laggard's: a store that is ahead can +// no longer checkpoint at a version it has passed, and would answer with a different one. +func TestTargetIsAboveTheMostAdvancedStore(t *testing.T) { + ahead, behind := newFakeStore(250), newFakeStore(10) + scheduler := newScheduler(t, 100, map[string]*fakeStore{"ahead": ahead, "behind": behind}) + + scheduler.scheduleNextCheckpoint() + + require.Equal(t, []int64{300}, ahead.offeredTargets()) + require.Equal(t, []int64{300}, behind.offeredTargets()) +} + +// A target is offered well before the stores reach it, which is what makes the shared version an +// invariant rather than a race. Nothing about the current version gates the offer. +func TestTargetIsOfferedAWholeIntervalAhead(t *testing.T) { + store := newFakeStore(1) + scheduler := newScheduler(t, 1000, map[string]*fakeStore{"only": store}) + + scheduler.scheduleNextCheckpoint() + + require.Equal(t, []int64{1000}, store.offeredTargets(), "offered at version 1, 999 blocks early") +} + +// One target is outstanding at a time, so a store that has not reached its target holds the next +// boundary back rather than collecting targets it would service late. +func TestNoNewTargetWhileOneIsOutstanding(t *testing.T) { + prompt, lagging := newFakeStore(0), newFakeStore(0) + scheduler := newScheduler(t, 10, map[string]*fakeStore{"prompt": prompt, "lagging": lagging}) + + scheduler.scheduleNextCheckpoint() + require.Equal(t, []int64{10}, prompt.offeredTargets()) + prompt.commitTo(35) + + // The laggard still holds target 10, so the boundaries at 20 and 30 pass without an offer. + scheduler.scheduleNextCheckpoint() + scheduler.scheduleNextCheckpoint() + require.Equal(t, []int64{10}, prompt.offeredTargets()) + + lagging.commitTo(10) + scheduler.scheduleNextCheckpoint() + require.Equal(t, []int64{10, 40}, prompt.offeredTargets()) + require.Equal(t, []int64{10, 40}, lagging.offeredTargets()) +} + +// A store still writing its snapshot holds the next boundary even after it has reached the height. +func TestNoNewTargetWhileAStoreIsWriting(t *testing.T) { + store := newFakeStore(0) + scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": store}) + + scheduler.scheduleNextCheckpoint() + store.commitTo(10) + store.setRunning(true) + scheduler.scheduleNextCheckpoint() + require.Equal(t, []int64{10}, store.offeredTargets()) + + store.setRunning(false) + scheduler.scheduleNextCheckpoint() + require.Equal(t, []int64{10, 20}, store.offeredTargets()) +} + +// CheckpointInProgress is an aggregate over the stores: true while any store is writing. +func TestCheckpointInProgressReportsAnyStore(t *testing.T) { + first, second := newFakeStore(0), newFakeStore(0) + scheduler := newScheduler(t, 10, map[string]*fakeStore{"first": first, "second": second}) + + require.False(t, scheduler.CheckpointInProgress()) + first.setRunning(true) + second.setRunning(true) + require.True(t, scheduler.CheckpointInProgress()) + + first.setRunning(false) + require.True(t, scheduler.CheckpointInProgress(), "second store is still writing") + second.setRunning(false) + require.False(t, scheduler.CheckpointInProgress()) +} + +// --------------------------------------------------------------------------- +// The minimum-time gate +// --------------------------------------------------------------------------- + +// pacedScheduler returns a scheduler whose minimum-time gate is wide enough that no test elapses it by +// running. Tests that need it elapsed move lastCheckpointAt rather than waiting. +func pacedScheduler(t *testing.T, stores map[string]*fakeStore) *CheckpointScheduler { + t.Helper() + return newConfiguredScheduler(t, CheckpointConfig{ + CheckpointInterval: 10, + MinTimeBetweenCheckpoints: time.Hour, + }, stores) +} + +// Nothing has been checkpointed yet, so there is no gap to enforce and the first boundary is not held. +func TestTheMinTimeGateDoesNotDelayTheFirstCheckpoint(t *testing.T) { + store := newFakeStore(0) + scheduler := pacedScheduler(t, map[string]*fakeStore{"only": store}) + + scheduler.scheduleNextCheckpoint() + + require.Equal(t, []int64{10}, store.offeredTargets()) +} + +// Once a checkpoint has been taken, the next boundary waits for the gate even though the stores have +// long since passed it. +func TestTheMinTimeGateHoldsBackTheNextTarget(t *testing.T) { + store := newFakeStore(0) + scheduler := pacedScheduler(t, map[string]*fakeStore{"only": store}) + + scheduler.scheduleNextCheckpoint() + require.Equal(t, []int64{10}, store.offeredTargets()) + store.commitTo(100) + + scheduler.scheduleNextCheckpoint() + scheduler.scheduleNextCheckpoint() + + require.Equal(t, []int64{10}, store.offeredTargets(), "the gate has not elapsed") +} + +func TestTheMinTimeGateReleasesOnceItElapses(t *testing.T) { + store := newFakeStore(0) + scheduler := pacedScheduler(t, map[string]*fakeStore{"only": store}) + + scheduler.scheduleNextCheckpoint() + store.commitTo(10) + scheduler.scheduleNextCheckpoint() + require.Equal(t, []int64{10}, store.offeredTargets()) + + scheduler.lastCheckpointAt = time.Now().Add(-2 * time.Hour) + scheduler.scheduleNextCheckpoint() + + require.Equal(t, []int64{10, 20}, store.offeredTargets()) +} + +// The gate runs from the checkpoint finishing, not from its target being dispatched. A store that is +// slow to reach its target would otherwise spend the gate's window getting there, and the next +// checkpoint would follow it by less than the configured gap. +func TestTheMinTimeGateIsTimedFromCompletion(t *testing.T) { + store := newFakeStore(0) + scheduler := pacedScheduler(t, map[string]*fakeStore{"only": store}) + + scheduler.scheduleNextCheckpoint() + require.True(t, scheduler.lastCheckpointAt.IsZero(), "dispatching a target does not start the gate") + + store.commitTo(10) + scheduler.scheduleNextCheckpoint() + require.False(t, scheduler.lastCheckpointAt.IsZero(), "the finished checkpoint starts the gate") +} + +func TestAZeroMinTimeLeavesTheBlockIntervalAsTheOnlyPacing(t *testing.T) { + store := newFakeStore(0) + scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": store}) + + scheduler.scheduleNextCheckpoint() + store.commitTo(10) + scheduler.scheduleNextCheckpoint() + + require.Equal(t, []int64{10, 20}, store.offeredTargets()) +} + +// --------------------------------------------------------------------------- +// Lifecycle +// --------------------------------------------------------------------------- + +// The run loop dispatches on its own, which is the one thing the direct-cycle tests above cannot show. +// +// The bound is well inside checkpointPollInterval on purpose: it holds the loop to running its first +// cycle at Start, rather than after sitting out a poll interval first. +func TestStartedSchedulerDispatchesOnItsOwn(t *testing.T) { + store := newFakeStore(0) + scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": store}) + require.NoError(t, scheduler.Start()) + t.Cleanup(func() { require.NoError(t, scheduler.Close()) }) + + require.Eventually(t, func() bool { + return slices.Equal(store.offeredTargets(), []int64{10}) + }, 100*time.Millisecond, time.Millisecond, "the run loop never offered a target") +} + +func TestNewRejectsAnEmptyStoreName(t *testing.T) { + _, err := NewCheckpointScheduler(context.Background(), CheckpointConfig{CheckpointInterval: 10}, + map[string]CheckpointableStore{"": newFakeStore(0)}) + require.ErrorContains(t, err, "store name is required") +} + +func TestNewRejectsANilStore(t *testing.T) { + _, err := NewCheckpointScheduler(context.Background(), CheckpointConfig{CheckpointInterval: 10}, + map[string]CheckpointableStore{"ss": nil}) + require.ErrorContains(t, err, "is nil") +} + +func TestNewRejectsANegativeInterval(t *testing.T) { + _, err := NewCheckpointScheduler(context.Background(), CheckpointConfig{CheckpointInterval: -1}, nil) + require.ErrorContains(t, err, "interval must not be negative") +} + +func TestNewRejectsANegativeMinTime(t *testing.T) { + _, err := NewCheckpointScheduler(context.Background(), CheckpointConfig{ + CheckpointInterval: 10, + MinTimeBetweenCheckpoints: -time.Second, + }, nil) + require.ErrorContains(t, err, "minimum time between checkpoints must not be negative") +} + +// A scheduler with no stores has nobody to checkpoint, so run returns without a loop. +func TestAnEmptyStoreSetRunsNoLoop(t *testing.T) { + scheduler := newScheduler(t, 10, nil) + + require.NoError(t, scheduler.Start()) + requireLoopStopped(t, scheduler) + require.False(t, scheduler.CheckpointInProgress()) + require.NoError(t, scheduler.Close()) +} + +// Interval 0 turns checkpointing off. run returns without a loop so a cycle never divides by zero. +func TestAZeroIntervalRunsNoLoop(t *testing.T) { + store := newFakeStore(0) + scheduler := newScheduler(t, 0, map[string]*fakeStore{"only": store}) + + require.NoError(t, scheduler.Start()) + requireLoopStopped(t, scheduler) + require.Empty(t, store.offeredTargets()) + require.False(t, scheduler.CheckpointInProgress()) + require.NoError(t, scheduler.Close()) +} + +func TestCloseIsIdempotentAndSafeBeforeStart(t *testing.T) { + scheduler := newScheduler(t, 10, nil) + + require.NoError(t, scheduler.Close()) + require.NoError(t, scheduler.Close()) + require.ErrorContains(t, scheduler.Start(), "closed") +} + +func TestCloseStopsTheRunLoop(t *testing.T) { + scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": newFakeStore(0)}) + require.NoError(t, scheduler.Start()) + + require.NoError(t, scheduler.Close()) + + requireLoopStopped(t, scheduler) +} + +// Cancelling the context ends the run loop. Asserted as the loop exiting rather than as an absence of +// further offers: the loop selects over the ticker and the cancellation together, so a cycle already +// runnable at the moment of cancellation may still run, and a test forbidding that fails a few runs in +// a thousand. +func TestCancellingTheContextStopsTheRunLoop(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + scheduler, err := NewCheckpointScheduler(ctx, CheckpointConfig{CheckpointInterval: 10}, + map[string]CheckpointableStore{"only": newFakeStore(0)}) + require.NoError(t, err) + require.NoError(t, scheduler.Start()) + + cancel() + + requireLoopStopped(t, scheduler) + require.NoError(t, scheduler.Close()) +} diff --git a/sei-db/controller/checkpointable_store.go b/sei-db/controller/checkpointable_store.go new file mode 100644 index 0000000000..a5c728dbba --- /dev/null +++ b/sei-db/controller/checkpointable_store.go @@ -0,0 +1,22 @@ +package controller + +// CheckpointableStore is a store whose checkpoints are scheduled by the CheckpointScheduler. +// +// A checkpoint is a point-in-time snapshot of the store at one version. The scheduler decides which +// version that is; a store decides how to create it. +type CheckpointableStore interface { + // ScheduleCheckpoint records that this store should checkpoint at targetVersion. It returns once + // the request is recorded; the store performs the checkpoint when its own write path reaches that + // version. A height the store cannot take — already at or past it, or one already pending — is + // ignored rather than failed: the scheduler has sent the task, and a wrong height must not + // become a nearby height instead. + ScheduleCheckpoint(targetVersion int64) + + // LatestVersion returns the newest version this store has committed, 0 when it has committed + // nothing. The scheduler picks a target above every store's answer, so a store that reports a + // version it has not durably reached is asking to be handed a target it will never see. + LatestVersion() int64 + + // CheckpointInProgress reports whether a checkpoint this store is writing has yet to finish. + CheckpointInProgress() bool +} diff --git a/sei-db/db_engine/pebbledb/mvcc/db_test.go b/sei-db/db_engine/pebbledb/mvcc/db_test.go index 375d0202ba..9d3ca40fca 100644 --- a/sei-db/db_engine/pebbledb/mvcc/db_test.go +++ b/sei-db/db_engine/pebbledb/mvcc/db_test.go @@ -9,9 +9,9 @@ import ( "github.com/stretchr/testify/suite" "github.com/sei-protocol/sei-chain/sei-db/config" - "github.com/sei-protocol/sei-chain/sei-db/controller" sstest "github.com/sei-protocol/sei-chain/sei-db/db_engine/test" "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" + sssnapshot "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/snapshot" ) func TestStorageTestSuite(t *testing.T) { @@ -63,13 +63,13 @@ func TestVersionedCheckpointPreservesFutureLiveMarker(t *testing.T) { dest := filepath.Join(t.TempDir(), "snapshot") done := make(chan error, 1) - controller.ScheduleCheckpoint(store, dest, nil, func(err error) { + sssnapshot.ScheduleCheckpoint(store, dest, nil, func(err error) { done <- err }) require.NoError(t, <-done) // The caller stamps only the label. Earliest is inherited from the // checkpointed DB because prune advances it before deleting history. - require.NoError(t, controller.SetCheckpointVersion(store, dest, 5)) + require.NoError(t, sssnapshot.SetCheckpointVersion(store, dest, 5)) require.Equal(t, int64(10), store.GetLatestVersion()) require.Equal(t, int64(4), store.GetEarliestVersion()) @@ -97,10 +97,10 @@ func TestScheduledCheckpointCanBeCanceledAtBarrier(t *testing.T) { dest := filepath.Join(t.TempDir(), "snapshot") done := make(chan error, 1) - controller.ScheduleCheckpoint(store, dest, func() bool { return false }, func(err error) { + sssnapshot.ScheduleCheckpoint(store, dest, func() bool { return false }, func(err error) { done <- err }) - require.ErrorIs(t, <-done, controller.ErrCheckpointCanceled) + require.ErrorIs(t, <-done, sssnapshot.ErrCheckpointCanceled) require.NoDirExists(t, dest) } diff --git a/sei-db/db_engine/types/types.go b/sei-db/db_engine/types/types.go index e4462949ec..3a2d9d0a13 100644 --- a/sei-db/db_engine/types/types.go +++ b/sei-db/db_engine/types/types.go @@ -180,8 +180,9 @@ type RollbackValidator interface { // The interfaces above are engine capabilities. Deciding when a checkpoint runs, // and what version it is labeled with, is coordination rather than engine -// behavior and lives in sei-db/management: CheckpointScheduler, -// ScheduleCheckpoint, SetCheckpointVersion and ErrCheckpointCanceled. +// behavior: sei-db/controller picks the height every store checkpoints at, and +// sei-db/state_db/ss/snapshot composes these capabilities into the checkpoint an +// SS member store performs. // --------------------------------------------------------------------------- // SS DB layer diff --git a/sei-db/state_db/sc/flatkv/checkpoint.go b/sei-db/state_db/sc/flatkv/checkpoint.go new file mode 100644 index 0000000000..19166df0e9 --- /dev/null +++ b/sei-db/state_db/sc/flatkv/checkpoint.go @@ -0,0 +1,79 @@ +package flatkv + +import ( + "github.com/sei-protocol/sei-chain/sei-db/controller" +) + +var _ controller.CheckpointableStore = (*CommitStore)(nil) + +// ScheduleCheckpoint records targetVersion as the version this store's next snapshot is written at. +// The snapshot itself is the one Commit already writes; this only fixes which block it happens on. +// +// A height at or below the committed version, a request on a read-only store, or a second request +// while one is already pending is ignored. +func (s *CommitStore) ScheduleCheckpoint(targetVersion int64) { + s.mu.RLock() + readOnly := s.readOnly + committed := s.committedVersion + s.mu.RUnlock() + + if readOnly || targetVersion <= committed { + return + } + s.pendingCheckpoint.CompareAndSwap(0, targetVersion) +} + +// LatestVersion returns the version this store has committed. +func (s *CommitStore) LatestVersion() int64 { + s.mu.RLock() + defer s.mu.RUnlock() + return s.committedVersion +} + +// CheckpointInProgress reports whether this store is currently writing a snapshot. +func (s *CommitStore) CheckpointInProgress() bool { + return s.snapshotInProgress.Load() +} + +// snapshotIfDue writes a snapshot when version is one this store owes a snapshot at. Two cadences can +// ask for one — the store-local SnapshotInterval, and a target a controller.CheckpointScheduler +// dispatched — and a version both name is one snapshot, not two. +// +// Commit calls this while holding the write lock, so the snapshot it writes is of exactly the block +// that just committed. +func (s *CommitStore) snapshotIfDue(version int64) { + scheduled := s.consumeCheckpointTarget(version) + periodic := s.config.SnapshotInterval > 0 && version%int64(s.config.SnapshotInterval) == 0 + if !scheduled && !periodic { + return + } + s.phaseTimer.SetPhase("commit_write_snapshot") + s.snapshotInProgress.Store(true) + defer s.snapshotInProgress.Store(false) + if err := s.WriteSnapshot(""); err != nil { + logger.Error("auto snapshot failed", "version", version, "err", err) + } +} + +// consumeCheckpointTarget reports whether version is the accepted checkpoint target, clearing the +// target once version has reached it. +// +// It clears on an overshoot as well as on a match: a target left set at a version this store has +// already passed is one no later commit can match. Overshooting is a bug rather than a race — Commit +// takes contiguous versions and a target is only accepted above the committed one — so it is +// reported rather than absorbed. +func (s *CommitStore) consumeCheckpointTarget(version int64) bool { + target := s.pendingCheckpoint.Load() + if target == 0 || version < target { + return false + } + if !s.pendingCheckpoint.CompareAndSwap(target, 0) { + return false + } + if version > target { + logger.Error("FlatKV passed a scheduled checkpoint version without snapshotting it", + "targetVersion", target, "committedVersion", version) + return false + } + return true +} diff --git a/sei-db/state_db/sc/flatkv/checkpoint_test.go b/sei-db/state_db/sc/flatkv/checkpoint_test.go new file mode 100644 index 0000000000..c6b3d2799b --- /dev/null +++ b/sei-db/state_db/sc/flatkv/checkpoint_test.go @@ -0,0 +1,114 @@ +package flatkv + +import ( + "sync" + "testing" + "time" + + "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/flatkv/config" + "github.com/stretchr/testify/require" +) + +// scheduledOnlyStore returns a store whose only source of snapshots is a dispatched checkpoint target. +// The periodic interval is off so a snapshot appearing is attributable to the scheduled path. +func scheduledOnlyStore(t *testing.T) *CommitStore { + t.Helper() + cfg := config.DefaultTestConfig(t) + cfg.SnapshotInterval = 0 + s, err := newCommitStoreWithWAL(t.Context(), cfg) + require.NoError(t, err) + require.NoError(t, s.LoadLatest()) + t.Cleanup(func() { _ = s.Close() }) + return s +} + +// The target names the block the snapshot is taken at, not "soon": the blocks before it pass without +// one, and the snapshot lands on exactly the version that was asked for. +func TestScheduledCheckpointSnapshotsAtTheTargetVersion(t *testing.T) { + s := scheduledOnlyStore(t) + + s.ScheduleCheckpoint(3) + require.Equal(t, int64(3), s.pendingCheckpoint.Load()) + + commitAndCheck(t, s) + commitAndCheck(t, s) + require.Equal(t, int64(3), s.pendingCheckpoint.Load(), "the target block has not arrived yet") + require.NotContains(t, snapshotVersions(t, s.flatkvDir()), int64(2)) + + commitAndCheck(t, s) + require.Zero(t, s.pendingCheckpoint.Load()) + require.Contains(t, snapshotVersions(t, s.flatkvDir()), int64(3)) +} + +// A target the store has already passed is ignored: accepting one would snapshot a height Commit +// will never see again. +func TestScheduleCheckpointIgnoresATargetAtOrBelowTheCommittedVersion(t *testing.T) { + s := scheduledOnlyStore(t) + commitAndCheck(t, s) + + s.ScheduleCheckpoint(1) + s.ScheduleCheckpoint(0) + require.Zero(t, s.pendingCheckpoint.Load()) +} + +func TestScheduleCheckpointIgnoresASecondTargetWhileOneIsPending(t *testing.T) { + s := scheduledOnlyStore(t) + + s.ScheduleCheckpoint(5) + s.ScheduleCheckpoint(6) + require.Equal(t, int64(5), s.pendingCheckpoint.Load()) +} + +func TestLatestVersionReportsTheCommittedVersion(t *testing.T) { + s := scheduledOnlyStore(t) + + require.Equal(t, int64(0), s.LatestVersion()) + commitAndCheck(t, s) + require.Equal(t, int64(1), s.LatestVersion()) +} + +// A scheduler queries these three methods from its own goroutine while the commit path holds the +// store's write lock across a snapshot. This runs the two against each other to pin the lock order; +// the test completing is the assertion, because the failure it looks for is a wedged commit loop. +// +// The querying is done directly rather than by running a real scheduler, which makes it both denser +// than the scheduler's poll interval and independent of it. Which versions end up snapshotted depends +// on when a query lands, so the only thing asserted about the result is the invariant that holds +// regardless: every snapshot sits on an interval boundary. +func TestScheduleCheckpointQueriesConcurrentWithCommits(t *testing.T) { + const interval = 4 + + s := scheduledOnlyStore(t) + + stop := make(chan struct{}) + var polling sync.WaitGroup + polling.Add(1) + go func() { + defer polling.Done() + for { + select { + case <-stop: + return + default: + } + if !s.CheckpointInProgress() { + version := s.LatestVersion() + s.ScheduleCheckpoint((version/interval + 1) * interval) + } + time.Sleep(50 * time.Microsecond) + } + }() + + for i := 0; i < 20; i++ { + commitAndCheck(t, s) + } + close(stop) + polling.Wait() + + for _, version := range snapshotVersions(t, s.flatkvDir()) { + if version == 0 { + continue // the empty snapshot a fresh store is initialized with + } + require.Zero(t, version%interval, "snapshot %d is not on an interval boundary", version) + } +} diff --git a/sei-db/state_db/sc/flatkv/store.go b/sei-db/state_db/sc/flatkv/store.go index e5985d4c1d..8d64b1ca92 100644 --- a/sei-db/state_db/sc/flatkv/store.go +++ b/sei-db/state_db/sc/flatkv/store.go @@ -10,6 +10,7 @@ import ( "runtime" "strings" "sync" + "sync/atomic" "time" "github.com/zbiljic/go-filelock" @@ -147,6 +148,14 @@ type CommitStore struct { lastSnapshotTime time.Time + // pendingCheckpoint is the version a controller.CheckpointScheduler asked this store to snapshot + // at, 0 when none is pending. Atomic rather than under mu because Commit consumes it under the + // write lock and ScheduleCheckpoint sets it from the scheduler's goroutine. + pendingCheckpoint atomic.Int64 + + // snapshotInProgress is true while WriteSnapshot is running for a due checkpoint. + snapshotInProgress atomic.Bool + // File lock prevents multiple processes from opening the same DB. fileLock filelock.TryLockerSafe diff --git a/sei-db/state_db/sc/flatkv/store_write.go b/sei-db/state_db/sc/flatkv/store_write.go index 60bd49a596..99c4cdae97 100644 --- a/sei-db/state_db/sc/flatkv/store_write.go +++ b/sei-db/state_db/sc/flatkv/store_write.go @@ -112,12 +112,7 @@ func (s *CommitStore) Commit(version int64) (committed int64, err error) { recordPendingWrites(s.ctx, miscDBDir, 0) // Periodic snapshot so WAL stays bounded and restarts are fast. - if s.config.SnapshotInterval > 0 && version%int64(s.config.SnapshotInterval) == 0 { - s.phaseTimer.SetPhase("commit_write_snapshot") - if err := s.WriteSnapshot(""); err != nil { - logger.Error("auto snapshot failed", "version", version, "err", err) - } - } + s.snapshotIfDue(version) // Best-effort WAL truncation, throttled to amortize ReadDir cost. if version%1000 == 0 { diff --git a/sei-db/state_db/ss/composite/checkpoint.go b/sei-db/state_db/ss/composite/checkpoint.go new file mode 100644 index 0000000000..4a9c999e07 --- /dev/null +++ b/sei-db/state_db/ss/composite/checkpoint.go @@ -0,0 +1,32 @@ +package composite + +import ( + "github.com/sei-protocol/sei-chain/sei-db/controller" +) + +var _ controller.CheckpointableStore = (*CompositeStateStore)(nil) + +// ScheduleCheckpoint records targetVersion as the height the next SS snapshot is taken at. The +// snapshot itself is the one the write path already stages and publishes across the members; this +// only fixes which height that happens on. +// +// A store with snapshots disabled, stopped, or already holding a target ignores the request, as does +// a height at or below the last snapshot this coordinator requested. +func (s *CompositeStateStore) ScheduleCheckpoint(targetVersion int64) { + if s.snapshotMgr == nil { + return + } + s.snapshotMgr.acceptTarget(targetVersion) +} + +// LatestVersion returns the newest version this store has committed. It reports the same height as +// GetLatestVersion, which is the state store's own interface; this is the name the checkpoint +// scheduler reads it under. +func (s *CompositeStateStore) LatestVersion() int64 { + return s.GetLatestVersion() +} + +// CheckpointInProgress reports whether a snapshot is currently being written. +func (s *CompositeStateStore) CheckpointInProgress() bool { + return s.snapshotMgr.checkpointInProgress() +} diff --git a/sei-db/state_db/ss/composite/snapshot.go b/sei-db/state_db/ss/composite/snapshot.go index 3c404470e2..5143db8528 100644 --- a/sei-db/state_db/ss/composite/snapshot.go +++ b/sei-db/state_db/ss/composite/snapshot.go @@ -7,7 +7,6 @@ import ( "sync/atomic" "time" - "github.com/sei-protocol/sei-chain/sei-db/controller" sssnapshot "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/snapshot" ) @@ -75,6 +74,9 @@ type snapshotCoordinator struct { // floor carries the newest height every member holds to the members' own retention, which counts // only its own directories and would otherwise let an unpaired newer height crowd it out. floor *sssnapshot.Floor + // target is the height a controller.CheckpointScheduler asked for, 0 when none is outstanding. + // Atomic rather than under mu because the write path consults it for every version. + target atomic.Int64 mu sync.Mutex // lastRequested is the newest label already requested or present in any @@ -150,6 +152,9 @@ func (s *CompositeStateStore) startSnapshotManager(members []snapshotMember, flo func (c *snapshotCoordinator) stop() { c.mu.Lock() c.stopped = true + // Dropped rather than serviced: no later block will arrive to reach it, and a scheduler polling + // checkpointInProgress across shutdown would otherwise be told a checkpoint is still coming. + c.target.Store(0) c.mu.Unlock() c.scheduling.Wait() c.publishing.Wait() @@ -161,11 +166,14 @@ func (c *snapshotCoordinator) isRunning() bool { return !c.stopped } -// maybeSnapshot takes a snapshot when version lands on an interval boundary. -// It is called from the write path for every version, so the common case is the -// modulo test and nothing else. +// maybeSnapshot takes a snapshot when version is a height this coordinator owes one at. It is called +// from the write path for every version, so the common case is the due test and nothing else. func (c *snapshotCoordinator) maybeSnapshot(version int64) { - if c == nil || version <= 0 || c.interval <= 0 || version%c.interval != 0 { + if c == nil || version <= 0 { + return + } + due, scheduled := c.due(version) + if !due { return } now := time.Now() @@ -179,7 +187,7 @@ func (c *snapshotCoordinator) maybeSnapshot(version int64) { // A repeated commit-path call is expected and is not a skipped attempt. case c.inFlight: skipReason = "in_flight" - case !c.lastRequestAt.IsZero() && now.Sub(c.lastRequestAt) < c.minTime: + case !scheduled && !c.lastRequestAt.IsZero() && now.Sub(c.lastRequestAt) < c.minTime: skipReason = "minimum_time_interval" default: c.lastRequested = version @@ -219,6 +227,70 @@ func (c *snapshotCoordinator) maybeSnapshot(version int64) { } } +// due reports whether version is a height this coordinator owes a snapshot at, and whether it was a +// scheduler that asked. Two cadences can ask — the configured SnapshotInterval, and a target a +// controller.CheckpointScheduler dispatched — and a height both name is one snapshot, not two. +// +// Which one asked decides whether the minimum-time gate applies. That gate paces the configured +// interval; applying it to a dispatched target would drop the one height the scheduler's other stores +// are checkpointing at, which is the whole guarantee it exists to make. Skipping it loses no pacing, +// because a scheduler gates its own dispatches: a dispatched target arrives already paced. +func (c *snapshotCoordinator) due(version int64) (due bool, scheduled bool) { + scheduled = c.consumeTarget(version) + return scheduled || (c.interval > 0 && version%c.interval == 0), scheduled +} + +// consumeTarget reports whether version is the height a scheduler asked for, clearing the target once +// version has reached it. With no target outstanding it is a single atomic load, which is what keeps +// it callable for every version on the write path. +// +// It clears on an overshoot as well as on a match: a target left set at a height already past is one +// no later block can match. Overshooting means the write path skipped the height between the target +// being accepted and the block arriving, so it is reported rather than absorbed. +func (c *snapshotCoordinator) consumeTarget(version int64) bool { + target := c.target.Load() + if target == 0 || version < target { + return false + } + if !c.target.CompareAndSwap(target, 0) { + return false + } + if version > target { + logger.Error("state store passed a scheduled checkpoint height without snapshotting it", + "targetVersion", target, "version", version) + return false + } + return true +} + +// acceptTarget records version as the height this coordinator owes a scheduled snapshot at. It holds +// one target at a time. A height already requested, a second request while one is pending, or a +// request after stop is ignored. +func (c *snapshotCoordinator) acceptTarget(version int64) { + c.mu.Lock() + defer c.mu.Unlock() + if c.stopped { + return + } + if c.target.Load() != 0 { + return + } + if version <= c.lastRequested { + return + } + c.target.Store(version) +} + +// checkpointInProgress reports whether a snapshot is currently being staged or published. +func (c *snapshotCoordinator) checkpointInProgress() bool { + if c == nil { + return false + } + c.mu.Lock() + defer c.mu.Unlock() + return c.inFlight +} + func (c *snapshotCoordinator) finishSnapshot() { c.mu.Lock() c.inFlight = false @@ -251,7 +323,7 @@ func (c *snapshotCoordinator) requestSnapshot(version int64, start time.Time) er return c.isRunning() && !canceled.Load() } - report := controller.FanIn(len(c.members), func(err error) { + report := sssnapshot.FanIn(len(c.members), func(err error) { c.startPublish(version, staged, err, start) }) for i, member := range c.members { @@ -295,7 +367,7 @@ func (c *snapshotCoordinator) startPublish( defer c.publishing.Done() defer c.finishSnapshot() if checkpointErr != nil { - if errors.Is(checkpointErr, controller.ErrCheckpointCanceled) { + if errors.Is(checkpointErr, sssnapshot.ErrCheckpointCanceled) { sssnapshot.RecordCompletion(start, "canceled") } else { sssnapshot.RecordCompletion(start, "failure") diff --git a/sei-db/state_db/ss/composite/snapshot_test.go b/sei-db/state_db/ss/composite/snapshot_test.go index 8ead3909ec..75fa6b02ef 100644 --- a/sei-db/state_db/ss/composite/snapshot_test.go +++ b/sei-db/state_db/ss/composite/snapshot_test.go @@ -8,7 +8,6 @@ import ( "time" "github.com/sei-protocol/sei-chain/sei-db/config" - "github.com/sei-protocol/sei-chain/sei-db/controller" "github.com/sei-protocol/sei-chain/sei-db/proto" "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/evm" sssnapshot "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/snapshot" @@ -31,7 +30,7 @@ func (s *controlledSnapshotScheduler) ScheduleCheckpoint( ) { s.pending <- func() { if !shouldRun() { - done(controller.ErrCheckpointCanceled) + done(sssnapshot.ErrCheckpointCanceled) return } if s.fail { @@ -328,13 +327,13 @@ func openTestManagerWithRetention( t.Helper() source := t.TempDir() manager, err := sssnapshot.Open(sssnapshot.Config{ - Name: name, - Root: root, - SourceDirs: []string{source}, - Backend: config.PebbleDBBackend, - KeepRecent: keepRecent, - Scheduler: scheduler, - Floor: floor, + Name: name, + Root: root, + SourceDirs: []string{source}, + Backend: config.PebbleDBBackend, + KeepRecent: keepRecent, + Checkpointer: scheduler, + Floor: floor, }) require.NoError(t, err) return manager diff --git a/sei-db/state_db/ss/cosmos/store.go b/sei-db/state_db/ss/cosmos/store.go index d2d076deed..35b550d452 100644 --- a/sei-db/state_db/ss/cosmos/store.go +++ b/sei-db/state_db/ss/cosmos/store.go @@ -7,7 +7,6 @@ import ( dbm "github.com/tendermint/tm-db" "github.com/sei-protocol/sei-chain/sei-db/config" - "github.com/sei-protocol/sei-chain/sei-db/controller" "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" "github.com/sei-protocol/sei-chain/sei-db/proto" sssnapshot "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/snapshot" @@ -103,15 +102,15 @@ func (s *CosmosStateStore) Close() error { } func (s *CosmosStateStore) SupportsCheckpoint() bool { - return controller.SupportsCheckpoint(s.db) + return sssnapshot.SupportsCheckpoint(s.db) } func (s *CosmosStateStore) ScheduleCheckpoint(destDir string, shouldRun func() bool, done func(error)) { - controller.ScheduleCheckpoint(s.db, destDir, shouldRun, done) + sssnapshot.ScheduleCheckpoint(s.db, destDir, shouldRun, done) } func (s *CosmosStateStore) SetCheckpointVersion(destDir string, version int64) error { - return controller.SetCheckpointVersion(s.db, destDir, version) + return sssnapshot.SetCheckpointVersion(s.db, destDir, version) } func (s *CosmosStateStore) StartSnapshots( @@ -127,7 +126,7 @@ func (s *CosmosStateStore) StartSnapshots( Backend: ssConfig.Backend, KeepRecent: ssConfig.SnapshotKeepRecent, ExternalPruning: ssConfig.ExternalPruning, - Scheduler: s, + Checkpointer: s, Floor: floor, }) if err != nil { diff --git a/sei-db/state_db/ss/evm/store.go b/sei-db/state_db/ss/evm/store.go index 1e98d1e9d2..02b86d299f 100644 --- a/sei-db/state_db/ss/evm/store.go +++ b/sei-db/state_db/ss/evm/store.go @@ -12,7 +12,6 @@ import ( commonevm "github.com/sei-protocol/sei-chain/sei-db/common/keys" "github.com/sei-protocol/sei-chain/sei-db/config" - "github.com/sei-protocol/sei-chain/sei-db/controller" "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" "github.com/sei-protocol/sei-chain/sei-db/proto" "github.com/sei-protocol/sei-chain/sei-db/state_db/ss/backend" @@ -427,7 +426,7 @@ func (s *EVMStateStore) Close() error { func (s *EVMStateStore) SupportsCheckpoint() bool { for _, db := range s.managedDBs { - if !controller.SupportsCheckpoint(db) { + if !sssnapshot.SupportsCheckpoint(db) { return false } } @@ -455,7 +454,7 @@ func (s *EVMStateStore) ScheduleCheckpoint(destDir string, shouldRun func() bool done(errors.New("EVM state store has no managed DB to checkpoint")) return } - controller.ScheduleCheckpoint(db, destDir, shouldRun, done) + sssnapshot.ScheduleCheckpoint(db, destDir, shouldRun, done) return } @@ -465,10 +464,10 @@ func (s *EVMStateStore) ScheduleCheckpoint(destDir string, shouldRun func() bool } storeTypes := AllEVMStoreTypes() - report := controller.FanIn(len(storeTypes), done) + report := sssnapshot.FanIn(len(storeTypes), done) for _, storeType := range storeTypes { name := StoreTypeName(storeType) - controller.ScheduleCheckpoint(s.subDBs[storeType], subDBPath(destDir, storeType), shouldRun, func(err error) { + sssnapshot.ScheduleCheckpoint(s.subDBs[storeType], subDBPath(destDir, storeType), shouldRun, func(err error) { if err != nil { err = fmt.Errorf("checkpoint EVM sub-DB %s: %w", name, err) } @@ -483,11 +482,11 @@ func (s *EVMStateStore) SetCheckpointVersion(destDir string, version int64) erro if db == nil { return errors.New("EVM state store has no managed DB to stamp") } - return controller.SetCheckpointVersion(db, destDir, version) + return sssnapshot.SetCheckpointVersion(db, destDir, version) } for _, storeType := range AllEVMStoreTypes() { name := StoreTypeName(storeType) - if err := controller.SetCheckpointVersion(s.subDBs[storeType], subDBPath(destDir, storeType), version); err != nil { + if err := sssnapshot.SetCheckpointVersion(s.subDBs[storeType], subDBPath(destDir, storeType), version); err != nil { return fmt.Errorf("set EVM sub-DB %s checkpoint version: %w", name, err) } } @@ -506,7 +505,7 @@ func (s *EVMStateStore) StartSnapshots( Backend: ssConfig.Backend, KeepRecent: ssConfig.SnapshotKeepRecent, ExternalPruning: ssConfig.ExternalPruning, - Scheduler: s, + Checkpointer: s, Floor: floor, }) if err != nil { diff --git a/sei-db/state_db/ss/snapshot/checkpoint.go b/sei-db/state_db/ss/snapshot/checkpoint.go new file mode 100644 index 0000000000..c121e4e256 --- /dev/null +++ b/sei-db/state_db/ss/snapshot/checkpoint.go @@ -0,0 +1,92 @@ +package snapshot + +import ( + "errors" + "fmt" + "sync" + + "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" +) + +// Checkpointer is the backend side of an SS member snapshot: the Manager owns the +// directories a snapshot occupies, and a Checkpointer writes the database into one. +// +// The engine-side capabilities this builds on — types.Checkpointable, +// types.DrainBarrier and types.CheckpointVersionSetter — stay with the engines that +// implement them. An SS member store implements Checkpointer by composing them over +// the one or more engines it manages. +type Checkpointer interface { + SupportsCheckpoint() bool + ScheduleCheckpoint(destDir string, shouldRun func() bool, done func(error)) + SetCheckpointVersion(destDir string, version int64) error +} + +// ErrCheckpointCanceled reports that a queued checkpoint was canceled before +// it started. +var ErrCheckpointCanceled = errors.New("state store checkpoint canceled") + +// SupportsCheckpoint reports whether db carries every engine capability a scheduled checkpoint needs. +// A store built from several engines answers for the set: one engine short of the capabilities makes +// the whole snapshot unpublishable. +func SupportsCheckpoint(db types.StateStore) bool { + _, checkpointable := db.(types.Checkpointable) + _, barrier := db.(types.DrainBarrier) + _, versionSetter := db.(types.CheckpointVersionSetter) + return checkpointable && barrier && versionSetter +} + +// FanIn returns a report callback for n parallel branches. Each branch calls it once, and the last +// call passes done the first error any branch reported, or nil. +func FanIn(n int, done func(error)) func(error) { + var ( + mu sync.Mutex + remaining = n + firstErr error + ) + return func(err error) { + mu.Lock() + if err != nil && firstErr == nil { + firstErr = err + } + remaining-- + isLast := remaining == 0 + // Read under the lock: a peer branch may report between the unlock and the call to done. + outcome := firstErr + mu.Unlock() + if isLast { + done(outcome) + } + } +} + +// ScheduleCheckpoint checkpoints an engine after all writes already enqueued +// on it have been applied. +func ScheduleCheckpoint(db types.StateStore, destDir string, shouldRun func() bool, done func(error)) { + cp, ok := db.(types.Checkpointable) + if !ok { + done(fmt.Errorf("state store backend %T does not support checkpoints", db)) + return + } + barrier, ok := db.(types.DrainBarrier) + if !ok { + done(fmt.Errorf("state store backend %T does not support ordered checkpoint barriers", db)) + return + } + barrier.ScheduleAtDrain(func() { + if shouldRun != nil && !shouldRun() { + done(ErrCheckpointCanceled) + return + } + done(cp.Checkpoint(destDir)) + }) +} + +// SetCheckpointVersion makes a completed checkpoint self-describing without +// changing the live database. +func SetCheckpointVersion(db types.StateStore, destDir string, version int64) error { + setter, ok := db.(types.CheckpointVersionSetter) + if !ok { + return fmt.Errorf("state store backend %T cannot set checkpoint version", db) + } + return setter.SetCheckpointVersion(destDir, version) +} diff --git a/sei-db/state_db/ss/snapshot/manager.go b/sei-db/state_db/ss/snapshot/manager.go index f6a06aa560..aa99ca6816 100644 --- a/sei-db/state_db/ss/snapshot/manager.go +++ b/sei-db/state_db/ss/snapshot/manager.go @@ -14,7 +14,6 @@ import ( "time" "github.com/sei-protocol/sei-chain/sei-db/common/utils" - "github.com/sei-protocol/sei-chain/sei-db/controller" "github.com/sei-protocol/seilog" ) @@ -42,7 +41,7 @@ type Config struct { Backend string KeepRecent int ExternalPruning bool - Scheduler controller.CheckpointScheduler + Checkpointer Checkpointer // Floor names a height this member's retention must keep. Leave it nil when the member is the only // one that has to hold the height a restore starts from. Floor *Floor @@ -115,7 +114,7 @@ type Manager struct { backend string keepRecent int externalPruning bool - scheduler controller.CheckpointScheduler + checkpointer Checkpointer floor *Floor snapshotSizes map[int64]int64 @@ -187,10 +186,10 @@ func ListSnapshotVersions(root string) ([]int64, error) { // Open prepares a snapshot root and returns a Manager for one SS member. func Open(cfg Config) (*Manager, error) { - if cfg.Scheduler == nil { - return nil, fmt.Errorf("%s snapshot scheduler is nil", cfg.Name) + if cfg.Checkpointer == nil { + return nil, fmt.Errorf("%s snapshot checkpointer is nil", cfg.Name) } - if !cfg.Scheduler.SupportsCheckpoint() { + if !cfg.Checkpointer.SupportsCheckpoint() { return nil, fmt.Errorf("%s backend %q does not support checkpoints", cfg.Name, cfg.Backend) } if err := verifyHardlinks(cfg.Root, cfg.SourceDirs); err != nil { @@ -202,7 +201,7 @@ func Open(cfg Config) (*Manager, error) { backend: cfg.Backend, keepRecent: cfg.KeepRecent, externalPruning: cfg.ExternalPruning, - scheduler: cfg.Scheduler, + checkpointer: cfg.Checkpointer, floor: cfg.Floor, snapshotSizes: map[int64]int64{}, } @@ -276,7 +275,7 @@ func (m *Manager) Schedule(staged *Staged, shouldRun func() bool, done func(erro done(fmt.Errorf("%s staged snapshot belongs to a different manager", m.name)) return } - m.scheduler.ScheduleCheckpoint(staged.tmpDir, shouldRun, done) + m.checkpointer.ScheduleCheckpoint(staged.tmpDir, shouldRun, done) } func (m *Manager) Commit(staged *Staged) error { @@ -297,7 +296,7 @@ func (m *Manager) Commit(staged *Staged) error { defer m.publishMu.Unlock() defer m.prune() - if err := m.scheduler.SetCheckpointVersion(staged.tmpDir, staged.version); err != nil { + if err := m.checkpointer.SetCheckpointVersion(staged.tmpDir, staged.version); err != nil { _ = os.RemoveAll(staged.tmpDir) return fmt.Errorf("set %s snapshot version: %w", m.name, err) } @@ -484,7 +483,7 @@ func (m *Manager) prune() { } func (m *Manager) pruneWALToOldestSnapshot() { - pruner, ok := m.scheduler.(snapshotWALPruner) + pruner, ok := m.checkpointer.(snapshotWALPruner) if !ok { return } diff --git a/sei-db/state_db/ss/snapshot/manager_test.go b/sei-db/state_db/ss/snapshot/manager_test.go index a2f51160cb..8a4fdc12b8 100644 --- a/sei-db/state_db/ss/snapshot/manager_test.go +++ b/sei-db/state_db/ss/snapshot/manager_test.go @@ -7,7 +7,6 @@ import ( "testing" "github.com/sei-protocol/sei-chain/sei-db/config" - "github.com/sei-protocol/sei-chain/sei-db/controller" "github.com/stretchr/testify/require" ) @@ -23,7 +22,7 @@ func (*controlledScheduler) SupportsCheckpoint() bool { func (s *controlledScheduler) ScheduleCheckpoint(destDir string, shouldRun func() bool, done func(error)) { s.pending <- func() { if !shouldRun() { - done(controller.ErrCheckpointCanceled) + done(ErrCheckpointCanceled) return } if s.fail { @@ -146,11 +145,11 @@ func TestOpenClearsLeftoverHardlinkProbes(t *testing.T) { require.NoError(t, os.WriteFile(filepath.Join(root, linkProbeName), nil, 0o600)) _, err := Open(Config{ - Name: "test", - Root: root, - SourceDirs: []string{source}, - Backend: config.PebbleDBBackend, - Scheduler: &controlledScheduler{pending: make(chan func(), 1)}, + Name: "test", + Root: root, + SourceDirs: []string{source}, + Backend: config.PebbleDBBackend, + Checkpointer: &controlledScheduler{pending: make(chan func(), 1)}, }) require.NoError(t, err) @@ -207,7 +206,7 @@ func openManagerWithFloor( Backend: config.PebbleDBBackend, KeepRecent: keepRecent, ExternalPruning: external, - Scheduler: scheduler, + Checkpointer: scheduler, Floor: floor, }) require.NoError(t, err) From 445e09048989a49b617343fc3ba3f5013d4a5c1e Mon Sep 17 00:00:00 2001 From: YimingZang Date: Wed, 26 Aug 2026 13:18:15 -0700 Subject: [PATCH 02/11] Not implement yet --- sei-db/state_db/sc/flatkv/checkpoint.go | 79 ------------- sei-db/state_db/sc/flatkv/checkpoint_test.go | 114 ------------------- sei-db/state_db/sc/flatkv/store.go | 9 -- sei-db/state_db/sc/flatkv/store_write.go | 7 +- sei-db/state_db/ss/composite/checkpoint.go | 32 ------ sei-db/state_db/ss/composite/snapshot.go | 83 +------------- 6 files changed, 11 insertions(+), 313 deletions(-) delete mode 100644 sei-db/state_db/sc/flatkv/checkpoint.go delete mode 100644 sei-db/state_db/sc/flatkv/checkpoint_test.go delete mode 100644 sei-db/state_db/ss/composite/checkpoint.go diff --git a/sei-db/state_db/sc/flatkv/checkpoint.go b/sei-db/state_db/sc/flatkv/checkpoint.go deleted file mode 100644 index 19166df0e9..0000000000 --- a/sei-db/state_db/sc/flatkv/checkpoint.go +++ /dev/null @@ -1,79 +0,0 @@ -package flatkv - -import ( - "github.com/sei-protocol/sei-chain/sei-db/controller" -) - -var _ controller.CheckpointableStore = (*CommitStore)(nil) - -// ScheduleCheckpoint records targetVersion as the version this store's next snapshot is written at. -// The snapshot itself is the one Commit already writes; this only fixes which block it happens on. -// -// A height at or below the committed version, a request on a read-only store, or a second request -// while one is already pending is ignored. -func (s *CommitStore) ScheduleCheckpoint(targetVersion int64) { - s.mu.RLock() - readOnly := s.readOnly - committed := s.committedVersion - s.mu.RUnlock() - - if readOnly || targetVersion <= committed { - return - } - s.pendingCheckpoint.CompareAndSwap(0, targetVersion) -} - -// LatestVersion returns the version this store has committed. -func (s *CommitStore) LatestVersion() int64 { - s.mu.RLock() - defer s.mu.RUnlock() - return s.committedVersion -} - -// CheckpointInProgress reports whether this store is currently writing a snapshot. -func (s *CommitStore) CheckpointInProgress() bool { - return s.snapshotInProgress.Load() -} - -// snapshotIfDue writes a snapshot when version is one this store owes a snapshot at. Two cadences can -// ask for one — the store-local SnapshotInterval, and a target a controller.CheckpointScheduler -// dispatched — and a version both name is one snapshot, not two. -// -// Commit calls this while holding the write lock, so the snapshot it writes is of exactly the block -// that just committed. -func (s *CommitStore) snapshotIfDue(version int64) { - scheduled := s.consumeCheckpointTarget(version) - periodic := s.config.SnapshotInterval > 0 && version%int64(s.config.SnapshotInterval) == 0 - if !scheduled && !periodic { - return - } - s.phaseTimer.SetPhase("commit_write_snapshot") - s.snapshotInProgress.Store(true) - defer s.snapshotInProgress.Store(false) - if err := s.WriteSnapshot(""); err != nil { - logger.Error("auto snapshot failed", "version", version, "err", err) - } -} - -// consumeCheckpointTarget reports whether version is the accepted checkpoint target, clearing the -// target once version has reached it. -// -// It clears on an overshoot as well as on a match: a target left set at a version this store has -// already passed is one no later commit can match. Overshooting is a bug rather than a race — Commit -// takes contiguous versions and a target is only accepted above the committed one — so it is -// reported rather than absorbed. -func (s *CommitStore) consumeCheckpointTarget(version int64) bool { - target := s.pendingCheckpoint.Load() - if target == 0 || version < target { - return false - } - if !s.pendingCheckpoint.CompareAndSwap(target, 0) { - return false - } - if version > target { - logger.Error("FlatKV passed a scheduled checkpoint version without snapshotting it", - "targetVersion", target, "committedVersion", version) - return false - } - return true -} diff --git a/sei-db/state_db/sc/flatkv/checkpoint_test.go b/sei-db/state_db/sc/flatkv/checkpoint_test.go deleted file mode 100644 index c6b3d2799b..0000000000 --- a/sei-db/state_db/sc/flatkv/checkpoint_test.go +++ /dev/null @@ -1,114 +0,0 @@ -package flatkv - -import ( - "sync" - "testing" - "time" - - "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/flatkv/config" - "github.com/stretchr/testify/require" -) - -// scheduledOnlyStore returns a store whose only source of snapshots is a dispatched checkpoint target. -// The periodic interval is off so a snapshot appearing is attributable to the scheduled path. -func scheduledOnlyStore(t *testing.T) *CommitStore { - t.Helper() - cfg := config.DefaultTestConfig(t) - cfg.SnapshotInterval = 0 - s, err := newCommitStoreWithWAL(t.Context(), cfg) - require.NoError(t, err) - require.NoError(t, s.LoadLatest()) - t.Cleanup(func() { _ = s.Close() }) - return s -} - -// The target names the block the snapshot is taken at, not "soon": the blocks before it pass without -// one, and the snapshot lands on exactly the version that was asked for. -func TestScheduledCheckpointSnapshotsAtTheTargetVersion(t *testing.T) { - s := scheduledOnlyStore(t) - - s.ScheduleCheckpoint(3) - require.Equal(t, int64(3), s.pendingCheckpoint.Load()) - - commitAndCheck(t, s) - commitAndCheck(t, s) - require.Equal(t, int64(3), s.pendingCheckpoint.Load(), "the target block has not arrived yet") - require.NotContains(t, snapshotVersions(t, s.flatkvDir()), int64(2)) - - commitAndCheck(t, s) - require.Zero(t, s.pendingCheckpoint.Load()) - require.Contains(t, snapshotVersions(t, s.flatkvDir()), int64(3)) -} - -// A target the store has already passed is ignored: accepting one would snapshot a height Commit -// will never see again. -func TestScheduleCheckpointIgnoresATargetAtOrBelowTheCommittedVersion(t *testing.T) { - s := scheduledOnlyStore(t) - commitAndCheck(t, s) - - s.ScheduleCheckpoint(1) - s.ScheduleCheckpoint(0) - require.Zero(t, s.pendingCheckpoint.Load()) -} - -func TestScheduleCheckpointIgnoresASecondTargetWhileOneIsPending(t *testing.T) { - s := scheduledOnlyStore(t) - - s.ScheduleCheckpoint(5) - s.ScheduleCheckpoint(6) - require.Equal(t, int64(5), s.pendingCheckpoint.Load()) -} - -func TestLatestVersionReportsTheCommittedVersion(t *testing.T) { - s := scheduledOnlyStore(t) - - require.Equal(t, int64(0), s.LatestVersion()) - commitAndCheck(t, s) - require.Equal(t, int64(1), s.LatestVersion()) -} - -// A scheduler queries these three methods from its own goroutine while the commit path holds the -// store's write lock across a snapshot. This runs the two against each other to pin the lock order; -// the test completing is the assertion, because the failure it looks for is a wedged commit loop. -// -// The querying is done directly rather than by running a real scheduler, which makes it both denser -// than the scheduler's poll interval and independent of it. Which versions end up snapshotted depends -// on when a query lands, so the only thing asserted about the result is the invariant that holds -// regardless: every snapshot sits on an interval boundary. -func TestScheduleCheckpointQueriesConcurrentWithCommits(t *testing.T) { - const interval = 4 - - s := scheduledOnlyStore(t) - - stop := make(chan struct{}) - var polling sync.WaitGroup - polling.Add(1) - go func() { - defer polling.Done() - for { - select { - case <-stop: - return - default: - } - if !s.CheckpointInProgress() { - version := s.LatestVersion() - s.ScheduleCheckpoint((version/interval + 1) * interval) - } - time.Sleep(50 * time.Microsecond) - } - }() - - for i := 0; i < 20; i++ { - commitAndCheck(t, s) - } - close(stop) - polling.Wait() - - for _, version := range snapshotVersions(t, s.flatkvDir()) { - if version == 0 { - continue // the empty snapshot a fresh store is initialized with - } - require.Zero(t, version%interval, "snapshot %d is not on an interval boundary", version) - } -} diff --git a/sei-db/state_db/sc/flatkv/store.go b/sei-db/state_db/sc/flatkv/store.go index 8d64b1ca92..e5985d4c1d 100644 --- a/sei-db/state_db/sc/flatkv/store.go +++ b/sei-db/state_db/sc/flatkv/store.go @@ -10,7 +10,6 @@ import ( "runtime" "strings" "sync" - "sync/atomic" "time" "github.com/zbiljic/go-filelock" @@ -148,14 +147,6 @@ type CommitStore struct { lastSnapshotTime time.Time - // pendingCheckpoint is the version a controller.CheckpointScheduler asked this store to snapshot - // at, 0 when none is pending. Atomic rather than under mu because Commit consumes it under the - // write lock and ScheduleCheckpoint sets it from the scheduler's goroutine. - pendingCheckpoint atomic.Int64 - - // snapshotInProgress is true while WriteSnapshot is running for a due checkpoint. - snapshotInProgress atomic.Bool - // File lock prevents multiple processes from opening the same DB. fileLock filelock.TryLockerSafe diff --git a/sei-db/state_db/sc/flatkv/store_write.go b/sei-db/state_db/sc/flatkv/store_write.go index 99c4cdae97..60bd49a596 100644 --- a/sei-db/state_db/sc/flatkv/store_write.go +++ b/sei-db/state_db/sc/flatkv/store_write.go @@ -112,7 +112,12 @@ func (s *CommitStore) Commit(version int64) (committed int64, err error) { recordPendingWrites(s.ctx, miscDBDir, 0) // Periodic snapshot so WAL stays bounded and restarts are fast. - s.snapshotIfDue(version) + if s.config.SnapshotInterval > 0 && version%int64(s.config.SnapshotInterval) == 0 { + s.phaseTimer.SetPhase("commit_write_snapshot") + if err := s.WriteSnapshot(""); err != nil { + logger.Error("auto snapshot failed", "version", version, "err", err) + } + } // Best-effort WAL truncation, throttled to amortize ReadDir cost. if version%1000 == 0 { diff --git a/sei-db/state_db/ss/composite/checkpoint.go b/sei-db/state_db/ss/composite/checkpoint.go deleted file mode 100644 index 4a9c999e07..0000000000 --- a/sei-db/state_db/ss/composite/checkpoint.go +++ /dev/null @@ -1,32 +0,0 @@ -package composite - -import ( - "github.com/sei-protocol/sei-chain/sei-db/controller" -) - -var _ controller.CheckpointableStore = (*CompositeStateStore)(nil) - -// ScheduleCheckpoint records targetVersion as the height the next SS snapshot is taken at. The -// snapshot itself is the one the write path already stages and publishes across the members; this -// only fixes which height that happens on. -// -// A store with snapshots disabled, stopped, or already holding a target ignores the request, as does -// a height at or below the last snapshot this coordinator requested. -func (s *CompositeStateStore) ScheduleCheckpoint(targetVersion int64) { - if s.snapshotMgr == nil { - return - } - s.snapshotMgr.acceptTarget(targetVersion) -} - -// LatestVersion returns the newest version this store has committed. It reports the same height as -// GetLatestVersion, which is the state store's own interface; this is the name the checkpoint -// scheduler reads it under. -func (s *CompositeStateStore) LatestVersion() int64 { - return s.GetLatestVersion() -} - -// CheckpointInProgress reports whether a snapshot is currently being written. -func (s *CompositeStateStore) CheckpointInProgress() bool { - return s.snapshotMgr.checkpointInProgress() -} diff --git a/sei-db/state_db/ss/composite/snapshot.go b/sei-db/state_db/ss/composite/snapshot.go index 5143db8528..c116f091f8 100644 --- a/sei-db/state_db/ss/composite/snapshot.go +++ b/sei-db/state_db/ss/composite/snapshot.go @@ -74,9 +74,6 @@ type snapshotCoordinator struct { // floor carries the newest height every member holds to the members' own retention, which counts // only its own directories and would otherwise let an unpaired newer height crowd it out. floor *sssnapshot.Floor - // target is the height a controller.CheckpointScheduler asked for, 0 when none is outstanding. - // Atomic rather than under mu because the write path consults it for every version. - target atomic.Int64 mu sync.Mutex // lastRequested is the newest label already requested or present in any @@ -152,9 +149,6 @@ func (s *CompositeStateStore) startSnapshotManager(members []snapshotMember, flo func (c *snapshotCoordinator) stop() { c.mu.Lock() c.stopped = true - // Dropped rather than serviced: no later block will arrive to reach it, and a scheduler polling - // checkpointInProgress across shutdown would otherwise be told a checkpoint is still coming. - c.target.Store(0) c.mu.Unlock() c.scheduling.Wait() c.publishing.Wait() @@ -166,14 +160,11 @@ func (c *snapshotCoordinator) isRunning() bool { return !c.stopped } -// maybeSnapshot takes a snapshot when version is a height this coordinator owes one at. It is called -// from the write path for every version, so the common case is the due test and nothing else. +// maybeSnapshot takes a snapshot when version lands on an interval boundary. +// It is called from the write path for every version, so the common case is the +// modulo test and nothing else. func (c *snapshotCoordinator) maybeSnapshot(version int64) { - if c == nil || version <= 0 { - return - } - due, scheduled := c.due(version) - if !due { + if c == nil || version <= 0 || c.interval <= 0 || version%c.interval != 0 { return } now := time.Now() @@ -187,7 +178,7 @@ func (c *snapshotCoordinator) maybeSnapshot(version int64) { // A repeated commit-path call is expected and is not a skipped attempt. case c.inFlight: skipReason = "in_flight" - case !scheduled && !c.lastRequestAt.IsZero() && now.Sub(c.lastRequestAt) < c.minTime: + case !c.lastRequestAt.IsZero() && now.Sub(c.lastRequestAt) < c.minTime: skipReason = "minimum_time_interval" default: c.lastRequested = version @@ -227,70 +218,6 @@ func (c *snapshotCoordinator) maybeSnapshot(version int64) { } } -// due reports whether version is a height this coordinator owes a snapshot at, and whether it was a -// scheduler that asked. Two cadences can ask — the configured SnapshotInterval, and a target a -// controller.CheckpointScheduler dispatched — and a height both name is one snapshot, not two. -// -// Which one asked decides whether the minimum-time gate applies. That gate paces the configured -// interval; applying it to a dispatched target would drop the one height the scheduler's other stores -// are checkpointing at, which is the whole guarantee it exists to make. Skipping it loses no pacing, -// because a scheduler gates its own dispatches: a dispatched target arrives already paced. -func (c *snapshotCoordinator) due(version int64) (due bool, scheduled bool) { - scheduled = c.consumeTarget(version) - return scheduled || (c.interval > 0 && version%c.interval == 0), scheduled -} - -// consumeTarget reports whether version is the height a scheduler asked for, clearing the target once -// version has reached it. With no target outstanding it is a single atomic load, which is what keeps -// it callable for every version on the write path. -// -// It clears on an overshoot as well as on a match: a target left set at a height already past is one -// no later block can match. Overshooting means the write path skipped the height between the target -// being accepted and the block arriving, so it is reported rather than absorbed. -func (c *snapshotCoordinator) consumeTarget(version int64) bool { - target := c.target.Load() - if target == 0 || version < target { - return false - } - if !c.target.CompareAndSwap(target, 0) { - return false - } - if version > target { - logger.Error("state store passed a scheduled checkpoint height without snapshotting it", - "targetVersion", target, "version", version) - return false - } - return true -} - -// acceptTarget records version as the height this coordinator owes a scheduled snapshot at. It holds -// one target at a time. A height already requested, a second request while one is pending, or a -// request after stop is ignored. -func (c *snapshotCoordinator) acceptTarget(version int64) { - c.mu.Lock() - defer c.mu.Unlock() - if c.stopped { - return - } - if c.target.Load() != 0 { - return - } - if version <= c.lastRequested { - return - } - c.target.Store(version) -} - -// checkpointInProgress reports whether a snapshot is currently being staged or published. -func (c *snapshotCoordinator) checkpointInProgress() bool { - if c == nil { - return false - } - c.mu.Lock() - defer c.mu.Unlock() - return c.inFlight -} - func (c *snapshotCoordinator) finishSnapshot() { c.mu.Lock() c.inFlight = false From 4b93aad3900b04e43c0c28341914fdd1ab701035 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Wed, 26 Aug 2026 13:35:18 -0700 Subject: [PATCH 03/11] Address comments for check complete too early --- sei-db/controller/checkpoint_scheduler.go | 11 ++++--- .../controller/checkpoint_scheduler_test.go | 33 ++++++++++++++----- sei-db/controller/checkpointable_store.go | 4 ++- 3 files changed, 34 insertions(+), 14 deletions(-) diff --git a/sei-db/controller/checkpoint_scheduler.go b/sei-db/controller/checkpoint_scheduler.go index d57e1e1988..548244ee1b 100644 --- a/sei-db/controller/checkpoint_scheduler.go +++ b/sei-db/controller/checkpoint_scheduler.go @@ -166,13 +166,14 @@ func (s *CheckpointScheduler) run() { } // scheduleNextCheckpoint runs one cycle: it hands the next boundary to every store, or does nothing -// when a store is still writing the last checkpoint, has not reached its scheduled version, or the -// last one finished too recently. +// when a store is still writing the last checkpoint, has not committed past its scheduled version, or +// the last one finished too recently. func (s *CheckpointScheduler) scheduleNextCheckpoint() { if s.CheckpointInProgress() { return } - if s.scheduledVersion != 0 && !s.allReached(s.scheduledVersion) { + // Stores bump LatestVersion on commit before the checkpoint write, so wait for the next version. + if s.scheduledVersion != 0 && !s.allStoresCommitted(s.scheduledVersion+1) { return } s.noteCheckpointFinished() @@ -189,8 +190,8 @@ func (s *CheckpointScheduler) scheduleNextCheckpoint() { "targetVersion", targetVersion, "stores", strings.Join(slices.Sorted(maps.Keys(s.stores)), ",")) } -// allReached reports whether every store has committed at least version. -func (s *CheckpointScheduler) allReached(version int64) bool { +// allStoresCommitted reports whether every store has committed at least version. +func (s *CheckpointScheduler) allStoresCommitted(version int64) bool { for _, store := range s.stores { if store.LatestVersion() < version { return false diff --git a/sei-db/controller/checkpoint_scheduler_test.go b/sei-db/controller/checkpoint_scheduler_test.go index 467f3b2b7c..e4ec9c11cd 100644 --- a/sei-db/controller/checkpoint_scheduler_test.go +++ b/sei-db/controller/checkpoint_scheduler_test.go @@ -166,8 +166,8 @@ func TestTargetIsOfferedAWholeIntervalAhead(t *testing.T) { require.Equal(t, []int64{1000}, store.offeredTargets(), "offered at version 1, 999 blocks early") } -// One target is outstanding at a time, so a store that has not reached its target holds the next -// boundary back rather than collecting targets it would service late. +// One target is outstanding at a time, so a store that has not committed past its target holds the +// next boundary back rather than collecting targets it would service late. func TestNoNewTargetWhileOneIsOutstanding(t *testing.T) { prompt, lagging := newFakeStore(0), newFakeStore(0) scheduler := newScheduler(t, 10, map[string]*fakeStore{"prompt": prompt, "lagging": lagging}) @@ -181,19 +181,36 @@ func TestNoNewTargetWhileOneIsOutstanding(t *testing.T) { scheduler.scheduleNextCheckpoint() require.Equal(t, []int64{10}, prompt.offeredTargets()) - lagging.commitTo(10) + lagging.commitTo(11) scheduler.scheduleNextCheckpoint() require.Equal(t, []int64{10, 40}, prompt.offeredTargets()) require.Equal(t, []int64{10, 40}, lagging.offeredTargets()) } -// A store still writing its snapshot holds the next boundary even after it has reached the height. -func TestNoNewTargetWhileAStoreIsWriting(t *testing.T) { +// Stores bump LatestVersion on commit and only then start the checkpoint write. A poll in that gap +// sees the scheduled version reached and CheckpointInProgress still false; that is not completion. +func TestNoNewTargetWhileLatestVersionIsStillTheScheduledVersion(t *testing.T) { store := newFakeStore(0) scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": store}) scheduler.scheduleNextCheckpoint() store.commitTo(10) + scheduler.scheduleNextCheckpoint() + require.Equal(t, []int64{10}, store.offeredTargets(), + "LatestVersion at the scheduled height is the commit, not a finished checkpoint") + + store.commitTo(11) + scheduler.scheduleNextCheckpoint() + require.Equal(t, []int64{10, 20}, store.offeredTargets()) +} + +// A store still writing its snapshot holds the next boundary even after it has committed past the height. +func TestNoNewTargetWhileAStoreIsWriting(t *testing.T) { + store := newFakeStore(0) + scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": store}) + + scheduler.scheduleNextCheckpoint() + store.commitTo(15) store.setRunning(true) scheduler.scheduleNextCheckpoint() require.Equal(t, []int64{10}, store.offeredTargets()) @@ -264,7 +281,7 @@ func TestTheMinTimeGateReleasesOnceItElapses(t *testing.T) { scheduler := pacedScheduler(t, map[string]*fakeStore{"only": store}) scheduler.scheduleNextCheckpoint() - store.commitTo(10) + store.commitTo(11) scheduler.scheduleNextCheckpoint() require.Equal(t, []int64{10}, store.offeredTargets()) @@ -284,7 +301,7 @@ func TestTheMinTimeGateIsTimedFromCompletion(t *testing.T) { scheduler.scheduleNextCheckpoint() require.True(t, scheduler.lastCheckpointAt.IsZero(), "dispatching a target does not start the gate") - store.commitTo(10) + store.commitTo(11) scheduler.scheduleNextCheckpoint() require.False(t, scheduler.lastCheckpointAt.IsZero(), "the finished checkpoint starts the gate") } @@ -294,7 +311,7 @@ func TestAZeroMinTimeLeavesTheBlockIntervalAsTheOnlyPacing(t *testing.T) { scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": store}) scheduler.scheduleNextCheckpoint() - store.commitTo(10) + store.commitTo(11) scheduler.scheduleNextCheckpoint() require.Equal(t, []int64{10, 20}, store.offeredTargets()) diff --git a/sei-db/controller/checkpointable_store.go b/sei-db/controller/checkpointable_store.go index a5c728dbba..6ab6ff4384 100644 --- a/sei-db/controller/checkpointable_store.go +++ b/sei-db/controller/checkpointable_store.go @@ -14,7 +14,9 @@ type CheckpointableStore interface { // LatestVersion returns the newest version this store has committed, 0 when it has committed // nothing. The scheduler picks a target above every store's answer, so a store that reports a - // version it has not durably reached is asking to be handed a target it will never see. + // version it has not durably reached is asking to be handed a target it will never see. A + // scheduled checkpoint is not treated as finished until every store reports a version strictly + // above it: stores bump this on commit and only then start the checkpoint write. LatestVersion() int64 // CheckpointInProgress reports whether a checkpoint this store is writing has yet to finish. From ba361d070a4804da4616ec92d36e537e3e584ac9 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Thu, 27 Aug 2026 22:17:21 -0700 Subject: [PATCH 04/11] Make checkpoint scheduler a poll model --- sei-db/config/checkpoint_config.go | 28 + sei-db/controller/checkpoint_scheduler.go | 294 ++++----- .../controller/checkpoint_scheduler_test.go | 561 ++++++++---------- sei-db/controller/checkpointable_store.go | 24 - 4 files changed, 374 insertions(+), 533 deletions(-) create mode 100644 sei-db/config/checkpoint_config.go delete mode 100644 sei-db/controller/checkpointable_store.go diff --git a/sei-db/config/checkpoint_config.go b/sei-db/config/checkpoint_config.go new file mode 100644 index 0000000000..075e39e394 --- /dev/null +++ b/sei-db/config/checkpoint_config.go @@ -0,0 +1,28 @@ +package config + +import ( + "time" + + "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/memiavl" +) + +// CheckpointConfig configures a CheckpointScheduler: how far apart the heights it picks are. +// +// A height has to clear every interval set above 0, so with both set the tighter one paces the +// cadence. A value of 0 or less is unused, and with neither set checkpointing is off. +type CheckpointConfig struct { + // TimeInterval is the wall-clock gap between checkpoints, measured from the last one completing. + TimeInterval time.Duration + + // BlockInterval is the block gap between checkpoints, measured from the last height picked. + BlockInterval int64 +} + +// DefaultCheckpointConfig returns a cadence mirroring the state-commit snapshot settings: a +// checkpoint every 10,000 blocks, and no more than one an hour. +func DefaultCheckpointConfig() CheckpointConfig { + return CheckpointConfig{ + TimeInterval: time.Duration(memiavl.DefaultSnapshotMinTimeInterval) * time.Second, + BlockInterval: memiavl.DefaultSnapshotInterval, + } +} diff --git a/sei-db/controller/checkpoint_scheduler.go b/sei-db/controller/checkpoint_scheduler.go index 548244ee1b..3f3e30ddca 100644 --- a/sei-db/controller/checkpoint_scheduler.go +++ b/sei-db/controller/checkpoint_scheduler.go @@ -4,229 +4,151 @@ package controller import ( - "context" - "errors" - "fmt" "maps" - "slices" - "strings" "sync" "time" "github.com/sei-protocol/seilog" + + "github.com/sei-protocol/sei-chain/sei-db/config" ) var checkpointLogger = seilog.NewLogger("db", "checkpoint") -// checkpointPollInterval is how often the scheduler looks for a boundary to dispatch. It has to stay -// well inside the time the stores take to cover an interval's worth of versions, or boundaries pass -// undispatched. -const checkpointPollInterval = 10 * time.Second - -// CheckpointConfig is the cadence a CheckpointScheduler holds every registered store to. -type CheckpointConfig struct { - // CheckpointInterval is how many blocks apart checkpoints are taken. 0 turns checkpointing off. - CheckpointInterval int64 - - // MinTimeBetweenCheckpoints is the shortest wall-clock gap allowed between one checkpoint - // finishing and the next being scheduled, which bounds how fast a node replaying blocks - // checkpoints. 0 leaves CheckpointInterval as the only pacing. - MinTimeBetweenCheckpoints time.Duration -} - -// Validate reports whether this config describes a cadence that can be scheduled. -func (c CheckpointConfig) Validate() error { - if c.CheckpointInterval < 0 { - return fmt.Errorf("checkpoint interval must not be negative, got %d", c.CheckpointInterval) - } - if c.MinTimeBetweenCheckpoints < 0 { - return fmt.Errorf("minimum time between checkpoints must not be negative, got %s", - c.MinTimeBetweenCheckpoints) - } - return nil -} - -// CheckpointScheduler drives one checkpoint cadence across every store registered with it, so that a -// node's stores hold checkpoints of the same versions rather than of whatever version each happened to -// be at. Each cycle picks the next interval boundary above every store's committed version and hands -// that one version to all of them, to checkpoint when their own write paths reach it; one target is -// outstanding at a time. +// CheckpointScheduler picks the heights every store checkpoints at, so the stores of one node hold +// checkpoints of the same versions rather than of whatever version each happened to reach. +// +// Stores ask ShouldCheckpoint at every version they commit and call MarkCheckpointComplete once +// that version's checkpoint has finished, whether or not it succeeded: every yes obliges a report. +// A store joins the schedule by asking, and a height is held until every store registered when it +// was picked, along with any that took it afterwards, has reported. A store behind the others is +// handed that same height rather than one that has moved on, and heights arrive no faster than the +// slowest store reaches them. +// +// A no is final and covers every height under it, so a version refused to one store is refused to +// all of them; a yes holds for every store that reaches that height before it is replaced. // -// Start and Close are the owner's to call, from one goroutine: nothing here is guarded. +// A time interval, a block interval, or both may be set. With both set a height has to clear both; +// a value of 0 or less is unused, and with neither set checkpointing is off. Both are measured from +// the last checkpoint, and from the scheduler's creation before there is one. +// +// Every store has to ask at each version it commits. One asking at only some of them never reaches +// the height being held, which stops the node checkpointing rather than only that store. type CheckpointScheduler struct { - config CheckpointConfig - ctx context.Context - stopCh chan struct{} - wg sync.WaitGroup - - stores map[string]CheckpointableStore - started bool - closed bool - - // Only the run loop touches these, through scheduleNextCheckpoint. - scheduledVersion int64 - lastCheckpointAt time.Time + mu sync.Mutex + config config.CheckpointConfig + + // registered holds every store that has asked, which is how a store joins the schedule. + registered map[string]struct{} + // awaiting holds the stores the current height is held for that have yet to report it. Empty + // means no checkpoint is outstanding. + awaiting map[string]struct{} + + // nextCheckpointVersion is the height stores checkpoint at, 0 before the first one is picked. + // It stays set once complete, so a store that has yet to reach it is still given that height. + nextCheckpointVersion int64 + // checkpointedAt is when the last checkpoint completed, the scheduler's creation before the first. + checkpointedAt time.Time + // rejectedVersion is the highest height answered no. A later ask at it is answered no again + // rather than turning yes once an interval elapses under it. + rejectedVersion int64 } -// NewCheckpointScheduler returns a scheduler that holds every store in stores to config, or that -// schedules nothing when config turns checkpointing off. Call Start to begin. -func NewCheckpointScheduler( - ctx context.Context, - config CheckpointConfig, - stores map[string]CheckpointableStore, -) (*CheckpointScheduler, error) { - if ctx == nil { - return nil, errors.New("context is required") - } - if err := config.Validate(); err != nil { - return nil, err - } - copied := make(map[string]CheckpointableStore, len(stores)) - for name, store := range stores { - if name == "" { - return nil, errors.New("checkpoint store name is required") - } - if store == nil { - return nil, fmt.Errorf("checkpoint store %q is nil", name) - } - copied[name] = store - } +// NewCheckpointScheduler returns a scheduler holding every store that asks it to one cadence. +func NewCheckpointScheduler(cfg config.CheckpointConfig) *CheckpointScheduler { + checkpointLogger.Info("checkpoint scheduler created", + "timeInterval", cfg.TimeInterval, "blockInterval", cfg.BlockInterval) return &CheckpointScheduler{ - config: config, - ctx: ctx, - stopCh: make(chan struct{}), - stores: copied, - }, nil -} - -// Start begins dispatching targets until Close is called or ctx is cancelled. Starting twice is an -// error. -func (s *CheckpointScheduler) Start() error { - if s.closed { - return errors.New("cannot start a closed checkpoint scheduler") + config: cfg, + registered: make(map[string]struct{}), + awaiting: make(map[string]struct{}), + checkpointedAt: time.Now(), } - if s.started { - return errors.New("checkpoint scheduler already started") - } - s.started = true - checkpointLogger.Info("checkpoint scheduler started", - "interval", s.config.CheckpointInterval, - "minTimeBetweenCheckpoints", s.config.MinTimeBetweenCheckpoints, - "stores", strings.Join(slices.Sorted(maps.Keys(s.stores)), ","), - ) - s.wg.Add(1) - go s.run() - return nil } -// Close stops dispatching and waits for the run loop to exit. -func (s *CheckpointScheduler) Close() error { - if s.closed { - return nil +// ShouldCheckpoint reports whether version is a height for store to checkpoint at, registering +// store with the schedule when this is its first ask. +func (s *CheckpointScheduler) ShouldCheckpoint(store string, version int64) bool { + if !s.checkpointEnabled() || version <= 0 { + return false } - s.closed = true - close(s.stopCh) - s.wg.Wait() - return nil -} -// CheckpointInProgress reports whether any registered store is still writing a checkpoint. -func (s *CheckpointScheduler) CheckpointInProgress() bool { - for _, store := range s.stores { - if store.CheckpointInProgress() { - return true - } - } - return false -} + s.mu.Lock() + defer s.mu.Unlock() + s.registered[store] = struct{}{} -func (s *CheckpointScheduler) run() { - defer s.wg.Done() - - if s.config.CheckpointInterval == 0 || len(s.stores) == 0 { - return + if version == s.nextCheckpointVersion { + s.holdFor(store) + return true } - - ticker := time.NewTicker(checkpointPollInterval) - defer ticker.Stop() - - for { - // Ahead of the first wait, not after it: there is already a boundary to announce by the time - // Start returns, and waiting out a poll interval only delays the first checkpoint. - s.scheduleNextCheckpoint() - - select { - case <-s.stopCh: - return - case <-s.ctx.Done(): - return - case <-ticker.C: - } + if s.alreadyRejected(version) { + return false + } + if s.checkpointOutstanding() || !s.hasReachedNextInterval(version) { + s.rejectedVersion = version + return false } + s.pickCheckpointHeight(version) + return true } -// scheduleNextCheckpoint runs one cycle: it hands the next boundary to every store, or does nothing -// when a store is still writing the last checkpoint, has not committed past its scheduled version, or -// the last one finished too recently. -func (s *CheckpointScheduler) scheduleNextCheckpoint() { - if s.CheckpointInProgress() { +// MarkCheckpointComplete records that store has finished the current height. Both intervals start +// once every store the height is held for has reported it. Any other version is ignored, as is a +// repeat from the same store. +// +// A store that took a height must report it on every path out of the checkpoint, a failed one +// included, which in practice means deferring the call. No further height is picked while one store +// is unreported, so a missed call stops the node checkpointing until it restarts. +func (s *CheckpointScheduler) MarkCheckpointComplete(store string, version int64) { + s.mu.Lock() + defer s.mu.Unlock() + + if version != s.nextCheckpointVersion { return } - // Stores bump LatestVersion on commit before the checkpoint write, so wait for the next version. - if s.scheduledVersion != 0 && !s.allStoresCommitted(s.scheduledVersion+1) { + if _, outstanding := s.awaiting[store]; !outstanding { return } - s.noteCheckpointFinished() - if s.withinMinTime() { + delete(s.awaiting, store) + + if s.checkpointOutstanding() { return } + s.checkpointedAt = time.Now() + checkpointLogger.Info("checkpoint complete", "version", version) +} - targetVersion := nextCheckpointVersion(s.stores, s.config.CheckpointInterval) - for _, store := range s.stores { - store.ScheduleCheckpoint(targetVersion) - } - s.scheduledVersion = targetVersion - checkpointLogger.Info("checkpoint scheduled", - "targetVersion", targetVersion, "stores", strings.Join(slices.Sorted(maps.Keys(s.stores)), ",")) +func (s *CheckpointScheduler) checkpointEnabled() bool { + return s.config.TimeInterval > 0 || s.config.BlockInterval > 0 } -// allStoresCommitted reports whether every store has committed at least version. -func (s *CheckpointScheduler) allStoresCommitted(version int64) bool { - for _, store := range s.stores { - if store.LatestVersion() < version { - return false - } - } - return true +func (s *CheckpointScheduler) checkpointOutstanding() bool { + return len(s.awaiting) > 0 } -// noteCheckpointFinished records that the last scheduled checkpoint completed, starting the -// minimum-time gate. scheduledVersion is what marks this as the completion: later cycles also find -// no store writing, and treating those as completions too would push the gate forward every poll. -func (s *CheckpointScheduler) noteCheckpointFinished() { - if s.scheduledVersion == 0 { - return - } - s.scheduledVersion = 0 - s.lastCheckpointAt = time.Now() +// alreadyRejected reports whether version has been turned down already, either by trailing the +// height in hand or by being asked and refused. Standing by that no is what stops two stores asking +// at different moments from splitting on one version. +func (s *CheckpointScheduler) alreadyRejected(version int64) bool { + return version < s.nextCheckpointVersion || version <= s.rejectedVersion } -// withinMinTime reports whether the last checkpoint finished too recently for another to be scheduled. -// -// Timed from the checkpoint finishing rather than from its dispatch: dispatch runs an interval of -// blocks ahead, so timing from there would leave the gap short by however long the stores took. -func (s *CheckpointScheduler) withinMinTime() bool { - if s.config.MinTimeBetweenCheckpoints == 0 || s.lastCheckpointAt.IsZero() { - return false - } - return time.Since(s.lastCheckpointAt) < s.config.MinTimeBetweenCheckpoints +// hasReachedNextInterval reports whether every configured interval has passed for version. +func (s *CheckpointScheduler) hasReachedNextInterval(version int64) bool { + timeElapsed := s.config.TimeInterval <= 0 || time.Since(s.checkpointedAt) >= s.config.TimeInterval + blocksElapsed := s.config.BlockInterval <= 0 || version-s.nextCheckpointVersion >= s.config.BlockInterval + return timeElapsed && blocksElapsed } -// nextCheckpointVersion returns the next interval-aligned height strictly above every store's latest version. -func nextCheckpointVersion(stores map[string]CheckpointableStore, interval int64) int64 { - var latest int64 - for _, store := range stores { - latest = max(latest, store.LatestVersion()) - } - return (latest/interval + 1) * interval +// pickCheckpointHeight makes version the height in hand, held for the stores registered now. One +// registering later is not held for here, at a version it may already have passed, and is held for +// from the moment it takes a height. +func (s *CheckpointScheduler) pickCheckpointHeight(version int64) { + s.nextCheckpointVersion = version + s.awaiting = maps.Clone(s.registered) +} + +// holdFor keeps the height in hand from being replaced until store has reported it. +func (s *CheckpointScheduler) holdFor(store string) { + s.awaiting[store] = struct{}{} } diff --git a/sei-db/controller/checkpoint_scheduler_test.go b/sei-db/controller/checkpoint_scheduler_test.go index e4ec9c11cd..99e6ba5646 100644 --- a/sei-db/controller/checkpoint_scheduler_test.go +++ b/sei-db/controller/checkpoint_scheduler_test.go @@ -1,418 +1,333 @@ package controller import ( - "context" - "slices" "sync" "testing" "time" "github.com/stretchr/testify/require" + + "github.com/sei-protocol/sei-chain/sei-db/config" ) -// Which target the scheduler picks, and whether it picks one at all, is decided synchronously in -// scheduleNextCheckpoint. Those tests drive it directly and assert exactly; only the tests about the run -// loop itself — that it dispatches, that it stops — start a goroutine. - -// fakeStore stands in for a store the scheduler drives. It records every target it is offered so a -// test can assert what the stores were asked for, which is the scheduler's whole output. -type fakeStore struct { - mu sync.Mutex - version int64 - pending int64 - running bool - offered []int64 +// newScheduler spells the two intervals out positionally, which reads better than a config literal +// in tests that vary nothing else. +func newScheduler(timeInterval time.Duration, blockInterval int64) *CheckpointScheduler { + return NewCheckpointScheduler(config.CheckpointConfig{ + TimeInterval: timeInterval, + BlockInterval: blockInterval, + }) } -func newFakeStore(version int64) *fakeStore { - return &fakeStore{version: version} +// elapseTimeInterval moves the interval's starting point back rather than waiting it out. +func (s *CheckpointScheduler) elapseTimeInterval() { + s.checkpointedAt = time.Now().Add(-s.config.TimeInterval) } -func (f *fakeStore) ScheduleCheckpoint(targetVersion int64) { - f.mu.Lock() - defer f.mu.Unlock() - f.offered = append(f.offered, targetVersion) - f.pending = targetVersion -} +// --------------------------------------------------------------------------- +// Which intervals are in use +// --------------------------------------------------------------------------- -func (f *fakeStore) LatestVersion() int64 { - f.mu.Lock() - defer f.mu.Unlock() - return f.version +func TestNeitherIntervalSetDisablesCheckpointing(t *testing.T) { + for _, scheduler := range []*CheckpointScheduler{ + newScheduler(0, 0), + newScheduler(-time.Second, 0), + newScheduler(0, -1), + } { + require.False(t, scheduler.ShouldCheckpoint("sc", 1)) + require.False(t, scheduler.ShouldCheckpoint("sc", 1_000_000)) + } } -func (f *fakeStore) CheckpointInProgress() bool { - f.mu.Lock() - defer f.mu.Unlock() - return f.running -} +func TestNonPositiveVersionsAreNeverCheckpointHeights(t *testing.T) { + scheduler := newScheduler(time.Hour, 10) + scheduler.elapseTimeInterval() -func (f *fakeStore) setRunning(running bool) { - f.mu.Lock() - defer f.mu.Unlock() - f.running = running + require.False(t, scheduler.ShouldCheckpoint("sc", 0)) + require.False(t, scheduler.ShouldCheckpoint("sc", -1)) } -// commitTo advances the store to version, performing an accepted checkpoint on the way past it. -func (f *fakeStore) commitTo(version int64) { - f.mu.Lock() - defer f.mu.Unlock() - f.version = version - if f.pending != 0 && version >= f.pending { - f.pending = 0 - } -} +// The first height clears the same intervals as every later one, measured from the scheduler's +// creation. Answering yes to whichever version happens to ask first would ignore the cadence. +func TestTheFirstHeightWaitsForTheTimeInterval(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) -func (f *fakeStore) offeredTargets() []int64 { - f.mu.Lock() - defer f.mu.Unlock() - return slices.Clone(f.offered) -} + require.False(t, scheduler.ShouldCheckpoint("sc", 10)) -// newScheduler returns an unstarted scheduler over stores. -func newScheduler(t *testing.T, interval int64, stores map[string]*fakeStore) *CheckpointScheduler { - t.Helper() - return newConfiguredScheduler(t, CheckpointConfig{CheckpointInterval: interval}, stores) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 11)) } -// newConfiguredScheduler is newScheduler for the tests that care about more of the cadence than the -// block interval. -func newConfiguredScheduler( - t *testing.T, - config CheckpointConfig, - stores map[string]*fakeStore, -) *CheckpointScheduler { - t.Helper() - copied := make(map[string]CheckpointableStore, len(stores)) - for name, store := range stores { - copied[name] = store - } - scheduler, err := NewCheckpointScheduler(context.Background(), config, copied) - require.NoError(t, err) - return scheduler +func TestTheFirstHeightWaitsForTheBlockInterval(t *testing.T) { + scheduler := newScheduler(0, 100) + + require.False(t, scheduler.ShouldCheckpoint("sc", 99)) + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) } -// requireLoopStopped waits, with a bound, for the run loop to have exited — or to never have started. -// The wait is bounded rather than a bare wg.Wait because the failure it looks for is a loop that keeps -// running, and that failure has to be a test failure rather than a test that hangs. -func requireLoopStopped(t *testing.T, scheduler *CheckpointScheduler) { - t.Helper() - stopped := make(chan struct{}) - go func() { - scheduler.wg.Wait() - close(stopped) - }() - select { - case <-stopped: - case <-time.After(time.Second): - t.Fatal("checkpoint scheduler run loop is still running") - } +// With both set a height has to clear both, so the tighter one paces the cadence. +func TestBothIntervalsMustElapseWhenBothAreSet(t *testing.T) { + scheduler := newScheduler(time.Hour, 10) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 10)) + scheduler.MarkCheckpointComplete("sc", 10) + + require.False(t, scheduler.ShouldCheckpoint("sc", 15), "neither interval has elapsed") + + scheduler.elapseTimeInterval() + require.False(t, scheduler.ShouldCheckpoint("sc", 19), "the block interval has not elapsed") + require.True(t, scheduler.ShouldCheckpoint("sc", 20)) } // --------------------------------------------------------------------------- -// Which target a cycle picks +// One answer per height // --------------------------------------------------------------------------- -func TestNextCheckpointVersionAlignsToInterval(t *testing.T) { - for _, tc := range []struct { - latest, interval, want int64 - }{ - {latest: 0, interval: 1000, want: 1000}, - {latest: 1, interval: 1000, want: 1000}, - {latest: 999, interval: 1000, want: 1000}, - {latest: 1000, interval: 1000, want: 2000}, - {latest: 2100, interval: 1000, want: 3000}, - {latest: 3700, interval: 1000, want: 4000}, - {latest: 4000, interval: 1000, want: 5000}, - } { - stores := map[string]CheckpointableStore{"only": newFakeStore(tc.latest)} - require.Equal(t, tc.want, nextCheckpointVersion(stores, tc.interval), - "latest %d interval %d", tc.latest, tc.interval) - } -} +func TestAPickedHeightIsYesForEveryStore(t *testing.T) { + scheduler := newScheduler(0, 10) + require.True(t, scheduler.ShouldCheckpoint("sc", 10)) -// The point of the scheduler: every store is asked for the same version, even when they are at -// different versions when the target is chosen. -func TestCycleOffersOneTargetToEveryStore(t *testing.T) { - fast, slow := newFakeStore(95), newFakeStore(80) - scheduler := newScheduler(t, 100, map[string]*fakeStore{"fast": fast, "slow": slow}) + require.True(t, scheduler.ShouldCheckpoint("ss", 10)) + require.True(t, scheduler.ShouldCheckpoint("receipt", 10)) +} - scheduler.scheduleNextCheckpoint() +// The height stays available after it completes: a store that reaches it later than the store that +// finished it is handed the same height rather than the next one. +func TestAPickedHeightStaysYesForALaggingStore(t *testing.T) { + scheduler := newScheduler(0, 10) + require.True(t, scheduler.ShouldCheckpoint("sc", 10)) + scheduler.MarkCheckpointComplete("sc", 10) - require.Equal(t, []int64{100}, fast.offeredTargets()) - require.Equal(t, []int64{100}, slow.offeredTargets()) + require.True(t, scheduler.ShouldCheckpoint("sc", 10)) } -// The target has to be above every store's version, not above the laggard's: a store that is ahead can -// no longer checkpoint at a version it has passed, and would answer with a different one. -func TestTargetIsAboveTheMostAdvancedStore(t *testing.T) { - ahead, behind := newFakeStore(250), newFakeStore(10) - scheduler := newScheduler(t, 100, map[string]*fakeStore{"ahead": ahead, "behind": behind}) +// A height answered no stays no once an interval elapses under it, so two stores asking either side +// of that moment cannot split on the same version. +func TestARejectedHeightStaysNoForEveryStore(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 10)) + scheduler.MarkCheckpointComplete("sc", 10) - scheduler.scheduleNextCheckpoint() + require.False(t, scheduler.ShouldCheckpoint("sc", 11)) - require.Equal(t, []int64{300}, ahead.offeredTargets()) - require.Equal(t, []int64{300}, behind.offeredTargets()) + scheduler.elapseTimeInterval() + require.False(t, scheduler.ShouldCheckpoint("ss", 11), "11 was already answered no") + require.True(t, scheduler.ShouldCheckpoint("sc", 12)) } -// A target is offered well before the stores reach it, which is what makes the shared version an -// invariant rather than a race. Nothing about the current version gates the offer. -func TestTargetIsOfferedAWholeIntervalAhead(t *testing.T) { - store := newFakeStore(1) - scheduler := newScheduler(t, 1000, map[string]*fakeStore{"only": store}) +// The refusal covers every height at or under the one refused, not just that exact height. A store +// that asks just short of the interval pushes the floor up to the height it was at, so a store +// lagging behind it cannot walk in under that floor the moment the interval elapses and take a +// height the store ahead of it was already refused. +func TestALaggingStoreCannotTakeAHeightUnderARefusedOne(t *testing.T) { + scheduler := newScheduler(5*time.Minute, 0) - scheduler.scheduleNextCheckpoint() - - require.Equal(t, []int64{1000}, store.offeredTargets(), "offered at version 1, 999 blocks early") -} + require.False(t, scheduler.ShouldCheckpoint("sc", 100), "the interval has not elapsed") -// One target is outstanding at a time, so a store that has not committed past its target holds the -// next boundary back rather than collecting targets it would service late. -func TestNoNewTargetWhileOneIsOutstanding(t *testing.T) { - prompt, lagging := newFakeStore(0), newFakeStore(0) - scheduler := newScheduler(t, 10, map[string]*fakeStore{"prompt": prompt, "lagging": lagging}) - - scheduler.scheduleNextCheckpoint() - require.Equal(t, []int64{10}, prompt.offeredTargets()) - prompt.commitTo(35) - - // The laggard still holds target 10, so the boundaries at 20 and 30 pass without an offer. - scheduler.scheduleNextCheckpoint() - scheduler.scheduleNextCheckpoint() - require.Equal(t, []int64{10}, prompt.offeredTargets()) - - lagging.commitTo(11) - scheduler.scheduleNextCheckpoint() - require.Equal(t, []int64{10, 40}, prompt.offeredTargets()) - require.Equal(t, []int64{10, 40}, lagging.offeredTargets()) -} + scheduler.elapseTimeInterval() + for _, lagging := range []int64{98, 99, 100} { + require.False(t, scheduler.ShouldCheckpoint("ss", lagging), "height %d is under the refused 100", lagging) + } -// Stores bump LatestVersion on commit and only then start the checkpoint write. A poll in that gap -// sees the scheduled version reached and CheckpointInProgress still false; that is not completion. -func TestNoNewTargetWhileLatestVersionIsStillTheScheduledVersion(t *testing.T) { - store := newFakeStore(0) - scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": store}) - - scheduler.scheduleNextCheckpoint() - store.commitTo(10) - scheduler.scheduleNextCheckpoint() - require.Equal(t, []int64{10}, store.offeredTargets(), - "LatestVersion at the scheduled height is the commit, not a finished checkpoint") - - store.commitTo(11) - scheduler.scheduleNextCheckpoint() - require.Equal(t, []int64{10, 20}, store.offeredTargets()) + require.True(t, scheduler.ShouldCheckpoint("sc", 101)) + require.True(t, scheduler.ShouldCheckpoint("ss", 101), "the lagging store reaches the same height") } -// A store still writing its snapshot holds the next boundary even after it has committed past the height. -func TestNoNewTargetWhileAStoreIsWriting(t *testing.T) { - store := newFakeStore(0) - scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": store}) - - scheduler.scheduleNextCheckpoint() - store.commitTo(15) - store.setRunning(true) - scheduler.scheduleNextCheckpoint() - require.Equal(t, []int64{10}, store.offeredTargets()) +func TestAHeightBelowTheCurrentOneIsNo(t *testing.T) { + scheduler := newScheduler(0, 10) + require.True(t, scheduler.ShouldCheckpoint("sc", 10)) - store.setRunning(false) - scheduler.scheduleNextCheckpoint() - require.Equal(t, []int64{10, 20}, store.offeredTargets()) + require.False(t, scheduler.ShouldCheckpoint("ss", 9)) } -// CheckpointInProgress is an aggregate over the stores: true while any store is writing. -func TestCheckpointInProgressReportsAnyStore(t *testing.T) { - first, second := newFakeStore(0), newFakeStore(0) - scheduler := newScheduler(t, 10, map[string]*fakeStore{"first": first, "second": second}) +func TestConcurrentAsksAtTheSameHeightAgree(t *testing.T) { + scheduler := newScheduler(0, 10) - require.False(t, scheduler.CheckpointInProgress()) - first.setRunning(true) - second.setRunning(true) - require.True(t, scheduler.CheckpointInProgress()) + var answers [8]bool + var wg sync.WaitGroup + for i := range answers { + wg.Add(1) + go func(i int) { + defer wg.Done() + answers[i] = scheduler.ShouldCheckpoint("sc", 10) + }(i) + } + wg.Wait() - first.setRunning(false) - require.True(t, scheduler.CheckpointInProgress(), "second store is still writing") - second.setRunning(false) - require.False(t, scheduler.CheckpointInProgress()) + for i, answer := range answers { + require.True(t, answer, "asker %d", i) + } } // --------------------------------------------------------------------------- -// The minimum-time gate +// Stores at different heights // --------------------------------------------------------------------------- -// pacedScheduler returns a scheduler whose minimum-time gate is wide enough that no test elapses it by -// running. Tests that need it elapsed move lastCheckpointAt rather than waiting. -func pacedScheduler(t *testing.T, stores map[string]*fakeStore) *CheckpointScheduler { - t.Helper() - return newConfiguredScheduler(t, CheckpointConfig{ - CheckpointInterval: 10, - MinTimeBetweenCheckpoints: time.Hour, - }, stores) -} - -// Nothing has been checkpointed yet, so there is no gap to enforce and the first boundary is not held. -func TestTheMinTimeGateDoesNotDelayTheFirstCheckpoint(t *testing.T) { - store := newFakeStore(0) - scheduler := pacedScheduler(t, map[string]*fakeStore{"only": store}) - - scheduler.scheduleNextCheckpoint() - - require.Equal(t, []int64{10}, store.offeredTargets()) +// walkLeaderAndLaggard runs two stores up to a height, the second one lag blocks behind the first, +// and returns the heights each checkpointed at. A non-zero elapseEvery elapses the time interval +// once every that many heights, standing in for wall-clock passing as the chain advances. +func walkLeaderAndLaggard( + scheduler *CheckpointScheduler, to, lag, elapseEvery int64, +) (leaderTook, laggardTook []int64) { + for height := int64(1); height <= to; height++ { + if elapseEvery > 0 && height%elapseEvery == 0 { + scheduler.elapseTimeInterval() + } + if scheduler.ShouldCheckpoint("leader", height) { + leaderTook = append(leaderTook, height) + scheduler.MarkCheckpointComplete("leader", height) + } + behind := height - lag + if behind > 0 && scheduler.ShouldCheckpoint("laggard", behind) { + laggardTook = append(laggardTook, behind) + scheduler.MarkCheckpointComplete("laggard", behind) + } + } + return leaderTook, laggardTook } -// Once a checkpoint has been taken, the next boundary waits for the gate even though the stores have -// long since passed it. -func TestTheMinTimeGateHoldsBackTheNextTarget(t *testing.T) { - store := newFakeStore(0) - scheduler := pacedScheduler(t, map[string]*fakeStore{"only": store}) +// A store lagging by more than the block interval still takes every height the leader takes, since +// the height is held until it arrives. The heights are no longer spaced by the interval alone: each +// one waits out the lag as well, which is the cadence cost of keeping the stores together. +func TestEveryStoreTakesTheSameHeightsHoweverFarBehind(t *testing.T) { + scheduler := newScheduler(0, 100) - scheduler.scheduleNextCheckpoint() - require.Equal(t, []int64{10}, store.offeredTargets()) - store.commitTo(100) + leaderTook, laggardTook := walkLeaderAndLaggard(scheduler, 1000, 150, 0) - scheduler.scheduleNextCheckpoint() - scheduler.scheduleNextCheckpoint() - - require.Equal(t, []int64{10}, store.offeredTargets(), "the gate has not elapsed") + require.Equal(t, []int64{100, 200, 351, 502, 653, 804, 955}, leaderTook) + require.Equal(t, []int64{200, 351, 502, 653, 804}, laggardTook, + "100 predates the laggard's first ask, and it has yet to reach 955") } -func TestTheMinTimeGateReleasesOnceItElapses(t *testing.T) { - store := newFakeStore(0) - scheduler := pacedScheduler(t, map[string]*fakeStore{"only": store}) - - scheduler.scheduleNextCheckpoint() - store.commitTo(11) - scheduler.scheduleNextCheckpoint() - require.Equal(t, []int64{10}, store.offeredTargets()) +func TestEveryStoreTakesTheSameHeightsOnATimeInterval(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) - scheduler.lastCheckpointAt = time.Now().Add(-2 * time.Hour) - scheduler.scheduleNextCheckpoint() + leaderTook, laggardTook := walkLeaderAndLaggard(scheduler, 1000, 150, 300) - require.Equal(t, []int64{10, 20}, store.offeredTargets()) + require.Equal(t, []int64{300, 600, 900}, leaderTook) + require.Equal(t, []int64{300, 600}, laggardTook, "the laggard has yet to reach 900") } -// The gate runs from the checkpoint finishing, not from its target being dispatched. A store that is -// slow to reach its target would otherwise spend the gate's window getting there, and the next -// checkpoint would follow it by less than the configured gap. -func TestTheMinTimeGateIsTimedFromCompletion(t *testing.T) { - store := newFakeStore(0) - scheduler := pacedScheduler(t, map[string]*fakeStore{"only": store}) +// The height is held even once the intervals have elapsed, so the leader cannot run ahead onto a +// height the store behind it would never be offered. +func TestAHeightIsHeldUntilEveryStoreHasReportedIt(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) + require.True(t, scheduler.ShouldCheckpoint("ss", 100)) + scheduler.MarkCheckpointComplete("sc", 100) - scheduler.scheduleNextCheckpoint() - require.True(t, scheduler.lastCheckpointAt.IsZero(), "dispatching a target does not start the gate") + scheduler.elapseTimeInterval() + require.False(t, scheduler.ShouldCheckpoint("sc", 200), "ss has not reported 100") - store.commitTo(11) - scheduler.scheduleNextCheckpoint() - require.False(t, scheduler.lastCheckpointAt.IsZero(), "the finished checkpoint starts the gate") + scheduler.MarkCheckpointComplete("ss", 100) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 300)) } -func TestAZeroMinTimeLeavesTheBlockIntervalAsTheOnlyPacing(t *testing.T) { - store := newFakeStore(0) - scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": store}) +// A store whose first ask lands while a height is held is not one of the stores that height waits +// for: it may already be past that version, and holding for it would wedge the schedule. +func TestAStoreRegisteringWhileAHeightIsHeldJoinsTheNextOne(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) - scheduler.scheduleNextCheckpoint() - store.commitTo(11) - scheduler.scheduleNextCheckpoint() + require.False(t, scheduler.ShouldCheckpoint("ss", 150), "ss registers by asking") + scheduler.MarkCheckpointComplete("sc", 100) - require.Equal(t, []int64{10, 20}, store.offeredTargets()) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 200)) + scheduler.MarkCheckpointComplete("sc", 200) + scheduler.elapseTimeInterval() + require.False(t, scheduler.ShouldCheckpoint("sc", 300), "200 is now held for ss as well") } // --------------------------------------------------------------------------- -// Lifecycle +// Completion // --------------------------------------------------------------------------- -// The run loop dispatches on its own, which is the one thing the direct-cycle tests above cannot show. -// -// The bound is well inside checkpointPollInterval on purpose: it holds the loop to running its first -// cycle at Start, rather than after sitting out a poll interval first. -func TestStartedSchedulerDispatchesOnItsOwn(t *testing.T) { - store := newFakeStore(0) - scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": store}) - require.NoError(t, scheduler.Start()) - t.Cleanup(func() { require.NoError(t, scheduler.Close()) }) - - require.Eventually(t, func() bool { - return slices.Equal(store.offeredTargets(), []int64{10}) - }, 100*time.Millisecond, time.Millisecond, "the run loop never offered a target") +// A height one store never reports stops the scheduler for good: no later height is picked however +// long the intervals have had to elapse, and nothing recovers short of a restart. This is the cost +// of a store skipping MarkCheckpointComplete, which its doc comment requires on every path. +func TestAnUnreportedHeightStopsTheScheduler(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) + scheduler.MarkCheckpointComplete("sc", 100) + + // ss registered before 200 was picked, then never reports it. + require.False(t, scheduler.ShouldCheckpoint("ss", 150)) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 200)) + scheduler.MarkCheckpointComplete("sc", 200) + + for _, height := range []int64{300, 400, 500} { + scheduler.elapseTimeInterval() + require.False(t, scheduler.ShouldCheckpoint("sc", height), "ss never reported 200") + } } -func TestNewRejectsAnEmptyStoreName(t *testing.T) { - _, err := NewCheckpointScheduler(context.Background(), CheckpointConfig{CheckpointInterval: 10}, - map[string]CheckpointableStore{"": newFakeStore(0)}) - require.ErrorContains(t, err, "store name is required") -} +// Reporting a height whose checkpoint failed is what keeps the scheduler moving, which is why the +// call carries no success flag: a store defers it and the next height comes due as usual. +func TestReportingAFailedHeightKeepsTheSchedulerMoving(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) -func TestNewRejectsANilStore(t *testing.T) { - _, err := NewCheckpointScheduler(context.Background(), CheckpointConfig{CheckpointInterval: 10}, - map[string]CheckpointableStore{"ss": nil}) - require.ErrorContains(t, err, "is nil") -} + scheduler.MarkCheckpointComplete("sc", 100) -func TestNewRejectsANegativeInterval(t *testing.T) { - _, err := NewCheckpointScheduler(context.Background(), CheckpointConfig{CheckpointInterval: -1}, nil) - require.ErrorContains(t, err, "interval must not be negative") + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 200)) } -func TestNewRejectsANegativeMinTime(t *testing.T) { - _, err := NewCheckpointScheduler(context.Background(), CheckpointConfig{ - CheckpointInterval: 10, - MinTimeBetweenCheckpoints: -time.Second, - }, nil) - require.ErrorContains(t, err, "minimum time between checkpoints must not be negative") -} +// The interval runs from the last store finishing rather than the first, so the gap to the next +// checkpoint is not spent by a store that is still writing this one. +func TestTheIntervalRunsFromTheLastStoreCompleting(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) + require.True(t, scheduler.ShouldCheckpoint("ss", 100)) -// A scheduler with no stores has nobody to checkpoint, so run returns without a loop. -func TestAnEmptyStoreSetRunsNoLoop(t *testing.T) { - scheduler := newScheduler(t, 10, nil) + scheduler.MarkCheckpointComplete("sc", 100) + afterFirst := scheduler.checkpointedAt + scheduler.MarkCheckpointComplete("ss", 100) - require.NoError(t, scheduler.Start()) - requireLoopStopped(t, scheduler) - require.False(t, scheduler.CheckpointInProgress()) - require.NoError(t, scheduler.Close()) + require.True(t, scheduler.checkpointedAt.After(afterFirst)) } -// Interval 0 turns checkpointing off. run returns without a loop so a cycle never divides by zero. -func TestAZeroIntervalRunsNoLoop(t *testing.T) { - store := newFakeStore(0) - scheduler := newScheduler(t, 0, map[string]*fakeStore{"only": store}) - - require.NoError(t, scheduler.Start()) - requireLoopStopped(t, scheduler) - require.Empty(t, store.offeredTargets()) - require.False(t, scheduler.CheckpointInProgress()) - require.NoError(t, scheduler.Close()) -} +func TestMarkCheckpointCompleteIgnoresAnotherVersion(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) -func TestCloseIsIdempotentAndSafeBeforeStart(t *testing.T) { - scheduler := newScheduler(t, 10, nil) + scheduler.MarkCheckpointComplete("sc", 99) + scheduler.MarkCheckpointComplete("sc", 101) - require.NoError(t, scheduler.Close()) - require.NoError(t, scheduler.Close()) - require.ErrorContains(t, scheduler.Start(), "closed") + require.True(t, scheduler.checkpointOutstanding(), "neither version is the height being held") } -func TestCloseStopsTheRunLoop(t *testing.T) { - scheduler := newScheduler(t, 10, map[string]*fakeStore{"only": newFakeStore(0)}) - require.NoError(t, scheduler.Start()) +func TestMarkCheckpointCompleteIgnoresAnUnregisteredStore(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) + require.True(t, scheduler.ShouldCheckpoint("ss", 100)) - require.NoError(t, scheduler.Close()) + scheduler.MarkCheckpointComplete("receipt", 100) - requireLoopStopped(t, scheduler) + require.True(t, scheduler.checkpointOutstanding(), "the height is still held for sc and ss") } -// Cancelling the context ends the run loop. Asserted as the loop exiting rather than as an absence of -// further offers: the loop selects over the ticker and the cancellation together, so a cycle already -// runnable at the moment of cancellation may still run, and a test forbidding that fails a few runs in -// a thousand. -func TestCancellingTheContextStopsTheRunLoop(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - scheduler, err := NewCheckpointScheduler(ctx, CheckpointConfig{CheckpointInterval: 10}, - map[string]CheckpointableStore{"only": newFakeStore(0)}) - require.NoError(t, err) - require.NoError(t, scheduler.Start()) - - cancel() - - requireLoopStopped(t, scheduler) - require.NoError(t, scheduler.Close()) +// A store reporting a height it already reported must not restart the intervals, which would push +// the next checkpoint out every time a straggler that took the height late reports. +func TestARepeatedReportDoesNotRestartTheIntervals(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) + + scheduler.MarkCheckpointComplete("sc", 100) + completedAt := scheduler.checkpointedAt + scheduler.MarkCheckpointComplete("sc", 100) + + require.Equal(t, completedAt, scheduler.checkpointedAt) } diff --git a/sei-db/controller/checkpointable_store.go b/sei-db/controller/checkpointable_store.go deleted file mode 100644 index 6ab6ff4384..0000000000 --- a/sei-db/controller/checkpointable_store.go +++ /dev/null @@ -1,24 +0,0 @@ -package controller - -// CheckpointableStore is a store whose checkpoints are scheduled by the CheckpointScheduler. -// -// A checkpoint is a point-in-time snapshot of the store at one version. The scheduler decides which -// version that is; a store decides how to create it. -type CheckpointableStore interface { - // ScheduleCheckpoint records that this store should checkpoint at targetVersion. It returns once - // the request is recorded; the store performs the checkpoint when its own write path reaches that - // version. A height the store cannot take — already at or past it, or one already pending — is - // ignored rather than failed: the scheduler has sent the task, and a wrong height must not - // become a nearby height instead. - ScheduleCheckpoint(targetVersion int64) - - // LatestVersion returns the newest version this store has committed, 0 when it has committed - // nothing. The scheduler picks a target above every store's answer, so a store that reports a - // version it has not durably reached is asking to be handed a target it will never see. A - // scheduled checkpoint is not treated as finished until every store reports a version strictly - // above it: stores bump this on commit and only then start the checkpoint write. - LatestVersion() int64 - - // CheckpointInProgress reports whether a checkpoint this store is writing has yet to finish. - CheckpointInProgress() bool -} From fc896ec4093958ad6ab4299aac92a857a420caa0 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Thu, 27 Aug 2026 22:34:47 -0700 Subject: [PATCH 05/11] Fix the block interval logic --- sei-db/config/checkpoint_config.go | 3 ++- sei-db/controller/checkpoint_scheduler.go | 14 +++++++------ .../controller/checkpoint_scheduler_test.go | 21 ++++++++++++++----- 3 files changed, 26 insertions(+), 12 deletions(-) diff --git a/sei-db/config/checkpoint_config.go b/sei-db/config/checkpoint_config.go index 075e39e394..35a7ea1c76 100644 --- a/sei-db/config/checkpoint_config.go +++ b/sei-db/config/checkpoint_config.go @@ -14,7 +14,8 @@ type CheckpointConfig struct { // TimeInterval is the wall-clock gap between checkpoints, measured from the last one completing. TimeInterval time.Duration - // BlockInterval is the block gap between checkpoints, measured from the last height picked. + // BlockInterval places checkpoints on multiples of itself: at 1000, heights 1000, 2000, 3000 + // and so on are eligible. BlockInterval int64 } diff --git a/sei-db/controller/checkpoint_scheduler.go b/sei-db/controller/checkpoint_scheduler.go index 3f3e30ddca..30ef884572 100644 --- a/sei-db/controller/checkpoint_scheduler.go +++ b/sei-db/controller/checkpoint_scheduler.go @@ -28,9 +28,11 @@ var checkpointLogger = seilog.NewLogger("db", "checkpoint") // A no is final and covers every height under it, so a version refused to one store is refused to // all of them; a yes holds for every store that reaches that height before it is replaced. // -// A time interval, a block interval, or both may be set. With both set a height has to clear both; -// a value of 0 or less is unused, and with neither set checkpointing is off. Both are measured from -// the last checkpoint, and from the scheduler's creation before there is one. +// A time interval, a block interval, or both may be set, and a value of 0 or less is unused. The +// block interval places heights on its multiples, so they stay on the same grid however far a +// checkpoint runs late. The time interval is measured from the last checkpoint completing, and from +// the scheduler's creation before there is one. With both set a height has to satisfy both, which +// makes it the first multiple reached after the time has passed. With neither, checkpointing is off. // // Every store has to ask at each version it commits. One asking at only some of them never reaches // the height being held, which stops the node checkpointing rather than only that store. @@ -133,11 +135,11 @@ func (s *CheckpointScheduler) alreadyRejected(version int64) bool { return version < s.nextCheckpointVersion || version <= s.rejectedVersion } -// hasReachedNextInterval reports whether every configured interval has passed for version. +// hasReachedNextInterval reports whether version satisfies every configured interval. func (s *CheckpointScheduler) hasReachedNextInterval(version int64) bool { timeElapsed := s.config.TimeInterval <= 0 || time.Since(s.checkpointedAt) >= s.config.TimeInterval - blocksElapsed := s.config.BlockInterval <= 0 || version-s.nextCheckpointVersion >= s.config.BlockInterval - return timeElapsed && blocksElapsed + onBlockBoundary := s.config.BlockInterval <= 0 || version%s.config.BlockInterval == 0 + return timeElapsed && onBlockBoundary } // pickCheckpointHeight makes version the height in hand, held for the stores registered now. One diff --git a/sei-db/controller/checkpoint_scheduler_test.go b/sei-db/controller/checkpoint_scheduler_test.go index 99e6ba5646..662eb6d5a6 100644 --- a/sei-db/controller/checkpoint_scheduler_test.go +++ b/sei-db/controller/checkpoint_scheduler_test.go @@ -65,6 +65,17 @@ func TestTheFirstHeightWaitsForTheBlockInterval(t *testing.T) { require.True(t, scheduler.ShouldCheckpoint("sc", 100)) } +// Heights sit on multiples of the block interval rather than a count from the last one, so a node +// that starts mid-grid still checkpoints at the same heights as one that did not. +func TestBlockIntervalHeightsAreMultiplesOfTheInterval(t *testing.T) { + scheduler := newScheduler(0, 100) + + for _, offGrid := range []int64{4001, 4050, 4099} { + require.False(t, scheduler.ShouldCheckpoint("sc", offGrid)) + } + require.True(t, scheduler.ShouldCheckpoint("sc", 4100)) +} + // With both set a height has to clear both, so the tighter one paces the cadence. func TestBothIntervalsMustElapseWhenBothAreSet(t *testing.T) { scheduler := newScheduler(time.Hour, 10) @@ -188,16 +199,16 @@ func walkLeaderAndLaggard( } // A store lagging by more than the block interval still takes every height the leader takes, since -// the height is held until it arrives. The heights are no longer spaced by the interval alone: each -// one waits out the lag as well, which is the cadence cost of keeping the stores together. +// the height is held until it arrives. Waiting for it costs whole boundaries rather than shifting +// the grid: 300 passes while 200 is still held, so the next checkpoint is 400. func TestEveryStoreTakesTheSameHeightsHoweverFarBehind(t *testing.T) { scheduler := newScheduler(0, 100) leaderTook, laggardTook := walkLeaderAndLaggard(scheduler, 1000, 150, 0) - require.Equal(t, []int64{100, 200, 351, 502, 653, 804, 955}, leaderTook) - require.Equal(t, []int64{200, 351, 502, 653, 804}, laggardTook, - "100 predates the laggard's first ask, and it has yet to reach 955") + require.Equal(t, []int64{100, 200, 400, 600, 800, 1000}, leaderTook) + require.Equal(t, []int64{200, 400, 600, 800}, laggardTook, + "100 predates the laggard's first ask, and it has yet to reach 1000") } func TestEveryStoreTakesTheSameHeightsOnATimeInterval(t *testing.T) { From 5b63a857301efc96081d25d505d5dc7c91205a08 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Fri, 28 Aug 2026 10:00:32 -0700 Subject: [PATCH 06/11] Polish comments --- sei-db/controller/checkpoint_scheduler.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sei-db/controller/checkpoint_scheduler.go b/sei-db/controller/checkpoint_scheduler.go index 30ef884572..09ae6b1cc3 100644 --- a/sei-db/controller/checkpoint_scheduler.go +++ b/sei-db/controller/checkpoint_scheduler.go @@ -25,8 +25,10 @@ var checkpointLogger = seilog.NewLogger("db", "checkpoint") // handed that same height rather than one that has moved on, and heights arrive no faster than the // slowest store reaches them. // -// A no is final and covers every height under it, so a version refused to one store is refused to -// all of them; a yes holds for every store that reaches that height before it is replaced. +// Answers hold across stores. Refusing a height refuses every height below it too — once 100 is +// refused, so are 99 and 98 — which stops a lagging store from taking a lower height the moment an +// interval elapses. An accepted height keeps its answer until it is replaced, so a store that +// reaches it late is told the same as the store that got there first. // // A time interval, a block interval, or both may be set, and a value of 0 or less is unused. The // block interval places heights on its multiples, so they stay on the same grid however far a From 6a467e56ba6f0d5dde194c0871eaae5761428e5e Mon Sep 17 00:00:00 2001 From: YimingZang Date: Fri, 28 Aug 2026 10:53:55 -0700 Subject: [PATCH 07/11] Fix comments --- sei-db/controller/checkpoint_scheduler.go | 110 +++++++++++------- .../controller/checkpoint_scheduler_test.go | 49 ++++++-- 2 files changed, 107 insertions(+), 52 deletions(-) diff --git a/sei-db/controller/checkpoint_scheduler.go b/sei-db/controller/checkpoint_scheduler.go index 09ae6b1cc3..2681b089fa 100644 --- a/sei-db/controller/checkpoint_scheduler.go +++ b/sei-db/controller/checkpoint_scheduler.go @@ -4,7 +4,6 @@ package controller import ( - "maps" "sync" "time" @@ -15,38 +14,36 @@ import ( var checkpointLogger = seilog.NewLogger("db", "checkpoint") -// CheckpointScheduler picks the heights every store checkpoints at, so the stores of one node hold -// checkpoints of the same versions rather than of whatever version each happened to reach. +// CheckpointScheduler picks the heights every store on a node checkpoints at. // -// Stores ask ShouldCheckpoint at every version they commit and call MarkCheckpointComplete once -// that version's checkpoint has finished, whether or not it succeeded: every yes obliges a report. -// A store joins the schedule by asking, and a height is held until every store registered when it -// was picked, along with any that took it afterwards, has reported. A store behind the others is -// handed that same height rather than one that has moved on, and heights arrive no faster than the -// slowest store reaches them. +// Stores ask ShouldCheckpoint at each live commit and call MarkCheckpointComplete when that +// checkpoint finishes, success or failure. Replay, WAL catch-up, and state-sync must not ask. +// ShouldCheckpoint will register the store. +// +// A height is held until every store registered when it was picked, along with any that took it +// afterwards, has reported. A store behind the others is handed that same height rather than one +// that has moved on, and heights arrive no faster than the slowest store reaches them. // // Answers hold across stores. Refusing a height refuses every height below it too — once 100 is // refused, so are 99 and 98 — which stops a lagging store from taking a lower height the moment an // interval elapses. An accepted height keeps its answer until it is replaced, so a store that // reaches it late is told the same as the store that got there first. // -// A time interval, a block interval, or both may be set, and a value of 0 or less is unused. The -// block interval places heights on its multiples, so they stay on the same grid however far a -// checkpoint runs late. The time interval is measured from the last checkpoint completing, and from -// the scheduler's creation before there is one. With both set a height has to satisfy both, which -// makes it the first multiple reached after the time has passed. With neither, checkpointing is off. +// A store that passes a held height without taking it — one whose version jumped over it — is +// released from that height when it next asks, so it costs that store one checkpoint rather than +// stopping the schedule. A store that stops asking altogether holds the height indefinitely. // -// Every store has to ask at each version it commits. One asking at only some of them never reaches -// the height being held, which stops the node checkpointing rather than only that store. +// A time interval, a block interval, or both may be set. +// Both set means both must hold; neither disables checkpointing. type CheckpointScheduler struct { mu sync.Mutex config config.CheckpointConfig // registered holds every store that has asked, which is how a store joins the schedule. registered map[string]struct{} - // awaiting holds the stores the current height is held for that have yet to report it. Empty - // means no checkpoint is outstanding. - awaiting map[string]struct{} + // awaiting maps each store nextCheckpointVersion is held for to whether that store has taken + // the height yet. Empty means every store has checkpointed this height. + awaiting map[string]bool // nextCheckpointVersion is the height stores checkpoint at, 0 before the first one is picked. // It stays set once complete, so a store that has yet to reach it is still given that height. @@ -65,13 +62,14 @@ func NewCheckpointScheduler(cfg config.CheckpointConfig) *CheckpointScheduler { return &CheckpointScheduler{ config: cfg, registered: make(map[string]struct{}), - awaiting: make(map[string]struct{}), + awaiting: make(map[string]bool), checkpointedAt: time.Now(), } } // ShouldCheckpoint reports whether version is a height for store to checkpoint at, registering -// store with the schedule when this is its first ask. +// store with the schedule when this is its first ask. Call it only from the live commit path, +// never during replay, WAL catch-up, or state-sync forward-fill. func (s *CheckpointScheduler) ShouldCheckpoint(store string, version int64) bool { if !s.checkpointEnabled() || version <= 0 { return false @@ -88,17 +86,18 @@ func (s *CheckpointScheduler) ShouldCheckpoint(store string, version int64) bool if s.alreadyRejected(version) { return false } - if s.checkpointOutstanding() || !s.hasReachedNextInterval(version) { + s.skipPastHeight(store) + if !(s.allStoresCheckpointed() && s.hasReachedNextInterval(version)) { s.rejectedVersion = version return false } - s.pickCheckpointHeight(version) + s.pickCheckpointHeight(store, version) return true } -// MarkCheckpointComplete records that store has finished the current height. Both intervals start -// once every store the height is held for has reported it. Any other version is ignored, as is a -// repeat from the same store. +// MarkCheckpointComplete records that store has finished the height it was given. Both intervals +// start once every store that height is held for has reported it. Any other version is ignored, as +// is a repeat from the same store. // // A store that took a height must report it on every path out of the checkpoint, a failed one // included, which in practice means deferring the call. No further height is picked while one store @@ -110,29 +109,24 @@ func (s *CheckpointScheduler) MarkCheckpointComplete(store string, version int64 if version != s.nextCheckpointVersion { return } - if _, outstanding := s.awaiting[store]; !outstanding { + if _, awaited := s.awaiting[store]; !awaited { return } delete(s.awaiting, store) - - if s.checkpointOutstanding() { - return - } - s.checkpointedAt = time.Now() - checkpointLogger.Info("checkpoint complete", "version", version) + s.updateCheckpointTime() } func (s *CheckpointScheduler) checkpointEnabled() bool { return s.config.TimeInterval > 0 || s.config.BlockInterval > 0 } -func (s *CheckpointScheduler) checkpointOutstanding() bool { - return len(s.awaiting) > 0 +func (s *CheckpointScheduler) allStoresCheckpointed() bool { + return len(s.awaiting) == 0 } -// alreadyRejected reports whether version has been turned down already, either by trailing the -// height in hand or by being asked and refused. Standing by that no is what stops two stores asking -// at different moments from splitting on one version. +// alreadyRejected reports whether version has been turned down already, either by trailing +// nextCheckpointVersion or by being asked and refused. Standing by that no is what stops two stores +// asking at different moments from splitting on one version. func (s *CheckpointScheduler) alreadyRejected(version int64) bool { return version < s.nextCheckpointVersion || version <= s.rejectedVersion } @@ -144,15 +138,41 @@ func (s *CheckpointScheduler) hasReachedNextInterval(version int64) bool { return timeElapsed && onBlockBoundary } -// pickCheckpointHeight makes version the height in hand, held for the stores registered now. One -// registering later is not held for here, at a version it may already have passed, and is held for -// from the moment it takes a height. -func (s *CheckpointScheduler) pickCheckpointHeight(version int64) { +// pickCheckpointHeight sets nextCheckpointVersion to version and holds it for the stores registered +// now, of which store is the one that has taken it. One registering later is not held for here, at +// a version it may already have passed, and is held for from the moment it takes a height. +func (s *CheckpointScheduler) pickCheckpointHeight(store string, version int64) { s.nextCheckpointVersion = version - s.awaiting = maps.Clone(s.registered) + s.awaiting = make(map[string]bool, len(s.registered)) + for name := range s.registered { + s.awaiting[name] = false + } + s.awaiting[store] = true } -// holdFor keeps the height in hand from being replaced until store has reported it. +// holdFor records that store has taken nextCheckpointVersion, which is held until store reports it. func (s *CheckpointScheduler) holdFor(store string) { - s.awaiting[store] = struct{}{} + s.awaiting[store] = true +} + +// skipPastHeight releases the hold store has on nextCheckpointVersion, for a store asking above +// that height and so past it for good. A store that took the height keeps its hold: it may be +// writing that checkpoint while it commits later versions, and the intervals wait for it. +// +// Callers must have ruled out versions at or under nextCheckpointVersion, which is what makes an +// ask proof that the store has passed it. +func (s *CheckpointScheduler) skipPastHeight(store string) { + if took, awaited := s.awaiting[store]; !awaited || took { + return + } + delete(s.awaiting, store) + s.updateCheckpointTime() +} + +// updateCheckpointTime records when nextCheckpointVersion completed, once every store has +// checkpointed this height. +func (s *CheckpointScheduler) updateCheckpointTime() { + if s.allStoresCheckpointed() { + s.checkpointedAt = time.Now() + } } diff --git a/sei-db/controller/checkpoint_scheduler_test.go b/sei-db/controller/checkpoint_scheduler_test.go index 662eb6d5a6..eaaffb9bb7 100644 --- a/sei-db/controller/checkpoint_scheduler_test.go +++ b/sei-db/controller/checkpoint_scheduler_test.go @@ -254,20 +254,55 @@ func TestAStoreRegisteringWhileAHeightIsHeldJoinsTheNextOne(t *testing.T) { require.False(t, scheduler.ShouldCheckpoint("sc", 300), "200 is now held for ss as well") } +// A store whose version jumps over the held height would otherwise hold it forever, since it never +// asks at that height again. Asking above it is proof it has passed, so it is released and only +// that store misses the checkpoint. +func TestAStoreThatPassesTheHeldHeightIsReleased(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + require.False(t, scheduler.ShouldCheckpoint("ss", 98), "ss registers before the interval elapses") + + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) + scheduler.MarkCheckpointComplete("sc", 100) + + require.False(t, scheduler.ShouldCheckpoint("ss", 101), "ss jumped from 99 to 101") + + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 200), "100 is no longer held for ss") +} + +// A store commits later versions while its own checkpoint of the held height is still running, so +// asking above that height is not proof it skipped it. Releasing it there would start the intervals +// while it is mid-write. +func TestAStoreWritingTheHeldHeightIsNotReleased(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) + + require.False(t, scheduler.ShouldCheckpoint("sc", 101), "sc is still writing 100") + + scheduler.elapseTimeInterval() + require.False(t, scheduler.ShouldCheckpoint("sc", 200), "100 is still held for sc") + + scheduler.MarkCheckpointComplete("sc", 100) + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 300)) +} + // --------------------------------------------------------------------------- // Completion // --------------------------------------------------------------------------- -// A height one store never reports stops the scheduler for good: no later height is picked however -// long the intervals have had to elapse, and nothing recovers short of a restart. This is the cost -// of a store skipping MarkCheckpointComplete, which its doc comment requires on every path. -func TestAnUnreportedHeightStopsTheScheduler(t *testing.T) { +// A store that stops asking holds its height for good: no later height is picked however long the +// intervals have had to elapse, and nothing recovers short of a restart. A store still asking is +// released once it passes the height, so this is the stalled-store case rather than the skipped one. +func TestAStalledStoreStopsTheScheduler(t *testing.T) { scheduler := newScheduler(time.Hour, 0) scheduler.elapseTimeInterval() require.True(t, scheduler.ShouldCheckpoint("sc", 100)) scheduler.MarkCheckpointComplete("sc", 100) - // ss registered before 200 was picked, then never reports it. + // ss registers, is held for 200, then stops asking entirely. require.False(t, scheduler.ShouldCheckpoint("ss", 150)) scheduler.elapseTimeInterval() require.True(t, scheduler.ShouldCheckpoint("sc", 200)) @@ -315,7 +350,7 @@ func TestMarkCheckpointCompleteIgnoresAnotherVersion(t *testing.T) { scheduler.MarkCheckpointComplete("sc", 99) scheduler.MarkCheckpointComplete("sc", 101) - require.True(t, scheduler.checkpointOutstanding(), "neither version is the height being held") + require.False(t, scheduler.allStoresCheckpointed(), "neither version is the height being held") } func TestMarkCheckpointCompleteIgnoresAnUnregisteredStore(t *testing.T) { @@ -326,7 +361,7 @@ func TestMarkCheckpointCompleteIgnoresAnUnregisteredStore(t *testing.T) { scheduler.MarkCheckpointComplete("receipt", 100) - require.True(t, scheduler.checkpointOutstanding(), "the height is still held for sc and ss") + require.False(t, scheduler.allStoresCheckpointed(), "the height is still held for sc and ss") } // A store reporting a height it already reported must not restart the intervals, which would push From 52e7f571d476d5ac603915498a40c38c7b6c17be Mon Sep 17 00:00:00 2001 From: YimingZang Date: Fri, 28 Aug 2026 11:00:51 -0700 Subject: [PATCH 08/11] Better comment --- sei-db/controller/checkpoint_scheduler.go | 21 ++++++++------------- 1 file changed, 8 insertions(+), 13 deletions(-) diff --git a/sei-db/controller/checkpoint_scheduler.go b/sei-db/controller/checkpoint_scheduler.go index 2681b089fa..ab54a214fc 100644 --- a/sei-db/controller/checkpoint_scheduler.go +++ b/sei-db/controller/checkpoint_scheduler.go @@ -20,21 +20,16 @@ var checkpointLogger = seilog.NewLogger("db", "checkpoint") // checkpoint finishes, success or failure. Replay, WAL catch-up, and state-sync must not ask. // ShouldCheckpoint will register the store. // -// A height is held until every store registered when it was picked, along with any that took it -// afterwards, has reported. A store behind the others is handed that same height rather than one -// that has moved on, and heights arrive no faster than the slowest store reaches them. +// The goal is that every registered store either checkpoints on the given exact same height, or none of them do. +// That is achieved by holding a yes until every store registered has taken the checkpoint later on. +// So a lagging store is still told yes at that height; and by treating a no as a floor: once 100 is refused, +// so are 99 and 98, so a store a few blocks behind cannot take a lower height the moment an interval elapses. // -// Answers hold across stores. Refusing a height refuses every height below it too — once 100 is -// refused, so are 99 and 98 — which stops a lagging store from taking a lower height the moment an -// interval elapses. An accepted height keeps its answer until it is replaced, so a store that -// reaches it late is told the same as the store that got there first. +// A store that passes a held height without taking the checkpoint — one whose version jumped over it — is +// released from that height when it asks next time, so it costs that store one checkpoint rather than +// stopping the whole schedule. A store that stops asking altogether could hold the height indefinitely. // -// A store that passes a held height without taking it — one whose version jumped over it — is -// released from that height when it next asks, so it costs that store one checkpoint rather than -// stopping the schedule. A store that stops asking altogether holds the height indefinitely. -// -// A time interval, a block interval, or both may be set. -// Both set means both must hold; neither disables checkpointing. +// A time interval, a block interval, or both may be set. Both set means both must hold; neither disables checkpointing. type CheckpointScheduler struct { mu sync.Mutex config config.CheckpointConfig From 07f918b22cf7c32ce9f569c68c1380888234f12a Mon Sep 17 00:00:00 2001 From: YimingZang Date: Fri, 28 Aug 2026 11:10:25 -0700 Subject: [PATCH 09/11] Better ways to explain --- sei-db/controller/checkpoint_scheduler.go | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/sei-db/controller/checkpoint_scheduler.go b/sei-db/controller/checkpoint_scheduler.go index ab54a214fc..5ee0404eef 100644 --- a/sei-db/controller/checkpoint_scheduler.go +++ b/sei-db/controller/checkpoint_scheduler.go @@ -20,10 +20,9 @@ var checkpointLogger = seilog.NewLogger("db", "checkpoint") // checkpoint finishes, success or failure. Replay, WAL catch-up, and state-sync must not ask. // ShouldCheckpoint will register the store. // -// The goal is that every registered store either checkpoints on the given exact same height, or none of them do. -// That is achieved by holding a yes until every store registered has taken the checkpoint later on. -// So a lagging store is still told yes at that height; and by treating a no as a floor: once 100 is refused, -// so are 99 and 98, so a store a few blocks behind cannot take a lower height the moment an interval elapses. +// The goal is that every registered store either checkpoints on the exact same height, or none of them do. +// That is achieved by holding a yes until every registered store has taken the checkpoint on the same height. +// So a lagging store is still told yes at that height, or no if the faster stores rejected that height already. // // A store that passes a held height without taking the checkpoint — one whose version jumped over it — is // released from that height when it asks next time, so it costs that store one checkpoint rather than From 4d6cb0eaa2b25e7ef939f129e55cab4faeb14be3 Mon Sep 17 00:00:00 2001 From: YimingZang Date: Fri, 28 Aug 2026 11:12:29 -0700 Subject: [PATCH 10/11] Fix a bug for skip height logic --- sei-db/controller/checkpoint_scheduler.go | 7 +++---- sei-db/controller/checkpoint_scheduler_test.go | 17 +++++++++++++++++ 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/sei-db/controller/checkpoint_scheduler.go b/sei-db/controller/checkpoint_scheduler.go index 5ee0404eef..de447da87b 100644 --- a/sei-db/controller/checkpoint_scheduler.go +++ b/sei-db/controller/checkpoint_scheduler.go @@ -77,10 +77,12 @@ func (s *CheckpointScheduler) ShouldCheckpoint(store string, version int64) bool s.holdFor(store) return true } + if version > s.nextCheckpointVersion { + s.skipPastHeight(store) + } if s.alreadyRejected(version) { return false } - s.skipPastHeight(store) if !(s.allStoresCheckpointed() && s.hasReachedNextInterval(version)) { s.rejectedVersion = version return false @@ -152,9 +154,6 @@ func (s *CheckpointScheduler) holdFor(store string) { // skipPastHeight releases the hold store has on nextCheckpointVersion, for a store asking above // that height and so past it for good. A store that took the height keeps its hold: it may be // writing that checkpoint while it commits later versions, and the intervals wait for it. -// -// Callers must have ruled out versions at or under nextCheckpointVersion, which is what makes an -// ask proof that the store has passed it. func (s *CheckpointScheduler) skipPastHeight(store string) { if took, awaited := s.awaiting[store]; !awaited || took { return diff --git a/sei-db/controller/checkpoint_scheduler_test.go b/sei-db/controller/checkpoint_scheduler_test.go index eaaffb9bb7..2110667b05 100644 --- a/sei-db/controller/checkpoint_scheduler_test.go +++ b/sei-db/controller/checkpoint_scheduler_test.go @@ -271,6 +271,23 @@ func TestAStoreThatPassesTheHeldHeightIsReleased(t *testing.T) { require.True(t, scheduler.ShouldCheckpoint("sc", 200), "100 is no longer held for ss") } +// A no is recorded before the jumper asks, so skipPastHeight never runs if it sits behind +// alreadyRejected. The jumper must still be released: otherwise it stays in awaiting forever. +func TestAStoreThatJumpsIsReleasedAfterARefusedHeightAbove(t *testing.T) { + scheduler := newScheduler(time.Hour, 0) + require.False(t, scheduler.ShouldCheckpoint("ss", 98), "ss registers before the interval elapses") + + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 100)) + scheduler.MarkCheckpointComplete("sc", 100) + + require.False(t, scheduler.ShouldCheckpoint("sc", 101), "sc asks above 100 first") + require.False(t, scheduler.ShouldCheckpoint("ss", 101), "ss lands on a height already refused") + + scheduler.elapseTimeInterval() + require.True(t, scheduler.ShouldCheckpoint("sc", 200), "ss must have been released from 100") +} + // A store commits later versions while its own checkpoint of the held height is still running, so // asking above that height is not proof it skipped it. Releasing it there would start the intervals // while it is mid-write. From e70dc9a4ca125cfcd74d47d2778085abfbb2d17d Mon Sep 17 00:00:00 2001 From: YimingZang Date: Fri, 28 Aug 2026 11:28:15 -0700 Subject: [PATCH 11/11] Fix lint --- sei-db/controller/checkpoint_scheduler.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sei-db/controller/checkpoint_scheduler.go b/sei-db/controller/checkpoint_scheduler.go index de447da87b..da48207977 100644 --- a/sei-db/controller/checkpoint_scheduler.go +++ b/sei-db/controller/checkpoint_scheduler.go @@ -83,7 +83,7 @@ func (s *CheckpointScheduler) ShouldCheckpoint(store string, version int64) bool if s.alreadyRejected(version) { return false } - if !(s.allStoresCheckpointed() && s.hasReachedNextInterval(version)) { + if !s.allStoresCheckpointed() || !s.hasReachedNextInterval(version) { s.rejectedVersion = version return false }