fix(durable): carry the next step through a checkpoint

durable/CHANGELOG.md

@@ -54,11 +54,17 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.
   The same record is still an idempotent no-op, and an Append with
   `WithIdempotencyKey` still dedups on the key alone. A `FileStore` ledger line from
   before this change has no digest, so its step keeps the old no-op behavior. A
-  recovered instance that reuses a step number now fails with `ErrStepConflict`
-  instead of silently dropping the step.
+  recovered instance that reuses a step number with a different record now fails with
+  `ErrStepConflict` instead of silently dropping the step.
 
 ### Fixed
 
+- **The next step survives a checkpoint that compacts tick barriers.** A checkpoint
+  now records the step that follows it, in a `nextStep` field of its envelope. Recovery
+  used the Trace count, which runs low past a tick barrier, so the next Fire could reuse a
+  recorded step, and high past a multi-Trace actor delivery, which skipped ordinals. A
+  checkpoint written before this change can still reuse a step: a different record fails
+  with `ErrStepConflict`, and an identical one is acknowledged without a new Record.
 - **`WithEventData` survives recovery.** `Handle.Fire` records the payload in the new
   `Record.EventData` and `Record.EventDataError`, and `Recover` and `StateAt` supply it
   on replay. This needs `state.FireEventData`, so `durable` requires `state` v1.1.0 or

durable/checkpoint.go

@@ -61,8 +61,8 @@ type pendingTimerWire struct {
 }
 
 // checkpointEnvelope is the durable layer's wrapper around the bytes a Store keeps
-// for a checkpoint: the opaque state.Snapshot bytes plus the pending-timer table
-// armed at the checkpoint instant. The durable layer fully owns the checkpoint
+// for a checkpoint: the opaque state.Snapshot bytes, the pending-timer table armed at
+// the checkpoint instant, and the next step ordinal. The durable layer fully owns the checkpoint
 // bytes (it both marshals them at Checkpoint and unmarshals them at recover), so
 // wrapping the kernel snapshot in this envelope is additive and invisible to the
 // Store, which treats checkpoint bytes as opaque. A checkpoint with no armed timer
@@ -77,14 +77,21 @@ type checkpointEnvelope struct {
 	// recovery whose compacted tail no longer carries the arming ScheduleAfter can
 	// still re-arm them at their recorded instants.
 	Timers []pendingTimerWire `json:"timers,omitempty"`
+	// NextStep is the step ordinal the first Record after this checkpoint takes. A
+	// recovery whose compacted tail is empty has no Record to derive it from, and the
+	// restored Trace count doesn't match the ordinals: a tick barrier consumes one and
+	// leaves no Trace, and an actor delivery can leave several Traces on one ordinal.
+	// Zero means none was recorded: a Start baseline, or a checkpoint from before this
+	// field.
+	NextStep int `json:"nextStep,omitempty"`
 }
 
 // marshalCheckpoint wraps the marshaled kernel snapshot and the armed pending
 // timers into a checkpoint envelope's bytes. Timers are emitted in scheduleID
 // order so the persisted form is deterministic regardless of the map iteration
-// order they were collected in.
-func marshalCheckpoint(snap []byte, timers map[string]pendingTimer) ([]byte, error) {
-	env := checkpointEnvelope{Snapshot: snap}
+// order they were collected in. nextStep is the ordinal that follows the checkpoint.
+func marshalCheckpoint(snap []byte, timers map[string]pendingTimer, nextStep int) ([]byte, error) {
+	env := checkpointEnvelope{Snapshot: snap, NextStep: nextStep}
 	if len(timers) > 0 {
 		ids := make([]string, 0, len(timers))
 		for id := range timers {
@@ -113,18 +120,19 @@ func marshalCheckpoint(snap []byte, timers map[string]pendingTimer) ([]byte, err
 }
 
 // unmarshalCheckpoint splits a checkpoint envelope's bytes back into the kernel
-// snapshot bytes and the persisted pending-timer wires. It tolerates a bare
-// (pre-envelope) snapshot: bytes that are not a checkpoint envelope are returned
-// as the snapshot with no timers, so an older checkpoint still restores. A
-// checkpoint envelope is distinguished by carrying a non-empty "snapshot" member.
-func unmarshalCheckpoint(b []byte) ([]byte, []pendingTimerWire, error) {
+// snapshot bytes, the persisted pending-timer wires and the next step ordinal. It
+// tolerates a bare (pre-envelope) snapshot: bytes that are not a checkpoint envelope
+// are returned as the snapshot with no timers and a zero ordinal, so an older
+// checkpoint still restores. A checkpoint envelope is distinguished by carrying a
+// non-empty "snapshot" member.
+func unmarshalCheckpoint(b []byte) ([]byte, []pendingTimerWire, int, error) {
 	var env checkpointEnvelope
 	if err := json.Unmarshal(b, &env); err != nil || len(env.Snapshot) == 0 {
 		// Not an envelope (or an envelope without the wrapped snapshot member): treat
 		// the bytes as a bare kernel snapshot, the pre-timer-survival checkpoint form.
-		return b, nil, nil
+		return b, nil, 0, nil
 	}
-	return env.Snapshot, env.Timers, nil
+	return env.Snapshot, env.Timers, env.NextStep, nil
 }
 
 // decodeTimerEvent decodes a persisted timer event wire back into the machine's

durable/checkpoint_internal_test.go

@@ -17,11 +17,11 @@ func TestMarshalCheckpoint_RoundTrip(t *testing.T) {
 	timers := map[string]pendingTimer{
 		"m:a:after:0": {id: "m:a:after:0", due: due, event: "elapsed"},
 	}
-	b, err := marshalCheckpoint(snap, timers)
+	b, err := marshalCheckpoint(snap, timers, 0)
 	if err != nil {
 		t.Fatalf("marshalCheckpoint: %v", err)
 	}
-	gotSnap, gotWires, err := unmarshalCheckpoint(b)
+	gotSnap, gotWires, _, err := unmarshalCheckpoint(b)
 	if err != nil {
 		t.Fatalf("unmarshalCheckpoint: %v", err)
 	}
@@ -45,7 +45,7 @@ func TestMarshalCheckpoint_RoundTrip(t *testing.T) {
 // armed, so an event-driven checkpoint's envelope stays minimal.
 func TestMarshalCheckpoint_NoTimers(t *testing.T) {
 	snap := []byte(`{"machine":"m"}`)
-	b, err := marshalCheckpoint(snap, nil)
+	b, err := marshalCheckpoint(snap, nil, 0)
 	if err != nil {
 		t.Fatalf("marshalCheckpoint: %v", err)
 	}
@@ -64,7 +64,7 @@ func TestMarshalCheckpoint_NoTimers(t *testing.T) {
 // checkpoint still restores.
 func TestUnmarshalCheckpoint_BareSnapshotFallback(t *testing.T) {
 	bare := []byte(`{"machine":"m","current":"a","configuration":["a"]}`)
-	snap, wires, err := unmarshalCheckpoint(bare)
+	snap, wires, _, err := unmarshalCheckpoint(bare)
 	if err != nil {
 		t.Fatalf("unmarshalCheckpoint: %v", err)
 	}
@@ -81,7 +81,7 @@ func TestUnmarshalCheckpoint_BareSnapshotFallback(t *testing.T) {
 // any real decode failure when it unmarshals the kernel snapshot).
 func TestUnmarshalCheckpoint_Garbage(t *testing.T) {
 	garbage := []byte(`not json`)
-	snap, wires, err := unmarshalCheckpoint(garbage)
+	snap, wires, _, err := unmarshalCheckpoint(garbage)
 	if err != nil {
 		t.Fatalf("unmarshalCheckpoint should tolerate garbage: %v", err)
 	}
@@ -155,7 +155,7 @@ func TestMarshalCheckpoint_BadEventErrors(t *testing.T) {
 	timers := map[string]pendingTimer{
 		"bad": {id: "bad", due: time.Unix(0, 1), event: make(chan int)},
 	}
-	if _, err := marshalCheckpoint([]byte(`{}`), timers); err == nil {
+	if _, err := marshalCheckpoint([]byte(`{}`), timers, 0); err == nil {
 		t.Fatal("want encode error for unmarshalable timer event, got nil")
 	}
 }
@@ -174,3 +174,18 @@ func TestTickInstant(t *testing.T) {
 		t.Fatalf("tickInstant: ok=%v got=%d", ok, got.UnixNano())
 	}
 }
+
+// TestCheckpoint_NextStepRoundTrips proves the envelope carries the ordinal after
+// the checkpoint, and that a checkpoint without one reports zero, the old form.
+func TestCheckpoint_NextStepRoundTrips(t *testing.T) {
+	b, err := marshalCheckpoint([]byte(`{"v":1}`), nil, 9)
+	if err != nil {
+		t.Fatalf("marshalCheckpoint: %v", err)
+	}
+	if _, _, next, err := unmarshalCheckpoint(b); err != nil || next != 9 {
+		t.Fatalf("unmarshalCheckpoint = (%d, %v), want (9, nil)", next, err)
+	}
+	if _, _, next, err := unmarshalCheckpoint([]byte(`{"snapshot":{"v":1}}`)); err != nil || next != 0 {
+		t.Fatalf("old envelope = (%d, %v), want (0, nil)", next, err)
+	}
+}

durable/runner.go

@@ -26,7 +26,7 @@ import (
 // (configuration, context, event payload, machine definition). The only input a
 // Runner must record to reproduce a run is therefore the driving event and any
 // payload it carried: each Fire appends a Record{Step, Event} where Step is the
-// produced Trace ordinal and Event is the kernel's structured Trace.EventPayload,
+// step's ordinal and Event is the kernel's structured Trace.EventPayload,
 // plus the payload's JSON form for a state.WithEventData Fire. Periodically — governed by
 // the checkpoint policy (WithCheckpointEvery) — the Runner also persists a full
 // marshaled Snapshot and compacts the journal tail through that step, so recovery
@@ -252,8 +252,8 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt
 	buf := make([]state.JournalEntry, 0)
 	recClock := newRecordingClock(r.cfg.clock, &buf)
 	// Force unbounded full-trace history: the runner journals each step's Trace
-	// (notably the EventPayload) and derives both the step ordinal and the checkpoint
-	// snapshots from the instance's retained Trace history, so a recover can replay it
+	// (notably the EventPayload) and derives the checkpoint snapshots from the
+	// instance's retained Trace history, so a recover can replay it
 	// byte-identically. WithUnboundedHistory elevates to full trace and retains every
 	// step; it is prepended so a caller's opts cannot downgrade it.
 	castOpts := append([]state.CastOption[S]{state.WithClock[S](recClock), state.WithUnboundedHistory[S]()}, opts...)
@@ -269,7 +269,7 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt
 	}
 	// Wrap the baseline snapshot in a checkpoint envelope (no timers armed yet) so
 	// every checkpoint a recover loads carries the same durable envelope shape.
-	baseEnv, err := marshalCheckpoint(snap, nil)
+	baseEnv, err := marshalCheckpoint(snap, nil, 0)
 	if err != nil {
 		return nil, fmt.Errorf("durable: marshaling start checkpoint envelope for %q: %w", id, err)
 	}
@@ -354,7 +354,7 @@ func (r *Runner[S, E, C]) Fire(ctx context.Context, id InstanceID, event E, opts
 
 // Fire drives one event through the Handle's live instance and records the step:
 // it Fires the kernel, appends a Record carrying the driving event at the
-// produced Trace ordinal (write-ahead, before returning), and — when the
+// step's ordinal (write-ahead, before returning), and — when the
 // checkpoint policy is due — persists a full Snapshot and compacts the tail. A
 // kernel transition error is recorded as a no-op (no step was produced) and
 // returned to the caller.
@@ -487,7 +487,7 @@ func (h *Handle[S, E, C]) appendStep(ctx context.Context, step int, rec *Record)
 		// Wrap the kernel snapshot with the absolute deadlines of the timers armed at
 		// this checkpoint, so a recovery whose compacted tail no longer carries their
 		// arming ScheduleAfter can re-arm them at their recorded instants.
-		env, err := marshalCheckpoint(snap, h.timers)
+		env, err := marshalCheckpoint(snap, h.timers, rec.followingStep())
 		if err != nil {
 			return fmt.Errorf("durable: marshaling checkpoint envelope at step %d for %q: %w", step, h.id, err)
 		}
@@ -537,10 +537,10 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S
 		return nil, fmt.Errorf("durable: instance %q has no checkpoint to restore", id)
 	}
 
-	// Unwrap the durable checkpoint envelope: the kernel snapshot bytes plus the
-	// absolute deadlines of the timers that were armed at the checkpoint. A bare
+	// Unwrap the durable checkpoint envelope: the kernel snapshot bytes, the absolute
+	// deadlines of the timers armed at the checkpoint, and the next step ordinal. A bare
 	// (pre-envelope) checkpoint yields the snapshot with no persisted timers.
-	kernelSnap, persistedTimers, err := unmarshalCheckpoint(snapBytes)
+	kernelSnap, persistedTimers, checkpointNext, err := unmarshalCheckpoint(snapBytes)
 	if err != nil {
 		return nil, fmt.Errorf("durable: unmarshaling checkpoint envelope for %q: %w", id, err)
 	}
@@ -716,15 +716,16 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S
 	// scheduler-tick barrier, which consumed one ordinal per timer it fired plus the
 	// barrier's own. The next ordinal is therefore the last tail Record's Step plus
 	// its own span. When the tail is empty (a fresh checkpoint compacted every
-	// Record), no actor or tick step survives in it, so the restored Trace count —
-	// one per pre-checkpoint step — gives the next ordinal directly.
-	next := len(inst.History())
+	// Record), the checkpoint carries the next ordinal. A Start baseline carries none and
+	// has no Traces, so its count of zero is right. A checkpoint from before this field
+	// also falls back to the Trace count, which can be off (see
+	// checkpointEnvelope.NextStep).
+	next := checkpointNext
+	if next == 0 {
+		next = len(inst.History())
+	}
 	if n := len(tail); n > 0 {
-		last := &tail[n-1]
-		next = last.Step + 1
-		if last.Tick {
-			next = last.Step + last.TickSteps + 1
-		}
+		next = tail[n-1].followingStep()
 	}
 	return &Handle[S, E, C]{
 		runner:   r,