From 8aa8edfed86c08b65e1b61505cf0355085b08152 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 5 Nov 2026 08:56:14 -0700 Subject: [PATCH] fix(durable): carry the next step through a checkpoint Recovery derived the next step from the restored Trace count when a checkpoint had compacted the tail. A tick barrier consumes an ordinal but leaves no Trace, so the count fell short and the next Fire reused a recorded step. The checkpoint envelope now records the step that follows it. A checkpoint written before this change keeps the old fallback. Relates to #214. Signed-off-by: Ray Karnes --- durable/CHANGELOG.md | 10 +- durable/checkpoint.go | 32 ++++-- durable/checkpoint_internal_test.go | 27 ++++- durable/runner.go | 35 +++--- durable/runner_test.go | 168 ++++++++++++++++++++++++++++ durable/store.go | 14 ++- durable/timetravel.go | 2 +- state/options.go | 4 +- state/snapshot.go | 4 +- 9 files changed, 252 insertions(+), 44 deletions(-) diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index a4e26561..543dc021 100644 --- a/durable/CHANGELOG.md +++ b/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 diff --git a/durable/checkpoint.go b/durable/checkpoint.go index a3a12b6a..c65eb830 100644 --- a/durable/checkpoint.go +++ b/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 diff --git a/durable/checkpoint_internal_test.go b/durable/checkpoint_internal_test.go index 45f61931..5ae66b4f 100644 --- a/durable/checkpoint_internal_test.go +++ b/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) + } +} diff --git a/durable/runner.go b/durable/runner.go index 9e1aa048..e77e5a11 100644 --- a/durable/runner.go +++ b/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, diff --git a/durable/runner_test.go b/durable/runner_test.go index a139451d..2d33819a 100644 --- a/durable/runner_test.go +++ b/durable/runner_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "testing" + "time" "github.com/stablekernel/crucible/durable" "github.com/stablekernel/crucible/state" @@ -476,6 +477,23 @@ func TestRunner_Start_RejectsDuplicate(t *testing.T) { } } +// pollingMachine re-arms a one-second timer every time it fires, so a host timer +// loop keeps producing steps, while `poke` bumps Count on the same state. +func pollingMachine() *state.Machine[string, string, *runCtx] { + return state.ForgeFor[*runCtx]("polling"). + Action("bump", func(c state.ActionCtx[*runCtx]) (state.Effect, error) { + c.Entity.Count++ + return nil, nil + }). + State("idle"). + State("polling"). + Initial("idle"). + Transition("idle").On("start").GoTo("polling"). + Transition("polling").On("poke").GoTo("polling").Do("bump"). + Transition("polling").After(time.Second).On("tick").GoTo("polling").Reenter(). + Quench() +} + // quantityMachine folds a quantity carried as event data into Count. The quantity // is a float64 because event data crosses the journal as JSON (doc.go, Caveats). func quantityMachine() *state.Machine[string, string, *runCtx] { @@ -557,3 +575,153 @@ func TestRecover_FireWithEventData_MatchesLive(t *testing.T) { t.Fatalf("recovered instance differs from live:\n live: %s\n got: %s", live, got) } } + +// TestRecover_StepAfterTickCheckpoint_IsPersisted proves the next step ordinal +// survives a checkpoint that compacts tick barriers: a Fire after recovery must +// not reuse a recorded step, which the Store would drop or refuse. +func TestRecover_StepAfterTickCheckpoint_IsPersisted(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) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + // Four timer-firing ticks put barriers on steps 1, 3, 5 and 7; the checkpoint + // at step 7 compacts the whole tail. + for range 4 { + 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) != 0 { + t.Fatalf("tail holds %d records, want the checkpoint to have compacted them", len(tail)) + } + + resumed, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = resumed.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire(poke): %v", err) + } + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover(again): %v", err) + } + if live, got := handleSnapshot(t, resumed), handleSnapshot(t, recovered); got != live { + t.Fatalf("the acknowledged poke was not persisted:\n live: %s\n got: %s", live, got) + } +} + +// lastStep returns the Step of the newest Record in a history-keeping store. +func lastStep(t *testing.T, store *durable.MemStore, id durable.InstanceID) int { + t.Helper() + _, records, err := store.History(context.Background(), id) + if err != nil { + t.Fatalf("History: %v", err) + } + if len(records) == 0 { + t.Fatal("no records") + } + return records[len(records)-1].Step +} + +// TestRecover_NextStep_IsTheLiveOrdinal proves the ordinal after a compacted tail is +// exactly the one a live run would take, not merely one past the journal. +func TestRecover_NextStep_IsTheLiveOrdinal(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := durable.NewMemStore(durable.WithHistory()) + 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) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + for range 4 { // barriers on steps 1, 3, 5 and 7, each spanning two ordinals + clk.Advance(time.Second) + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + } + resumed, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = resumed.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire(poke): %v", err) + } + if got := lastStep(t, store, id); got != 9 { + t.Fatalf("poke recorded at step %d, want 9", got) + } +} + +// oldEnvelopeStore serves checkpoints as a version before nextStep wrote them. +type oldEnvelopeStore struct{ *durable.MemStore } + +func (s oldEnvelopeStore) Load(ctx context.Context, id durable.InstanceID) ([]byte, []durable.Record, error) { + snap, tail, err := s.MemStore.Load(ctx, id) + if err != nil { + return nil, nil, err + } + var env map[string]json.RawMessage + if err = json.Unmarshal(snap, &env); err != nil { + return nil, nil, err + } + delete(env, "nextStep") + snap, err = json.Marshal(env) + return snap, tail, err +} + +// TestRecover_CheckpointWithoutNextStep_UsesTheTraceCount proves a checkpoint from before +// the field still recovers, from the restored Trace count. +func TestRecover_CheckpointWithoutNextStep_UsesTheTraceCount(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + mem := durable.NewMemStore(durable.WithHistory()) + store := oldEnvelopeStore{mem} + id := durable.InstanceID("poll-1") + opts := []durable.Option[string, string, *runCtx]{durable.WithCheckpointEvery[string, string, *runCtx](3)} + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + for _, ev := range []string{"start", "poke", "poke"} { // the checkpoint compacts all three + if _, err = h.Fire(ctx, ev); err != nil { + t.Fatalf("Fire(%s): %v", ev, err) + } + } + resumed, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = resumed.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire(poke): %v", err) + } + if got := lastStep(t, mem, id); got != 3 { + t.Fatalf("poke recorded at step %d, want 3", got) + } +} diff --git a/durable/store.go b/durable/store.go index 1ae61605..31b12a79 100644 --- a/durable/store.go +++ b/durable/store.go @@ -24,8 +24,9 @@ type InstanceID string // after an upgrade reads as a conflict. It must also be copied in cloneRecord, or the // digest ignores it. The same applies to state.JournalEntry and state.EffectEnvelope. type Record struct { - // Step is the Fire ordinal this Record was produced at, indexing the - // instance's recorded Traces. It is the per-instance idempotency key: a + // Step is the ordinal this Record was produced at. Ordinals rise with each Record, + // and a tick barrier consumes one per timer it fired plus its own, so they don't + // index the instance's Traces. It is the per-instance idempotency key: a // Store treats a second Append of the same Record at the same Step as // already-applied. Step int @@ -84,6 +85,15 @@ type Record struct { TickSteps int } +// followingStep returns the step ordinal after r. A tick barrier consumes one +// ordinal per timer it fired plus its own; every other Record consumes one. +func (r *Record) followingStep() int { + if r.Tick { + return r.Step + r.TickSteps + 1 + } + return r.Step + 1 +} + // Store is the durable-execution persistence seam. It records an instance's // per-step Records and periodic Snapshot checkpoints, and reconstructs the // instance from the latest checkpoint plus the tail of Records appended after diff --git a/durable/timetravel.go b/durable/timetravel.go index 00cf8932..4d5be559 100644 --- a/durable/timetravel.go +++ b/durable/timetravel.go @@ -202,7 +202,7 @@ func restoreBaseline[S comparable, E comparable, C any]( baseline []byte, repClock state.Clock, ) (*state.Instance[S, E, C], error) { - kernelSnap, _, err := unmarshalCheckpoint(baseline) + kernelSnap, _, _, err := unmarshalCheckpoint(baseline) if err != nil { return nil, fmt.Errorf("unmarshaling baseline envelope: %w", err) } diff --git a/state/options.go b/state/options.go index c6a5a074..fdd419ae 100644 --- a/state/options.go +++ b/state/options.go @@ -378,8 +378,8 @@ func WithRestoreFullTrace[S comparable]() RestoreOption[S] { // WithRestoreUnboundedHistory restores an instance in full trace mode with // unbounded history retention, mirroring WithUnboundedHistory at Cast. The durable -// runner uses it on recover: it reconstructs the step ordinal and checkpoint -// snapshots from the retained Trace history, so post-recovery Fires must keep +// runner uses it on recover: it reconstructs checkpoint snapshots from the retained +// Trace history, so post-recovery Fires must keep // retaining every trace exactly as a freshly started durable instance does. func WithRestoreUnboundedHistory[S comparable]() RestoreOption[S] { return func(cfg *restoreConfig[S]) { diff --git a/state/snapshot.go b/state/snapshot.go index 4bbf7119..4bcbf732 100644 --- a/state/snapshot.go +++ b/state/snapshot.go @@ -227,8 +227,8 @@ const ( // Clock.Now() reads, and host randomness — each correlated by a stable id reused // from the effect that armed it (invocationID / actorInvocationID / scheduleID). type JournalEntry struct { - // Step is the Fire ordinal the result resolved at, indexing the instance's - // recorded Traces, so replay applies the recorded value at the right step. + // Step is the ordinal the result resolved at, so replay applies the recorded value + // at the right step. Step int `json:"step"` // Kind classifies which nondeterministic source produced the result. Kind JournalKind `json:"kind"`