From 6b6d3c35adc8207915ea30fb53b1cd3f5d7adaa6 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 22 Oct 2026 15:07:32 -0600 Subject: [PATCH] test(durable): add failing tests for recovery, timer and effect defects Each test asserts what any correct fix must produce, so it fails on main today and passes once its defect is fixed. Contract choices the fixes depend on are listed in the PR, not tested. - runner: two Handles on one instance split the journal; Fire during the timer loop loses steps; event data is dropped on recovery; the step after a tick-barrier checkpoint is dropped; a failed Append leaves the live instance ahead of the journal - clock: Start doesn't arm the initial state's timer; ticks before a timer is due append records - service: a timer armed by a service result is lost on replay; a result from a canceled turn is committed on recovery - timer_checkpoint: a checkpointed timer that fires in the tail is missed - effect: effects from Start, timer and service transitions never reach the handler; a failed dispatch is compacted away by a checkpoint Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01U2eeCbrr5qRKgUAioqVKmc Signed-off-by: Ray Karnes --- durable/clock_test.go | 65 ++++++++ durable/effect_test.go | 125 ++++++++++++++- durable/runner_test.go | 260 +++++++++++++++++++++++++++++++ durable/service_test.go | 97 ++++++++++++ durable/timer_checkpoint_test.go | 39 +++++ 5 files changed, 582 insertions(+), 4 deletions(-) diff --git a/durable/clock_test.go b/durable/clock_test.go index 85d24db4..7e59a944 100644 --- a/durable/clock_test.go +++ b/durable/clock_test.go @@ -414,3 +414,68 @@ func recoverTimerBytes(t *testing.T, ctx context.Context, m *state.Machine[strin } return string(b) } + +// TestClock_StartInTimedState_ArmsTimer proves Start arms the `after` timer of the +// state an instance starts in, as a plain kernel Cast does, and that recovery +// reaches the same state. +func TestClock_StartInTimedState_ArmsTimer(t *testing.T) { + ctx := context.Background() + m := timerMachine() + store := durable.NewMemStore() + id := durable.InstanceID("timer-1") + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *timerCtx]{durable.WithRunnerClock[string, string, *timerCtx](clk)} + runner := durable.NewRunner(m, store, opts...) + h, err := runner.Start(ctx, id, &timerCtx{}, state.WithInitialState("arming")) + if err != nil { + t.Fatalf("Start: %v", err) + } + clk.Advance(6 * time.Second) + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + if got := h.Instance().Current(); got != "done" { + t.Fatalf("state %q 6s after starting in arming, want done", got) + } + + 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) + } +} + +// TestClock_TickWithNothingDue_AppendsNoRecord proves a timer loop ticking while a +// timer is pending but not yet due does not grow the journal. +func TestClock_TickWithNothingDue_AppendsNoRecord(t *testing.T) { + ctx := context.Background() + store := durable.NewMemStore() + id := durable.InstanceID("timer-1") + clk := state.NewFakeClock(epoch) + runner := durable.NewRunner(timerMachine(), store, durable.WithRunnerClock[string, string, *timerCtx](clk)) + h, err := runner.Start(ctx, id, &timerCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "arm"); err != nil { + t.Fatalf("Fire(arm): %v", err) + } + _, before, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + for range 100 { + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + } + _, after, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(after) != len(before) { + t.Fatalf("100 ticks before the 5s timer was due grew the journal from %d to %d records", len(before), len(after)) + } +} diff --git a/durable/effect_test.go b/durable/effect_test.go index c2a33a3b..66aff45b 100644 --- a/durable/effect_test.go +++ b/durable/effect_test.go @@ -432,9 +432,10 @@ func TestRunner_EffectDispatch_NonDispatchStore_NoOp(t *testing.T) { } } -// timerMachine emits a kernel driver effect (a delayed `after` transition) on the -// same step a domain effect fires, so the dispatcher must route the driver effect -// to the Scheduler and only the domain effect to the handler. +// timerNotifyMachine emits a kernel driver effect (a delayed `after` transition) on +// the same step a domain effect fires, so the dispatcher must route the driver +// effect to the Scheduler and only the domain effect to the handler. The timer +// transition emits a second notify, which only a test that ticks observes. func timerNotifyMachine() *state.Machine[string, string, *runCtx] { return state.ForgeFor[*runCtx]("timer"). Action("notify", func(state.ActionCtx[*runCtx]) (state.Effect, error) { @@ -445,7 +446,7 @@ func timerNotifyMachine() *state.Machine[string, string, *runCtx] { State("done").Final(). Initial("idle"). Transition("idle").On("go").GoTo("waiting").Do("notify"). - Transition("waiting").After(time.Second).GoTo("done"). + Transition("waiting").After(time.Second).GoTo("done").Do("notify"). Quench() } @@ -506,3 +507,119 @@ func TestMemStore_DispatchSet_MarkAndQuery(t *testing.T) { t.Fatalf("unmarked id reported dispatched") } } + +// entryNotifyMachine emits a host effect on entering the state an instance starts +// in, so the effect originates in Start rather than Fire. +func entryNotifyMachine() *state.Machine[string, string, *runCtx] { + return state.ForgeFor[*runCtx]("entryNotify"). + Action("notify", func(state.ActionCtx[*runCtx]) (state.Effect, error) { + return notifyEffect{To: "ops@example.com"}, nil + }). + State("ready").OnEntry("notify"). + Initial("ready"). + Quench() +} + +// TestRunner_EffectDispatch_Start_AppliesEntryEffect proves a domain effect emitted +// on entering the initial state reaches the handler. +func TestRunner_EffectDispatch_Start_AppliesEntryEffect(t *testing.T) { + ctx := context.Background() + rec := newEffectRecorder() + r := durable.NewRunner(entryNotifyMachine(), durable.NewMemStore(), + durable.WithEffectHandler[string, string, *runCtx](rec.handler())) + if _, err := r.Start(ctx, "i1", &runCtx{}, state.WithInitialState("ready")); err != nil { + t.Fatalf("Start: %v", err) + } + if n := rec.totalApplies(); n != 1 { + t.Fatalf("handler applied %d effects, want 1 (the initial state's entry notify)", n) + } +} + +// TestRunner_EffectDispatch_TimerTransition_AppliesEffect proves a domain effect +// emitted by a timer-driven transition reaches the handler. +func TestRunner_EffectDispatch_TimerTransition_AppliesEffect(t *testing.T) { + ctx := context.Background() + rec := newEffectRecorder() + clk := state.NewFakeClock(epoch) + r := durable.NewRunner(timerNotifyMachine(), durable.NewMemStore(), + durable.WithEffectHandler[string, string, *runCtx](rec.handler()), + durable.WithRunnerClock[string, string, *runCtx](clk)) + h, err := r.Start(ctx, "i1", &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "go"); err != nil { + t.Fatalf("Fire go: %v", err) + } + clk.Advance(2 * time.Second) + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + if n := rec.totalApplies(); n != 2 { + t.Fatalf("handler applied %d effects, want 2 (the timer transition's notify)", n) + } +} + +// TestRunner_EffectDispatch_ServiceSettle_AppliesEffect proves a domain effect +// emitted by the transition a service result drives reaches the handler. +func TestRunner_EffectDispatch_ServiceSettle_AppliesEffect(t *testing.T) { + ctx := context.Background() + rec := newEffectRecorder() + var calls int64 + fn := counterService(&calls) + r := durable.NewRunner(fetchThenWaitMachine(fn), durable.NewMemStore(), + durable.WithEffectHandler[string, string, svcCtx](rec.handler()), + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn))) + h, err := r.Start(ctx, "i1", svcCtx{}, 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) + } + mustRunService(t, h, state.InvokeID("fetchThenWait", "loading", 0)) + if n := rec.totalApplies(); n != 1 { + t.Fatalf("handler applied %d effects, want 1 (the settle transition's notify)", n) + } +} + +// TestRunner_EffectDispatch_FailedThenCheckpointed_AppliedOnce proves an effect +// whose dispatch failed is still applied exactly once, live or on recovery, after +// a later checkpoint compacts its step. +func TestRunner_EffectDispatch_FailedThenCheckpointed_AppliedOnce(t *testing.T) { + ctx := context.Background() + st := durable.NewMemStore() + live := newEffectRecorder() + live.failKinds["demo.notify"] = errors.New("downstream unavailable") + opts := []durable.Option[string, string, *runCtx]{ + durable.WithEffectHandler[string, string, *runCtx](live.handler()), + durable.WithCheckpointEvery[string, string, *runCtx](2), + } + h, err := durable.NewRunner(effectMachine(), st, opts...).Start(ctx, "i1", &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "go"); err == nil { // step 0: its notify fails + t.Fatalf("Fire go: want the handler error") + } + delete(live.failKinds, "demo.notify") + if _, err = h.Fire(ctx, "bill"); err != nil { // step 1: checkpoint compacts step 0 + // A Handle that refuses Fires after a failed dispatch is resumed by Recover. + if h, err = durable.Recover(ctx, effectMachine(), st, "i1", opts...); err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = h.Fire(ctx, "bill"); err != nil { + t.Fatalf("Fire bill: %v", err) + } + } + + healthy := newEffectRecorder() + if _, err = durable.Recover(ctx, effectMachine(), st, "i1", + durable.WithEffectHandler[string, string, *runCtx](healthy.handler())); err != nil { + t.Fatalf("Recover: %v", err) + } + // go emits notify; bill emits charge and notify. Each applies exactly once. + if n := live.totalApplies() + healthy.totalApplies(); n != 3 { + t.Fatalf("applied %d effects across live and recovery, want 3", n) + } +} diff --git a/durable/runner_test.go b/durable/runner_test.go index 931af3c5..ffbbcbcd 100644 --- a/durable/runner_test.go +++ b/durable/runner_test.go @@ -4,7 +4,11 @@ import ( "context" "encoding/json" "errors" + "runtime" + "sync" + "sync/atomic" "testing" + "time" "github.com/stablekernel/crucible/durable" "github.com/stablekernel/crucible/state" @@ -142,6 +146,17 @@ func recoveredSnapshotBytes[C any](t *testing.T, m *state.Machine[string, string return b } +// handleSnapshot returns the marshaled snapshot of a durable Handle's instance, so +// a live Handle and a recovered one can be compared byte-for-byte. +func handleSnapshot[C any](t *testing.T, h *durable.Handle[string, string, C]) string { + t.Helper() + b, err := state.MarshalSnapshot(h.Instance().Snapshot()) + if err != nil { + t.Fatalf("marshal: %v", err) + } + return string(b) +} + // TestRecover_ByteIdenticalAcrossFixtures is the acceptance gate: a recovered // instance's marshaled snapshot is byte-identical to a never-durable live run // across linear, branching, context-mutating, and parallel machines. @@ -464,3 +479,248 @@ func TestRunner_Start_RejectsDuplicate(t *testing.T) { t.Fatalf("want ErrInstanceExists, got %v", err) } } + +// 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] { + return state.ForgeFor[*runCtx]("quantity"). + Reducer("addQuantity", func(in state.AssignCtx[*runCtx]) *runCtx { + c := *in.Entity + if q, ok := in.Event.(float64); ok { + c.Count += int(q) + } + return &c + }). + State("open"). + Initial("open"). + Transition("open").On("add").GoTo("open").Assign("addQuantity"). + Quench() +} + +// flakyAppendStore fails the first Append of one step, then delegates. +type flakyAppendStore struct { + durable.Store + failStep int + failed bool +} + +func (s *flakyAppendStore) Append(ctx context.Context, id durable.InstanceID, rec durable.Record, opts ...durable.AppendOption) (int64, error) { + if rec.Step == s.failStep && !s.failed { + s.failed = true + return 0, errors.New("store unavailable") + } + return s.Store.Append(ctx, id, rec, opts...) +} + +// TestRecover_TwoHandlesOnOneInstance_JournalMatchesAnAcknowledgedHistory proves +// two Handles on one instance (a stalled worker, and the worker that recovered the +// instance after it) cannot both have their Fires acknowledged while the journal +// records something else: recovery reproduces the state of any Handle whose Fires +// all returned nil. Takeover on Recover and optimistic concurrency on the step +// both satisfy it. +func TestRecover_TwoHandlesOnOneInstance_JournalMatchesAnAcknowledgedHistory(t *testing.T) { + ctx := context.Background() + m := branchingMachine() + store := durable.NewMemStore() + id := durable.InstanceID("split-1") + if _, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("start")); err != nil { + t.Fatalf("Start: %v", err) + } + stale, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover(stale): %v", err) + } + owner, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover(owner): %v", err) + } + _, ownerErr := owner.Fire(ctx, "goLeft") + _, staleErr := stale.Fire(ctx, "goRight") + if staleErr == nil { + _, staleErr = stale.Fire(ctx, "finish") + } + + recovered, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover: %v", err) + } + got := recovered.Instance().Current() + if ownerErr == nil && owner.Instance().Current() != got { + t.Fatalf("owner's Fire was acknowledged at %q, but recovery yields %q", owner.Instance().Current(), got) + } + if staleErr == nil && stale.Instance().Current() != got { + t.Fatalf("stale Handle's Fires were acknowledged at %q, but recovery yields %q", stale.Instance().Current(), got) + } +} + +// TestRunner_FireDuringTimerLoop_LosesNoAcknowledgedSteps proves the "Tick from its +// own timer loop" pattern that drainClock documents holds while requests Fire on +// the same Handle: the recovered instance matches the live one, and every +// acknowledged Fire survives. +func TestRunner_FireDuringTimerLoop_LosesNoAcknowledgedSteps(t *testing.T) { + if runtime.GOMAXPROCS(0) < 2 { + t.Skip("needs two CPUs to interleave Fire and Tick") + } + 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)} + runner := durable.NewRunner(m, store, opts...) + h, err := runner.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) + } + + const n = 200 + var acked atomic.Int64 + var wg sync.WaitGroup + wg.Add(2) + go func() { // the host's timer loop + defer wg.Done() + for range n { + clk.Advance(time.Second) + _, _ = h.Tick(ctx) + } + }() + go func() { // request traffic + defer wg.Done() + for range n { + if _, ferr := h.Fire(ctx, "poke"); ferr == nil { + acked.Add(1) + } + } + }() + wg.Wait() + + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if got, want := recovered.Instance().Entity().Count, int(acked.Load()); got != want { + t.Fatalf("recovered %d pokes, want %d acknowledged", got, want) + } + 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) + } +} + +// TestRecover_FireWithEventData_MatchesLive proves an instance driven with +// state.WithEventData recovers to the live state: the Handle either journals the +// data or refuses it, but never acknowledges data that recovery drops. +func TestRecover_FireWithEventData_MatchesLive(t *testing.T) { + ctx := context.Background() + m := quantityMachine() + store := durable.NewMemStore() + id := durable.InstanceID("qty-1") + h, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("open")) + if err != nil { + t.Fatalf("Start: %v", err) + } + _, _ = h.Fire(ctx, "add", state.WithEventData(3.0)) + + recovered, err := durable.Recover(ctx, m, store, id) + 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) + } +} + +// 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 as an idempotent retry. +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) + } + } + + 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) + } +} + +// TestRunner_FireAfterFailedAppend_MatchesJournal proves a Fire whose Append failed +// leaves no trace in the live instance: once a later Fire is acknowledged, the +// live instance and the journal agree. +func TestRunner_FireAfterFailedAppend_MatchesJournal(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := &flakyAppendStore{Store: durable.NewMemStore(), failStep: 1} + id := durable.InstanceID("poll-1") + h, err := durable.NewRunner(m, store).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) + } + if _, err = h.Fire(ctx, "poke"); err == nil { // step 1: the Append fails + t.Fatalf("Fire(poke): want the Append error") + } + if _, err = h.Fire(ctx, "poke"); err != nil { + return // refusing further Fires after a failed Append is also correct + } + + recovered, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if live, got := handleSnapshot(t, h), handleSnapshot(t, recovered); got != live { + t.Fatalf("live instance differs from the journal:\n live: %s\n got: %s", live, got) + } +} diff --git a/durable/service_test.go b/durable/service_test.go index df1d4db8..07f2abd6 100644 --- a/durable/service_test.go +++ b/durable/service_test.go @@ -6,6 +6,7 @@ import ( "fmt" "sync/atomic" "testing" + "time" "github.com/stablekernel/crucible/durable" "github.com/stablekernel/crucible/state" @@ -470,3 +471,99 @@ func TestService_Determinism(t *testing.T) { t.Fatalf("two live runs diverged:\n run1=%s\n run2=%s", s1, s2) } } + +// fetchThenWaitMachine routes its service result into a state that arms a timer, +// emitting a host effect on the way, so both originate in RunService. +func fetchThenWaitMachine(fn state.ServiceFn[svcCtx]) *state.Machine[string, string, svcCtx] { + return state.ForgeFor[svcCtx]("fetchThenWait"). + Service("fetch", fn). + Action("notify", func(state.ActionCtx[svcCtx]) (state.Effect, error) { + return notifyEffect{To: "ops@example.com"}, nil + }). + State("idle"). + State("loading").Invoke("fetch", state.WithInvokeOnDone("ok")). + State("waiting"). + State("done").Final(). + Initial("idle"). + Transition("idle").On("start").GoTo("loading"). + Transition("loading").On("ok").GoTo("waiting").Do("notify"). + Transition("waiting").After(5 * time.Second).On("elapsed").GoTo("done"). + Quench() +} + +// TestService_TimerArmedBySettle_ReplaysOnRecover proves a timer armed by a service +// result survives recovery, with an idle tick between the settle and the tick that +// fires it. +func TestService_TimerArmedBySettle_ReplaysOnRecover(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := fetchThenWaitMachine(fn) + store := durable.NewMemStore() + id := durable.InstanceID("fetch-1") + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, svcCtx]{ + durable.WithRunnerClock[string, string, svcCtx](clk), + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn)), + } + runner := durable.NewRunner(m, store, opts...) + h, err := runner.Start(ctx, id, svcCtx{}, 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) + } + mustRunService(t, h, state.InvokeID("fetchThenWait", "loading", 0)) + for _, d := range []time.Duration{2 * time.Second, 4 * time.Second} { // idle, then due + clk.Advance(d) + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + } + if got := h.Instance().Current(); got != "done" { + t.Fatalf("live state %q, want done", got) + } + + 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) + } +} + +// TestService_ResultFromCanceledTurn_NotCommittedOnRecover proves a service result +// whose settle is canceled with the turn's context, and so never reaches the live +// instance, is not committed by recovery either. +func TestService_ResultFromCanceledTurn_NotCommittedOnRecover(t *testing.T) { + ctx := context.Background() + turnCtx, cancel := context.WithCancel(ctx) + defer cancel() + fn := func(context.Context, state.ServiceCtx[svcCtx]) (any, error) { + cancel() // the turn's deadline passes while the call is in flight + return "late", nil + } + m := fetchMachine(fn) + store := durable.NewMemStore() + id := durable.InstanceID("fetch-1") + opts := []durable.Option[string, string, svcCtx]{durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn))} + runner := durable.NewRunner(m, store, opts...) + h, err := runner.Start(ctx, id, svcCtx{}, 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) + } + _, _, _ = h.RunService(turnCtx, state.InvokeID("fetch", "loading", 0)) + + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if live, got := h.Instance().Current(), recovered.Instance().Current(); got != live { + t.Fatalf("recovered %q, live %q: the canceled turn's result was committed on recovery", got, live) + } +} diff --git a/durable/timer_checkpoint_test.go b/durable/timer_checkpoint_test.go index 3834a216..9d988de3 100644 --- a/durable/timer_checkpoint_test.go +++ b/durable/timer_checkpoint_test.go @@ -381,3 +381,42 @@ func TestTimerCheckpoint_RecoveryDeterministic(t *testing.T) { t.Fatalf("recovery across checkpoint nondeterministic\n first: %s\n second: %s", first, second) } } + +// TestTimerCheckpoint_FiredAfterCheckpoint_ByteIdentical proves a timer persisted by +// a checkpoint and fired later in the tail, after an idle tick, replays at its +// recorded instant instead of being missed. +func TestTimerCheckpoint_FiredAfterCheckpoint_ByteIdentical(t *testing.T) { + ctx := context.Background() + m := pendingTimerMachine() + store := durable.NewMemStore() + id := durable.InstanceID("timer-1") + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *timerCtx]{ + durable.WithRunnerClock[string, string, *timerCtx](clk), + durable.WithCheckpointEvery[string, string, *timerCtx](3), + } + runner := durable.NewRunner(m, store, opts...) + h, err := runner.Start(ctx, id, &timerCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + for _, ev := range []string{"arm", "poke", "poke"} { // checkpoint lands with the timer pending + if _, err = h.Fire(ctx, ev); err != nil { + t.Fatalf("Fire(%s): %v", ev, err) + } + } + for _, d := range []time.Duration{5 * time.Second, 6 * time.Second} { // idle, then due + clk.Advance(d) + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + } + + 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) + } +}