diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index 89e0bbf..f20370c 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -91,6 +91,12 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. - **A failed checkpoint no longer makes the next step conflict.** `Tick`, `RunService` and `DeliverToActor` now advance the next step as soon as the append lands. +- **`WithCheckpointEvery` now bounds the tail of a timer-driven instance.** A `Tick` that + fires timers takes several step ordinals, so it could skip the interval forever and the + tail grew without limit. A checkpoint now lands on the first step that reaches or passes + the interval. Such an instance now compacts, so on a Store without `HistoryStore`, + `StateAt` can't reach the steps before its latest checkpoint: it returns the + checkpoint's later state with a nil error, or `ErrStepOutOfRange` if the tail is empty. - **Timers survive recovery.** `Start` now arms the initial state's timers, which it never did, so a new instance starting in a timed state fires a timer it didn't before. It carries them in the baseline; an instance started earlier keeps its initial timer diff --git a/durable/runner.go b/durable/runner.go index 42f8ff9..79345ca 100644 --- a/durable/runner.go +++ b/durable/runner.go @@ -578,9 +578,12 @@ func (h *Handle[S, E, C]) commitStep(ctx context.Context, rec *Record, effects [ return h.checkpointStep(ctx, rec.Step, rec) } -// checkpointDue reports whether the checkpoint policy lands a checkpoint at step. -func (h *Handle[S, E, C]) checkpointDue(step int) bool { - return h.runner.cfg.checkpointEvery > 0 && (step+1)%h.runner.cfg.checkpointEvery == 0 +// checkpointDue reports whether rec crosses a checkpoint boundary. It compares the +// ordinals rec spans rather than testing rec's own Step, because a tick barrier consumes +// several ordinals and can step over the one a fixed multiple would land on. +func (h *Handle[S, E, C]) checkpointDue(rec *Record) bool { + every := h.runner.cfg.checkpointEvery + return every > 0 && rec.followingStep()/every > rec.Step/every } // appendStep marshals the due checkpoint snapshot into rec (so the Record carries it) @@ -590,7 +593,7 @@ func (h *Handle[S, E, C]) checkpointDue(step int) bool { // — so a crash between append and dispatch leaves the effect recoverable from the // still-present tail rather than orphaned behind the compaction. func (h *Handle[S, E, C]) appendStep(ctx context.Context, step int, rec *Record) error { - if h.checkpointDue(step) { + if h.checkpointDue(rec) { snap, err := state.MarshalSnapshot(h.inst.Snapshot()) if err != nil { return fmt.Errorf("durable: marshaling checkpoint at step %d for %q: %w", step, h.id, err) @@ -615,7 +618,7 @@ func (h *Handle[S, E, C]) appendStep(ctx context.Context, step int, rec *Record) // step's domain effects have been dispatched, so the compaction never discards a // not-yet-dispatched effect. func (h *Handle[S, E, C]) checkpointStep(ctx context.Context, step int, rec *Record) error { - if !h.checkpointDue(step) { + if !h.checkpointDue(rec) { return nil } if err := h.runner.store.Checkpoint(ctx, h.id, rec.Snapshot, step, WithCheckpointFence(h.runner.cfg.fence)); err != nil { diff --git a/durable/runner_options.go b/durable/runner_options.go index f6a6245..776c854 100644 --- a/durable/runner_options.go +++ b/durable/runner_options.go @@ -16,9 +16,10 @@ type Option[S comparable, E comparable, C any] func(*runnerConfig[S, E, C]) // runnerConfig holds resolved Option state for a Runner. type runnerConfig[S comparable, E comparable, C any] struct { - // checkpointEvery is the checkpoint policy: persist a full Snapshot and - // compact the journal tail every N fired steps. Zero (the default) takes no - // periodic checkpoint, so recovery replays from the start baseline. + // checkpointEvery is the checkpoint policy: persist a full Snapshot and compact the + // journal tail after each step that brings the next ordinal to or past a multiple of + // N. Zero (the default) takes no periodic checkpoint, so recovery replays from the + // start baseline. checkpointEvery int // eventCodec reconstructs a recorded event for replay; it defaults to JSON. eventCodec EventCodec[E] @@ -51,10 +52,12 @@ type runnerConfig[S comparable, E comparable, C any] struct { } // WithCheckpointEvery sets the checkpoint policy: the Runner persists a full -// Snapshot and compacts the journal tail every n fired steps, bounding both the -// stored tail and the replay length on recovery. A non-positive n disables -// periodic checkpointing (the default), so recovery replays the whole run from -// the start baseline. +// Snapshot and compacts the journal tail after each step that brings the next step +// ordinal to or past a multiple of n, so n=8 lands on Step 7 for plain steps. This bounds +// both the stored tail and the replay length on recovery. Each recorded step takes one +// ordinal, except a Tick that fires timers, which takes one plus one per timer fired. A +// non-positive n disables periodic checkpointing (the default), so recovery replays the +// whole run from the start baseline. func WithCheckpointEvery[S comparable, E comparable, C any](n int) Option[S, E, C] { return func(c *runnerConfig[S, E, C]) { if n > 0 { diff --git a/durable/timer_checkpoint_test.go b/durable/timer_checkpoint_test.go index dee7476..6399cd9 100644 --- a/durable/timer_checkpoint_test.go +++ b/durable/timer_checkpoint_test.go @@ -641,7 +641,7 @@ func TestRecover_ChainedTimerAfterSlowAction_NotFiredEarly(t *testing.T) { id := durable.InstanceID("timer-1") opts := []durable.Option[string, string, *timerCtx]{ durable.WithRunnerClock[string, string, *timerCtx](clk), - durable.WithCheckpointEvery[string, string, *timerCtx](2), + durable.WithCheckpointEvery[string, string, *timerCtx](3), // the first tick checkpoints, the second stays in the tail } h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, &timerCtx{}, state.WithInitialState("idle")) if err != nil { @@ -654,13 +654,17 @@ func TestRecover_ChainedTimerAfterSlowAction_NotFiredEarly(t *testing.T) { if _, err = h.Tick(ctx); err != nil { t.Fatalf("Tick: %v", err) } - if _, err = h.Fire(ctx, "poke"); err != nil { - t.Fatalf("Fire(poke): %v", err) - } clk.now = epoch.Add(other) if _, err = h.Tick(ctx); err != nil { t.Fatalf("Tick: %v", err) } + _, tail, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(tail) == 0 { + t.Fatal("tail is empty, want the second tick left for replay") + } recovered, err := durable.Recover(ctx, m, store, id, opts...) if err != nil { t.Fatalf("Recover: %v", err) @@ -920,3 +924,48 @@ func TestRecover_TimerDueAfterTheFirstRead_NotPersisted(t *testing.T) { t.Fatalf("recovered Tick fired %d timers, want 0", len(again)) } } + +// TestTimerCheckpoint_BarriersCannotStepOverTheInterval proves a checkpoint still lands +// when tick barriers keep stepping over the configured multiple. +func TestTimerCheckpoint_BarriersCannotStepOverTheInterval(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := durable.NewMemStore() + id := durable.InstanceID("poll-1") + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *runCtx]{ + durable.WithRunnerClock[string, string, *runCtx](clk), + durable.WithCheckpointEvery[string, string, *runCtx](8), + } + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + // Two Fires put the barriers on even steps, so none lands on the step before a multiple + // of 8, where a plain run checkpoints. + for _, ev := range []string{"start", "poke"} { + if _, err = h.Fire(ctx, ev); err != nil { + t.Fatalf("Fire(%s): %v", ev, err) + } + } + for range 50 { + clk.Advance(time.Second) + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + } + _, tail, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(tail) > 8 { + t.Fatalf("tail holds %d records after 50 ticks, want the checkpoint to bound it", len(tail)) + } + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if live, got := handleSnapshot(t, h), handleSnapshot(t, recovered); got != live { + t.Fatalf("recovered instance differs from live:\n live: %s\n got: %s", live, got) + } +}