fix(durable): land checkpoints when tick barriers step over the interval

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

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 {

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 {

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)
+	}
+}