diff --git a/durable/clock_test.go b/durable/clock_test.go index 85d24db..7e59a94 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 c2a33a3..66aff45 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 931af3c..ffbbcbc 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 df1d4db..07f2abd 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 3834a21..9d988de 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) + } +}