From faeac084e118ff503d6d468b62b1a4af90f5d947 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 5 Nov 2026 09:22:29 -0700 Subject: [PATCH] fix(durable): keep timers across recovery and skip idle ticks Start now arms the initial state's timers and carries them in the baseline. Recovery re-arms checkpointed timers before the tail replays, so one fired in the tail isn't missed. RunService records the clock read that arms a timer, and replay arms it. A tick that fires nothing records nothing, so a timer loop no longer grows the journal. Relates to #214. Signed-off-by: Ray Karnes --- durable/CHANGELOG.md | 15 + durable/checkpoint.go | 29 +- durable/checkpoint_internal_test.go | 38 +- durable/clock.go | 121 +++++- durable/clock_internal_test.go | 23 ++ durable/clock_test.go | 94 +++++ durable/doc.go | 4 +- durable/runner.go | 205 ++++++---- durable/service.go | 3 +- durable/service_test.go | 562 ++++++++++++++++++++++++++++ durable/store.go | 2 +- durable/timer_checkpoint_test.go | 539 ++++++++++++++++++++++++++ durable/timetravel.go | 53 ++- 13 files changed, 1545 insertions(+), 143 deletions(-) diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index 9c3ce2b1..89e0bbf6 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -91,6 +91,21 @@ 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. +- **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 + unarmed. Recovery re-arms checkpointed timers before the tail replays, so one fired in + the tail isn't missed, and a timer armed by a service result now replays. `StateAt` + re-arms baseline timers too, so its result changes when one fires in the window, and it + fails if a timer's event won't decode with the options passed. +- **A tick that fires nothing records nothing.** It used to append a Record every time, + so a timer loop grew the journal while a timer was pending. An idle `Tick` doesn't touch + the Store, so it can't report `ErrFenced` or a Store failure: a stale owner finds out on + its next real step. Replay still reads older journals, whose idle Records and misplaced + settle clock reads it aligns on. Two kinds of older journal can't be repaired, and can + recover differently or fail `Recover`: an instance that continued after an older + recovery missed a fired timer, and a checkpoint taken after a tick that followed a settle + or whose clock moved mid-tick, which can hold a wrong timer table. - **Effects reach the handler from `Start`, `Tick` and `RunService`.** They were only dispatched from `Fire`. `Start` applies the initial entry's effects after encoding the baseline and before writing it, and only if the entry succeeded. Those effects are diff --git a/durable/checkpoint.go b/durable/checkpoint.go index c65eb830..bec5eab9 100644 --- a/durable/checkpoint.go +++ b/durable/checkpoint.go @@ -23,10 +23,10 @@ import ( // The declared `after` Delay re-arms a fresh timer at now+Delay; a timer pending at // a checkpoint has already burned part of its delay, so the durable layer persists // each armed timer's absolute deadline alongside the checkpoint and, on recovery, -// re-arms the remainder (deadline − resumed now) rather than the full declared -// Delay. The deadline is itself a recorded clock read, so the re-armed timer fires -// at the same recorded instant regardless of the wall clock at recovery time, and -// the recovered run is byte-identical to one that never crashed. +// re-arms it at that deadline rather than for the full declared Delay. The deadline +// is itself a recorded clock read, so the re-armed timer fires at the same recorded +// instant regardless of the wall clock at recovery time, and the recovered run is +// byte-identical to one that never crashed. // pendingTimer is one armed delayed (`after`) timer the durable Handle tracks // alongside the kernel scheduler, so its absolute deadline can be persisted at a @@ -40,7 +40,7 @@ type pendingTimer struct { // scheduler's pending set key identically. id string // due is the timer's absolute fire deadline (the recorded arming clock read plus - // the declared Delay), the value persisted so recovery re-arms the remainder. + // the declared Delay), the value persisted so recovery re-arms it at that deadline. due time.Time // event is the delayed event the timer re-fires through Fire when it elapses, // type-erased for the abstract ScheduleAfter surface and re-typed by the @@ -158,27 +158,20 @@ func mirrorArmed(armed map[string]pendingTimer, effects []state.Effect, now time } } -// reArmEffects builds the ScheduleAfter effects that re-arm persisted timers at -// their recorded absolute deadlines, relative to now: each timer's remaining delay -// is its deadline minus now, so absorbing the effects through the live scheduler -// reproduces the same absolute deadline the live run held. A deadline already at or -// before now yields a non-positive delay, which the scheduler arms as immediately -// due, so an elapsed-while-down timer fires on the next Tick. Timers whose id is -// already armed (re-established by tail replay) are skipped, so a timer the tail -// still carries is not double-armed. Returns the effects and the durable-side -// pendingTimer table to seed the Handle, both in scheduleID order. +// reArmEffects builds the ScheduleAfter effects that re-arm persisted timers at their +// recorded absolute deadlines, relative to now. now can be any fixed instant: the caller +// holds the scheduler's clock on it while absorbing the effects, so each timer's deadline +// comes out exact, and an elapsed one fires on the next Tick. Recovery calls it before +// the tail replays, because the tail can fire these timers. Returns the effects and the +// durable-side pending-timer table. func reArmEffects[E comparable]( codec EventCodec[E], wires []pendingTimerWire, - already map[string]pendingTimer, now time.Time, ) ([]state.Effect, map[string]pendingTimer, error) { var effects []state.Effect seeded := map[string]pendingTimer{} for _, w := range wires { - if _, armed := already[w.ID]; armed { - continue // tail replay already re-armed this timer; do not double-arm. - } ev, err := decodeTimerEvent(codec, w.Event) if err != nil { return nil, nil, fmt.Errorf("durable: decoding pending timer %q event: %w", w.ID, err) diff --git a/durable/checkpoint_internal_test.go b/durable/checkpoint_internal_test.go index 5ae66b4f..45ea9cae 100644 --- a/durable/checkpoint_internal_test.go +++ b/durable/checkpoint_internal_test.go @@ -90,22 +90,19 @@ func TestUnmarshalCheckpoint_Garbage(t *testing.T) { } } -// TestReArmEffects_SkipsAlreadyArmed confirms a persisted timer already armed by -// tail replay is not re-armed (no double-arm), while an un-armed one is re-armed at -// its remaining delay. -func TestReArmEffects_SkipsAlreadyArmed(t *testing.T) { +// TestReArmEffects_RemainingDelay confirms a persisted timer is re-armed at its +// remaining delay. +func TestReArmEffects_RemainingDelay(t *testing.T) { now := time.Unix(0, 1_000_000_000).UTC() wires := []pendingTimerWire{ - {ID: "armed", DueUnixNano: now.Add(5 * time.Second).UnixNano(), Event: json.RawMessage(`"x"`)}, {ID: "compacted", DueUnixNano: now.Add(8 * time.Second).UnixNano(), Event: json.RawMessage(`"y"`)}, } - already := map[string]pendingTimer{"armed": {id: "armed"}} - effects, seeded, err := reArmEffects[string](jsonEventCodec[string]{}, wires, already, now) + effects, seeded, err := reArmEffects[string](jsonEventCodec[string]{}, wires, now) if err != nil { t.Fatalf("reArmEffects: %v", err) } if len(effects) != 1 || len(seeded) != 1 { - t.Fatalf("want exactly the compacted timer re-armed, got effects=%d seeded=%d", len(effects), len(seeded)) + t.Fatalf("want the timer re-armed, got effects=%d seeded=%d", len(effects), len(seeded)) } sa, ok := effects[0].(state.ScheduleAfter) if !ok || sa.ID != "compacted" { @@ -124,7 +121,7 @@ func TestReArmEffects_ElapsedDeadlineImmediate(t *testing.T) { wires := []pendingTimerWire{ {ID: "past", DueUnixNano: now.Add(-3 * time.Second).UnixNano(), Event: json.RawMessage(`"z"`)}, } - effects, _, err := reArmEffects[string](jsonEventCodec[string]{}, wires, map[string]pendingTimer{}, now) + effects, _, err := reArmEffects[string](jsonEventCodec[string]{}, wires, now) if err != nil { t.Fatalf("reArmEffects: %v", err) } @@ -142,7 +139,7 @@ func TestReArmEffects_BadEventErrors(t *testing.T) { wires := []pendingTimerWire{ {ID: "bad", DueUnixNano: 1, Event: json.RawMessage(`{not valid`)}, } - _, _, err := reArmEffects[string](jsonEventCodec[string]{}, wires, map[string]pendingTimer{}, time.Unix(0, 0)) + _, _, err := reArmEffects[string](jsonEventCodec[string]{}, wires, time.Unix(0, 0)) if err == nil { t.Fatal("want decode error for corrupt timer event, got nil") } @@ -160,18 +157,21 @@ func TestMarshalCheckpoint_BadEventErrors(t *testing.T) { } } -// TestTickInstant covers the no-clock-read branch (a tick that fired nothing -// records no read) alongside the happy path. -func TestTickInstant(t *testing.T) { - if _, ok := tickInstant(nil); ok { +// TestAbsorbRead confirms each result's instant comes from the tick's own reads, the last +// in the Record, past a read an older journal left at the front. +func TestAbsorbRead(t *testing.T) { + if _, ok := absorbRead(nil, 0, 1); ok { t.Fatal("empty entries should report no instant") } - entries := []state.JournalEntry{ - {Kind: state.JournalClockRead, ClockUnixNano: 42}, + read := func(n int64) state.JournalEntry { + return state.JournalEntry{Kind: state.JournalClockRead, ClockUnixNano: n} } - got, ok := tickInstant(entries) - if !ok || got.UnixNano() != 42 { - t.Fatalf("tickInstant: ok=%v got=%d", ok, got.UnixNano()) + entries := []state.JournalEntry{read(1), read(2), read(3), read(4), read(5), read(6)} + for i, want := range []int64{3, 5} { + got, ok := absorbRead(entries, i, 2) + if !ok || got.UnixNano() != want { + t.Fatalf("absorbRead(%d): ok=%v got=%d, want %d", i, ok, got.UnixNano(), want) + } } } diff --git a/durable/clock.go b/durable/clock.go index ad5081f4..b794cdac 100644 --- a/durable/clock.go +++ b/durable/clock.go @@ -31,9 +31,9 @@ import ( // step may absorb several effects. The recordingClock accumulates reads into a // shared buffer the Handle flushes into the Record.Entries of the step during // which they occurred (the step whose Fire/Tick produced them), in read order. -// Replay is order-driven, not step-keyed: the replayClock walks every recorded -// JournalClockRead across the loaded tail in recorded order and returns each -// reading once, so the scheduler re-derives identical deadlines. +// Replay walks every recorded JournalClockRead across the loaded tail in recorded +// order, aligning on each Record because a step's own readings are the last in it, so +// the scheduler re-derives identical deadlines. // recordingClock wraps a real state.Clock and records every Now() reading as a // JournalClockRead into a shared buffer, in read order. After delegates to the @@ -117,18 +117,20 @@ func (c *recordingClock) armReadingAt(i int) (time.Time, bool) { // readings and returns each once. // // When the cursor is exhausted — a Now() beyond the recorded readings, which -// happens on the live re-arm after replay — it falls through to a real clock so +// happens on live continuation after replay — it falls through to a real clock so // post-recovery timing continues against wall time. The fallthrough never affects // the replayed prefix, which is fully determined by the journal. type replayClock struct { mu sync.Mutex readings []int64 cursor int + held *time.Time // a fixed reading set by hold, nil otherwise + last time.Time // the last reading Now returned, see lastRead fallback state.Clock } // newReplayClock builds a replayClock over the recorded readings, falling through -// to fallback once the readings are exhausted (the live re-arm after replay). +// to fallback once the readings are exhausted (live continuation after replay). func newReplayClock(readings []int64, fallback state.Clock) *replayClock { return &replayClock{readings: readings, fallback: fallback} } @@ -138,12 +140,115 @@ func newReplayClock(readings []int64, fallback state.Clock) *replayClock { func (c *replayClock) Now() time.Time { c.mu.Lock() defer c.mu.Unlock() + if c.held != nil { + return *c.held + } if c.cursor < len(c.readings) { - t := time.Unix(0, c.readings[c.cursor]).UTC() + c.last = time.Unix(0, c.readings[c.cursor]).UTC() c.cursor++ - return t + return c.last + } + c.last = c.fallback.Now() + return c.last +} + +// skipTo moves the cursor forward to reading n and never back. Replay aligns on each +// Record this way, because a step's own readings are the last ones in its Record: an +// older journal left a settle's reading at the front of the next Record, which replay +// either consumed at the settle or must pass over. +func (c *replayClock) skipTo(n int) { + c.mu.Lock() + defer c.mu.Unlock() + if c.cursor < n { + c.cursor = n + } +} + +// lastRead returns the instant the last Now returned outside a hold: the one a +// Scheduler.Absorb just armed its timers at. +func (c *replayClock) lastRead() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.last +} + +// readingAt returns reading i without moving the cursor. +func (c *replayClock) readingAt(i int) (time.Time, bool) { + c.mu.Lock() + defer c.mu.Unlock() + if i < 0 || i >= len(c.readings) { + return time.Time{}, false + } + return time.Unix(0, c.readings[i]).UTC(), true +} + +// settleClock aligns a replayed settle's Absorb. A settle's own reading ends its Record. +// An older journal's settle Record holds none: its reading opens the next Record, if the +// process that took the settle lived to take the next step. If not, that Record holds +// only its own readings, so a settle must not consume them. It peeks instead, and each +// Record still aligns from its end. +type settleClock struct { + clock *replayClock + older int // read-less settles since the last Record that held a reading +} + +// next records that replay reached rec. +func (s *settleClock) next(rec *Record) { + if clockReadCount(rec) > 0 { + s.older = 0 + } +} + +// absorb runs absorb for a settle's timers at the reading they were armed at, and returns +// that instant. lost is true when the journal holds no such reading, so the fallback +// clock supplied it, and the caller should carry that read into the next step's Record. +func (s *settleClock) absorb(rec *Record, end int, absorb func()) (at time.Time, lost bool) { + if clockReadCount(rec) > 0 { + s.clock.skipTo(end - 1) + absorb() + return s.clock.lastRead(), false } - return c.fallback.Now() + if t, ok := s.clock.readingAt(end + s.older); ok { + s.older++ + s.clock.hold(t) + absorb() + s.clock.release() + return t, false + } + t := s.clock.fallback.Now() + s.clock.hold(t) + absorb() + s.clock.release() + return t, true +} + +// tickReads returns how many clock readings a tick that fired n timers takes: one to pick +// each timer, one to absorb its effects, and a last read that finds nothing due. +func tickReads(n int) int { return 2*n + 1 } + +// clockReadCount returns how many clock readings rec holds. +func clockReadCount(rec *Record) int { + n := 0 + for _, e := range rec.Entries { + if e.Kind == state.JournalClockRead { + n++ + } + } + return n +} + +// hold makes Now return t without consuming a reading, until release. +func (c *replayClock) hold(t time.Time) { + c.mu.Lock() + defer c.mu.Unlock() + c.held = &t +} + +// release ends a hold. +func (c *replayClock) release() { + c.mu.Lock() + defer c.mu.Unlock() + c.held = nil } // After delegates to the fallback clock; replay drives elapses through Now + Tick diff --git a/durable/clock_internal_test.go b/durable/clock_internal_test.go index a0a01241..827c27c3 100644 --- a/durable/clock_internal_test.go +++ b/durable/clock_internal_test.go @@ -167,3 +167,26 @@ func TestHasTimerEffect(t *testing.T) { t.Fatal("no effects must not count as a timer effect") } } + +// TestSettleClock_ReadLessSettlesPeekWithoutConsuming confirms read-less settles take the +// readings after their Record in order and leave the cursor, so the next Record can still +// read its own, and that a settle past the last reading reports it lost. +func TestSettleClock_ReadLessSettlesPeekWithoutConsuming(t *testing.T) { + at := func(n int64) time.Time { return time.Unix(0, n).UTC() } + c := newReplayClock([]int64{10, 20}, state.NewFakeClock(clockEpoch)) + s := settleClock{clock: c} + rec := &Record{} + for _, want := range []int64{10, 20} { + got, lost := s.absorb(rec, 0, func() { c.Now() }) + if lost || !got.Equal(at(want)) { + t.Fatalf("absorb: got %v lost %v, want %d", got, lost, want) + } + } + got, lost := s.absorb(rec, 0, func() { c.Now() }) + if !lost || !got.Equal(clockEpoch) { + t.Fatalf("a settle past the last reading: got %v lost %v, want the fallback clock", got, lost) + } + if c.cursor != 0 { + t.Fatalf("cursor at %d, want 0: a settle must leave the next Record's readings", c.cursor) + } +} diff --git a/durable/clock_test.go b/durable/clock_test.go index 85d24db4..ab339ac5 100644 --- a/durable/clock_test.go +++ b/durable/clock_test.go @@ -414,3 +414,97 @@ 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 host absorbing InitialEffects 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) + } + + // A clock still at the start proves the timer replays at its recorded instant. + recovered, err := durable.Recover(ctx, m, store, id, durable.WithRunnerClock[string, string, *timerCtx](state.NewFakeClock(epoch))) + 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 { + clk.Advance(10 * time.Millisecond) + 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)) + } +} + +// TestClock_TickWithNoTimer_AppendsNoRecord proves a timer loop ticking an instance with no +// timer armed does not grow the journal. +func TestClock_TickWithNoTimer_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) + } + for range 100 { + clk.Advance(10 * time.Millisecond) + 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("100 ticks with no timer armed appended %d records", len(tail)) + } +} diff --git a/durable/doc.go b/durable/doc.go index 99ead028..d5b21ff8 100644 --- a/durable/doc.go +++ b/durable/doc.go @@ -92,8 +92,8 @@ // // Clock (WithRunnerClock): a running instance reads time only through its host // scheduler, which arms and ticks delayed `after` transitions. The Runner wraps -// the real clock in a recording clock on the live path so every Now() reading the -// scheduler consumes is journaled as a JournalClockRead in the step's Record. +// the real clock in a recording clock on the live path so the Now() readings of a step +// are journaled as JournalClockRead entries in its Record. An idle Tick records nothing. // On recovery a replay clock returns those readings in order, making timer-driven // transitions wall-clock-independent: the same timers fire at the same recorded // instants regardless of the wall clock at recovery time. A timer also survives diff --git a/durable/runner.go b/durable/runner.go index 826a7f53..42f8ff96 100644 --- a/durable/runner.go +++ b/durable/runner.go @@ -111,6 +111,7 @@ type Handle[S comparable, E comparable, C any] struct { sched *state.Scheduler[S, E, C] recClock *recordingClock // owns the clock-read accumulator and its mutex + carried []state.JournalEntry // a read recovery took for an older settle, for the next step's Record timers map[string]pendingTimer // armed `after` timers with absolute deadlines, persisted per checkpoint svc *state.ServiceRunner[S, E, C] // host driver for invoked services, nil when none wired @@ -144,7 +145,9 @@ func (h *Handle[S, E, C]) drainClock() []state.JournalEntry { if h.recClock == nil { return nil } - return h.recClock.drain() + out := append(h.carried, h.recClock.drain()...) + h.carried = nil + return out } // absorbDrivers feeds a transition's effects into every wired host driver so the @@ -221,10 +224,10 @@ func (h *Handle[S, E, C]) armNow(before int) time.Time { } // armTimersFrom mirrors the ScheduleAfter/CancelScheduled effects a chained timer -// transition emitted into the Handle's timer table at the tick instant now, so a -// timer a fired transition re-armed (the next link in an `after` chain) is tracked -// for the next checkpoint. It is the Tick-side analog of absorbTimers, sharing the -// tick's single recorded instant rather than reading the clock again. +// transition emitted into the Handle's timer table at now, the instant the scheduler +// armed them, so a timer a fired transition re-armed (the next link in an `after` chain) +// is tracked for the next checkpoint. It is the Tick-side analog of absorbTimers, using a +// recorded instant rather than reading the clock again. func (h *Handle[S, E, C]) armTimersFrom(effects []state.Effect, now time.Time) { for _, eff := range effects { switch e := eff.(type) { @@ -239,28 +242,32 @@ func (h *Handle[S, E, C]) armTimersFrom(effects []state.Effect, now time.Time) { } } -// tickInstant returns the single clock instant a tick's recorded reads share (the -// FakeClock does not advance within a tick; a SystemClock's intra-tick drift is -// sub-timer). It reads the first recorded JournalClockRead. ok is false when the -// tick recorded no clock read (it fired nothing), so the caller skips reconciliation. -func tickInstant(entries []state.JournalEntry) (time.Time, bool) { +// absorbRead returns the clock instant the scheduler armed result i's timers at, in a +// tick that fired n timers. A Scheduler.Tick reads the clock once to pick each due timer +// and once to absorb its effects, and a transition reads no clock, so result i's instant +// is read 2i+1 of the tick's own reads, the last tickReads(n) in entries. +func absorbRead(entries []state.JournalEntry, i, n int) (time.Time, bool) { + var reads []int64 for _, e := range entries { if e.Kind == state.JournalClockRead { - return time.Unix(0, e.ClockUnixNano).UTC(), true + reads = append(reads, e.ClockUnixNano) } } - return time.Time{}, false + at := len(reads) - tickReads(n) + 2*i + 1 + if at < 0 || at >= len(reads) { + return time.Time{}, false + } + return time.Unix(0, reads[at]).UTC(), true } -// dropFiredTimers removes from the Handle's timer table the timers a Tick fired, so -// a checkpoint taken after the tick does not persist an already-fired timer. The -// scheduler removes a due timer before re-firing it; the Handle mirrors that by -// dropping every timer whose deadline is at or before now, matching the scheduler's -// dueLocked selection. -func (h *Handle[S, E, C]) dropFiredTimers(now time.Time) { - for id, pt := range h.timers { - if !pt.due.After(now) { - delete(h.timers, id) +// dropFiredTimers removes from table every timer the scheduler no longer holds, so a +// checkpoint taken after a tick does not persist one that fired. It asks the scheduler +// rather than comparing deadlines, because a timer can come due partway through a tick, +// or arm and fire within one. +func dropFiredTimers[S comparable, E comparable, C any](table map[string]pendingTimer, sched *state.Scheduler[S, E, C]) { + for id := range table { + if !sched.HasPending(id) { + delete(table, id) } } } @@ -282,6 +289,9 @@ func (h *Handle[S, E, C]) ID() InstanceID { return h.id } // InstanceID that already exists in the Store reports ErrInstanceExists rather // than clobbering its recorded baseline. // +// Start arms the initial state's `after` timers and carries them in the baseline, so a +// Tick fires them and recovery re-arms them. +// // Start applies the initial entry's domain effects before the baseline write, and only // if the entry succeeded: a failed entry still returns a Handle and a nil error, and // Instance().InitialErr reports it. They're at-least-once, unlike a step's: a handler failure or a @@ -309,6 +319,20 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt castOpts := append([]state.CastOption[S]{state.WithClock[S](recClock), state.WithUnboundedHistory[S]()}, opts...) inst := r.machine.Cast(input, castOpts...) + // Arm the timers of the state the instance starts in, and carry their deadlines in + // the baseline, which is what a recovery re-arms them from. The arming clock read + // belongs to no step, so it doesn't reach the first Record. + initial := inst.InitialEffects() + sched := state.NewScheduler(inst) + timers := map[string]pendingTimer{} + if hasTimerEffect(initial) { + sched.Absorb(ctx, initial) + reads := recClock.drain() + if n := len(reads); n > 0 { + mirrorArmed(timers, initial, time.Unix(0, reads[n-1].ClockUnixNano).UTC()) + } + } + // Encode a baseline checkpoint at baselineStep (below the first fired step), so the // instance is reconstructable from the Store before any event is fired. The baseline // is the cast snapshot; the tail then accumulates fired steps on top of it. Encoding @@ -317,9 +341,9 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt if err != nil { return nil, fmt.Errorf("durable: marshaling start snapshot for %q: %w", id, err) } - // 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, 0) + // Wrap the baseline snapshot in a checkpoint envelope so every checkpoint a recover + // loads carries the same durable envelope shape. + baseEnv, err := marshalCheckpoint(snap, timers, 0) if err != nil { return nil, fmt.Errorf("durable: marshaling start checkpoint envelope for %q: %w", id, err) } @@ -328,7 +352,7 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt // re-derive these effects after the baseline, so they're at-least-once. An entry that // failed applies none, as a failed Fire dispatches none. if inst.InitialErr() == nil { - if err := r.applyStartEffects(ctx, id, dispatchableEffects(baselineStep, inst.InitialEffects())); err != nil { + if err := r.applyStartEffects(ctx, id, dispatchableEffects(baselineStep, initial)); err != nil { return nil, err } } @@ -364,9 +388,9 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt id: id, inst: inst, nextStep: 0, - sched: state.NewScheduler(inst), + sched: sched, recClock: recClock, - timers: map[string]pendingTimer{}, + timers: timers, svc: svc, svcBuf: &svcBuf, actors: actors, @@ -469,9 +493,13 @@ func (h *Handle[S, E, C]) Fire(ctx context.Context, event E, opts ...state.FireO // due order, re-firing the delayed events through the durable instance and // recording the clock readings the tick consumed. A host calls it from its own // timer loop (with a real clock) or a test calls it after advancing a fake clock. -// It records one tick barrier — carrying the consumed clock readings and the -// count of timers fired — so recovery re-derives the same timers at their -// recorded instants. It returns the FireResults of the timers it fired, in order. +// When a timer fired it records one tick barrier — carrying the consumed clock readings +// and the count of timers fired — so recovery re-derives the same timers at their +// recorded instants. A tick that fires nothing records nothing. It returns the +// FireResults of the timers it fired, in order. A timer whose transition fails is +// consumed and reported only in its FireResult's Err, not in Tick's error. Tick applies +// the step's domain effects: a failed dispatch, or a KindedEffect that won't marshal, +// fails the Handle. func (h *Handle[S, E, C]) Tick(ctx context.Context) ([]state.FireResult[S], error) { h.mu.Lock() defer h.mu.Unlock() @@ -495,25 +523,25 @@ func (h *Handle[S, E, C]) Tick(ctx context.Context) ([]state.FireResult[S], erro } } + // A tick that fired nothing changes nothing, so it records nothing, and its clock + // read goes with it: replay runs no tick for it, so it consumes no recorded reading. + if len(results) == 0 { + if h.recClock != nil { + h.recClock.drain() + } + return results, nil + } step := h.nextStep rec := Record{Step: step, Tick: true, TickSteps: len(results), Entries: h.drainClock()} // Reconcile the Handle's timer table with the tick: drop the timers it fired and // arm any a chained transition scheduled, so a checkpoint after the tick persists - // exactly the still-pending deadlines. The clock does not advance within one tick - // (a FakeClock advances only between ticks; a SystemClock's intra-tick drift is - // sub-timer), so the tick's recorded reads share one instant: use it for both the - // drop (timers due at or before it fired) and any chained re-arm. - if now, ok := tickInstant(rec.Entries); ok { - h.dropFiredTimers(now) - for i := range results { + // exactly the still-pending deadlines. + for i := range results { + if now, ok := absorbRead(rec.Entries, i, len(results)); ok { h.armTimersFrom(results[i].Effects, now) } } - // A tick that read no clock and fired nothing records nothing: there is no - // nondeterminism to journal and no step was produced. - if len(rec.Entries) == 0 && rec.TickSteps == 0 { - return results, nil - } + dropFiredTimers(h.timers, h.sched) var effects []state.Effect for i := range results { effects = append(effects, results[i].Effects...) @@ -667,12 +695,32 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S actorBuf := make([]state.JournalEntry, 0) actors := r.newActorSystem(inst) - // armed mirrors the scheduler's pending timers re-established by tail replay, - // with their recorded deadlines, so a persisted-but-compacted timer is re-armed - // only when the tail did not already re-arm it, and the Handle's table is seeded - // with every still-pending timer's deadline for the next checkpoint. + // armed mirrors the scheduler's pending timers with their recorded deadlines. It starts + // from the timers the checkpoint carries, tail replay updates it, and it seeds the + // Handle's table for the next checkpoint. armed := map[string]pendingTimer{} + // Re-arm the timers pending at the checkpoint before the tail replays, because the + // tail can fire them. Their deadlines are absolute, so any fixed instant works as the + // arming clock read, and holding the replay clock on one keeps this read from + // consuming a recorded reading. + const armedAt = 0 + reEffects, reSeeded, reErr := reArmEffects(r.cfg.eventCodec, persistedTimers, time.Unix(armedAt, 0).UTC()) + if reErr != nil { + return nil, fmt.Errorf("durable: re-arming pending timers for %q: %w", id, reErr) + } + if len(reEffects) > 0 { + repClock.hold(time.Unix(armedAt, 0).UTC()) + sched.Absorb(ctx, reEffects) + repClock.release() + } + for id, pt := range reSeeded { + armed[id] = pt + } + + reads := 0 // clock readings through the current Record + settles := settleClock{clock: repClock} + carry := false for i := range tail { // A canceled context can stop a replayed settle or tick with no error, leaving // state short of the journal, so check it before each Record. @@ -680,6 +728,8 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S return nil, fmt.Errorf("durable: recovering %q: %w", id, err) } rec := &tail[i] + reads += clockReadCount(rec) + settles.next(rec) switch { case hasActorEntry(rec): // Re-fire each recorded parent transition the live delivery drove, in @@ -705,26 +755,32 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S } settled = append(settled, effects...) } + // Arm the timers the settle scheduled, as the live RunService did. + if hasTimerEffect(settled) { + at, lost := settles.absorb(rec, reads, func() { sched.Absorb(ctx, settled) }) + mirrorArmed(armed, settled, at) + carry = carry || lost + } if err := r.dispatchReplayEffects(ctx, id, rec.Step, settled); err != nil { return nil, err } + case rec.Tick && rec.TickSteps == 0: + // Only an older journal holds a tick that fired nothing. Don't tick: a timer + // re-armed from a checkpoint could be due by now, and the Record says otherwise. case rec.Tick: // Re-derive the timers the live tick fired: ticking the replay clock // returns the recorded readings, so the same deadlines come due and the // same timers fire, at their recorded instants. Mirror the tick into the // armed table — drop the timers it fired, arm any a chained transition // re-scheduled — so the deadline table tracks the still-pending set. + repClock.skipTo(reads - tickReads(rec.TickSteps)) tickRes := sched.Tick(ctx) - if now, ok := tickInstant(rec.Entries); ok { - for id, pt := range armed { - if !pt.due.After(now) { - delete(armed, id) - } - } - for ri := range tickRes { + for ri := range tickRes { + if now, ok := absorbRead(rec.Entries, ri, rec.TickSteps); ok { mirrorArmed(armed, tickRes[ri].Effects, now) } } + dropFiredTimers(armed, sched) var ticked []state.Effect for ri := range tickRes { ticked = append(ticked, tickRes[ri].Effects...) @@ -753,13 +809,11 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S // the recorded clock read the Absorb consumed (captured at the recorded // step rather than the live wall clock). if hasTimerEffect(res.Effects) { + repClock.skipTo(reads - 1) sched.Absorb(ctx, res.Effects) - // The step's recorded clock read (the one the live Absorb consumed) is - // the instant to derive each re-armed deadline from, so the mirror uses - // recorded time, not the live wall clock. - if now, ok := tickInstant(rec.Entries); ok { - mirrorArmed(armed, res.Effects, now) - } + // The recorded read the Absorb consumed is the instant to derive each + // deadline from, so the mirror uses recorded time, not the wall clock. + mirrorArmed(armed, res.Effects, repClock.lastRead()) } // Arm this step's services so a service it entered is in flight for the // recorded settlement that follows, mirroring the live Fire path. @@ -778,7 +832,15 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S // Drain any clock readings the replay re-recorded after the cursor exhausted // (none in the common case where replay consumes exactly the recorded - // readings): they belong to no new step and must not leak into the next Fire. + // readings): they belong to no new step and must not leak into the next Fire. The + // exception is a read an older settle lost: the next step's Record carries it, so a + // later replay arms that settle's timers at the same instant. Skip the cursor to the + // end, so no reading left in the tail reaches the live clock. + repClock.skipTo(reads) + var carried []state.JournalEntry + if carry { + carried = append(carried, buf...) + } buf = buf[:0] // Arm the live actors at the resume point from the final configuration's @@ -791,28 +853,6 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S actors.Absorb(ctx, inst.StartEffects()) } - // Re-arm any timer that was pending at the checkpoint but whose arming - // ScheduleAfter was compacted out of the tail, so it survives compaction: the - // tail replay could not re-arm it (its arm is gone), so re-arm it here at its - // persisted absolute deadline. Already-armed timers (re-established by the tail) - // are skipped, so a timer the tail still carried is never double-armed. The - // remaining delay is the persisted deadline minus the recovery clock's now, so - // the re-armed timer fires at its recorded instant — an already-elapsed deadline - // arms as immediately due and fires on the next Tick. - reEffects, reSeeded, reErr := reArmEffects(r.cfg.eventCodec, persistedTimers, armed, repClock.Now()) - if reErr != nil { - return nil, fmt.Errorf("durable: re-arming pending timers for %q: %w", id, reErr) - } - if len(reEffects) > 0 { - sched.Absorb(ctx, reEffects) - } - for id, pt := range reSeeded { - armed[id] = pt - } - // The re-arm reads the recovery clock to compute remaining delays; that reading - // belongs to no recorded step and must not leak into the next Fire's Record. - buf = buf[:0] - // The next step ordinal continues past the highest assigned step. Step ordinals // are assigned monotonically: each Record consumes one ordinal regardless of how // many parent Traces it produced (an actor delivery can drive several), except a @@ -840,6 +880,7 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S nextStep: next, sched: sched, recClock: recClock, + carried: carried, timers: armed, svc: svc, svcBuf: &svcBuf, diff --git a/durable/service.go b/durable/service.go index 6cbf4308..170602e4 100644 --- a/durable/service.go +++ b/durable/service.go @@ -144,7 +144,8 @@ func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.Fire *h.svcBuf = append(*h.svcBuf, entry) } - rec := Record{Step: h.nextStep, Entries: h.drainServices()} + // The clock read belongs to this step when the settle armed a timer. + rec := Record{Step: h.nextStep, Entries: append(h.drainServices(), h.drainClock()...)} if err := h.commitStep(ctx, &rec, res.Effects); err != nil { return res, true, err } diff --git a/durable/service_test.go b/durable/service_test.go index 9d9223b1..7d7387c0 100644 --- a/durable/service_test.go +++ b/durable/service_test.go @@ -487,10 +487,572 @@ func fetchThenWaitMachine(fn state.ServiceFn[svcCtx]) *state.Machine[string, str Initial("idle"). Transition("idle").On("start").GoTo("loading"). Transition("loading").On("ok").GoTo("waiting").Do("notify"). + Transition("waiting").On("poke").GoTo("waiting"). 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) + } + clk.Advance(time.Second) // the settle's read differs from the earlier ones + 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) + } + + // A clock still at the start proves the timer replays at its recorded instant. + recovered, err := durable.Recover(ctx, m, store, id, append(opts, durable.WithRunnerClock[string, string, svcCtx](state.NewFakeClock(epoch)))...) + 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) + } + if got := atomic.LoadInt64(&calls); got != 1 { + t.Fatalf("service ran %d times, want 1: recovery replays the recorded result", got) + } +} + +// TestStateAt_TimerArmedBySettle_Fires proves a historical read arms a timer from a +// recorded service result, so it fires in the replayed window as it did live. +func TestStateAt_TimerArmedBySettle_Fires(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := fetchThenWaitMachine(fn) + store := durable.NewMemStore(durable.WithHistory()) + 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)), + } + h, err := durable.NewRunner(m, store, opts...).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)) + clk.Advance(6 * time.Second) + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + view, err := durable.StateAt(ctx, m, store, id, 2, opts...) // the tick's barrier + if err != nil { + t.Fatalf("StateAt: %v", err) + } + if got := view.Instance().Current(); got != "done" { + t.Fatalf("StateAt state %q, want done", got) + } +} + +// TestRecover_TimerArmedBySettle_SurvivesALaterCheckpoint proves recovery keeps a timer +// armed by a replayed service result in the table the next checkpoint persists, so a +// second recovery from that checkpoint still fires it. +func TestRecover_TimerArmedBySettle_SurvivesALaterCheckpoint(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)), + durable.WithCheckpointEvery[string, string, svcCtx](3), + } + h, err := durable.NewRunner(m, store, opts...).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)) + + first, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("first Recover: %v", err) + } + for range 3 { // cross a checkpoint boundary after the settle has replayed + if _, err = first.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire(poke): %v", err) + } + } + + second, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("second Recover: %v", err) + } + clk.Advance(6 * time.Second) + if _, err = second.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + if got := second.Instance().Current(); got != "done" { + t.Fatalf("state %q after the second recovery, want done", got) + } +} + +// fetchChainMachine arms a timer on a service result, then chains a second one. +func fetchChainMachine(fn state.ServiceFn[svcCtx]) *state.Machine[string, string, svcCtx] { + return state.ForgeFor[svcCtx]("fetchChain"). + Service("fetch", fn). + State("idle"). + State("loading").Invoke("fetch", state.WithInvokeOnDone("ok")). + State("waiting"). + State("next"). + State("done"). + Initial("idle"). + Transition("idle").On("start").GoTo("loading"). + Transition("loading").On("ok").GoTo("waiting"). + Transition("waiting").On("poke").GoTo("waiting"). + Transition("waiting").After(5 * time.Second).On("first").GoTo("next"). + Transition("next").After(5 * time.Second).On("second").GoTo("done"). + Quench() +} + +// olderSettle splits a settle Record as an older journal wrote it: the Record without its +// clock read, and that read, which the older code left for the next Record to carry. +func olderSettle(settle durable.Record) (durable.Record, state.JournalEntry) { + var read state.JournalEntry + var kept []state.JournalEntry + for _, e := range settle.Entries { + if e.Kind == state.JournalClockRead { + read = e + continue + } + kept = append(kept, e) + } + settle.Entries = kept + return settle, read +} + +// clockReadAt is a recorded clock read d after epoch. +func clockReadAt(d time.Duration) state.JournalEntry { + return state.JournalEntry{Kind: state.JournalClockRead, ClockUnixNano: epoch.Add(d).UnixNano()} +} + +// startFetchChain starts a fetchChainMachine and settles its service. +func startFetchChain( + t *testing.T, + m *state.Machine[string, string, svcCtx], + store durable.Store, + id durable.InstanceID, + opts []durable.Option[string, string, svcCtx], +) *durable.Handle[string, string, svcCtx] { + t.Helper() + ctx := context.Background() + h, err := durable.NewRunner(m, store, opts...).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("fetchChain", "loading", 0)) + return h +} + +// appendAll appends recs to store, failing the test on any error. +func appendAll(t *testing.T, store durable.Store, id durable.InstanceID, recs ...durable.Record) { + t.Helper() + for _, rec := range recs { + if _, err := store.Append(context.Background(), id, rec); err != nil { + t.Fatalf("Append: %v", err) + } + } +} + +// TestRecover_OlderJournal_SettleReadInIdleTick_KeepsReplayClockAligned proves replay stays +// on the recorded readings when an older journal carries a settle's clock read in the idle +// tick Record after it. Replay consumes that read at the settle, so skipping the idle tick +// must not consume it again, or the last tick reads the live clock and fires a timer no +// Record holds. +func TestRecover_OlderJournal_SettleReadInIdleTick_KeepsReplayClockAligned(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := fetchChainMachine(fn) + clk := state.NewFakeClock(epoch) + id := durable.InstanceID("fetch-1") + opts := []durable.Option[string, string, svcCtx]{ + durable.WithRunnerClock[string, string, svcCtx](clk), + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn)), + } + store := durable.NewMemStore() + h := startFetchChain(t, m, store, id, opts) + clk.Advance(6 * 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) + } + + settle, settleRead := olderSettle(tail[1]) + firstIdle := durable.Record{Step: 2, Tick: true, Entries: []state.JournalEntry{settleRead, clockReadAt(time.Second)}} + secondIdle := durable.Record{Step: 3, Tick: true, Entries: []state.JournalEntry{clockReadAt(2 * time.Second)}} + tick := tail[2] + tick.Step = 4 + older := durable.NewMemStore() + if _, err = durable.NewRunner(m, older, opts...).Start(ctx, id, svcCtx{}, state.WithInitialState("idle")); err != nil { + t.Fatalf("Start: %v", err) + } + appendAll(t, older, id, tail[0], settle, firstIdle, secondIdle, tick) + + clk.Advance(94 * time.Second) // the wall clock is far past every timer at recovery + recovered, err := durable.Recover(ctx, m, older, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if got := recovered.Instance().Current(); got != "next" { + t.Fatalf("recovered state %q, want next", got) + } +} + +// TestRecover_OlderJournal_TickPassesAnEarlierSettleRead proves a tail tick fires its +// timer when an older journal left a read at the front of its Record that no replayed +// settle consumed, because a checkpoint compacted the settle away. +func TestRecover_OlderJournal_TickPassesAnEarlierSettleRead(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := fetchChainMachine(fn) + clk := state.NewFakeClock(epoch) + id := durable.InstanceID("fetch-1") + opts := []durable.Option[string, string, svcCtx]{ + durable.WithRunnerClock[string, string, svcCtx](clk), + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn)), + } + // One run takes the checkpoint at the settle; a second, with none, gives the tick Record. + checkpointed := durable.NewMemStore() + startFetchChain(t, m, checkpointed, id, append(opts[:2:2], durable.WithCheckpointEvery[string, string, svcCtx](2))) + snapshot, compacted, err := checkpointed.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(compacted) != 0 { + t.Fatalf("tail holds %d records, want the settle checkpointed", len(compacted)) + } + store := durable.NewMemStore() + h := startFetchChain(t, m, store, id, opts) + clk.Advance(6 * 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) + } + + tick := tail[2] + tick.Entries = append([]state.JournalEntry{clockReadAt(0)}, tick.Entries...) + older := durable.NewMemStore() + if _, err = durable.NewRunner(m, older, opts...).Start(ctx, id, svcCtx{}, state.WithInitialState("idle")); err != nil { + t.Fatalf("Start: %v", err) + } + appendAll(t, older, id, durable.Record{Step: 0}, durable.Record{Step: 1}) + if err = older.Checkpoint(ctx, id, snapshot, 1); err != nil { + t.Fatalf("Checkpoint: %v", err) + } + appendAll(t, older, id, tick) + + recovered, err := durable.Recover(ctx, m, older, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if got := recovered.Instance().Current(); got != "next" { + t.Fatalf("recovered state %q, want next", got) + } + view, err := durable.StateAt(ctx, m, older, id, tick.Step, opts...) + if err != nil { + t.Fatalf("StateAt: %v", err) + } + if got := view.Instance().Current(); got != "next" { + t.Fatalf("StateAt state %q, want next", got) + } +} + +// TestRecover_OlderJournal_SettleSkipsAnUnusedRead proves a replayed settle arms from its +// own reading, not a stale one an older journal left in an earlier Record that no Absorb +// consumed. +func TestRecover_OlderJournal_SettleSkipsAnUnusedRead(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := fetchChainMachine(fn) + clk := state.NewFakeClock(epoch) + id := durable.InstanceID("fetch-1") + opts := []durable.Option[string, string, svcCtx]{ + durable.WithRunnerClock[string, string, svcCtx](clk), + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn)), + } + store := durable.NewMemStore() + startFetchChain(t, m, store, id, opts) + _, tail, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + + fire := tail[0] + fire.Entries = append([]state.JournalEntry{clockReadAt(0)}, fire.Entries...) // unused + settle, _ := olderSettle(tail[1]) + older := durable.NewMemStore() + if _, err = durable.NewRunner(m, older, opts...).Start(ctx, id, svcCtx{}, state.WithInitialState("idle")); err != nil { + t.Fatalf("Start: %v", err) + } + appendAll(t, older, id, fire, settle, + durable.Record{Step: 2, Tick: true, Entries: []state.JournalEntry{clockReadAt(time.Second)}}) + + recovered, err := durable.Recover(ctx, m, older, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + clk.Advance(5*time.Second + 500*time.Millisecond) // the timer is due at 6s, from the settle's read + for range 2 { // a misaligned replay leaves a reading for the first + fired, err := recovered.Tick(ctx) + if err != nil { + t.Fatalf("Tick: %v", err) + } + if len(fired) != 0 { + t.Fatalf("Tick fired %d timers before the deadline, want 0", len(fired)) + } + } +} + +// TestRecover_OlderJournal_SettleReadLostToARestart_LeavesTheNextStepsReads proves a +// read-less settle doesn't consume the next Record's reading when a restart lost its own. +// That reading stands in for the lost one, and the tick after it still fires what it +// recorded. +func TestRecover_OlderJournal_SettleReadLostToARestart_LeavesTheNextStepsReads(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := fetchChainMachine(fn) + clk := state.NewFakeClock(epoch) + id := durable.InstanceID("fetch-1") + opts := []durable.Option[string, string, svcCtx]{ + durable.WithRunnerClock[string, string, svcCtx](clk), + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn)), + } + store := durable.NewMemStore() + h := startFetchChain(t, m, store, id, opts) + clk.Advance(6 * 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) + } + + settle, _ := olderSettle(tail[1]) // the restart lost its read + older := durable.NewMemStore() + if _, err = durable.NewRunner(m, older, opts...).Start(ctx, id, svcCtx{}, state.WithInitialState("idle")); err != nil { + t.Fatalf("Start: %v", err) + } + tick := tail[2] + tick.Step = 3 + appendAll(t, older, id, tail[0], settle, + durable.Record{Step: 2, Tick: true, Entries: []state.JournalEntry{clockReadAt(time.Second)}}, tick) + + recovered, err := durable.Recover(ctx, m, older, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if got := recovered.Instance().Current(); got != "next" { + t.Fatalf("recovered state %q, want next", got) + } + view, err := durable.StateAt(ctx, m, older, id, tick.Step, opts...) + if err != nil { + t.Fatalf("StateAt: %v", err) + } + if got := view.Instance().Current(); got != "next" { + t.Fatalf("StateAt state %q, want next", got) + } +} + +// TestRecover_OlderJournal_TailEndingOnASettle_CarriesItsRead proves recovery keeps the +// read it took for a settle that ended an older journal: the next step's Record carries +// it, so a later recovery arms that timer at the same instant. +func TestRecover_OlderJournal_TailEndingOnASettle_CarriesItsRead(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := fetchChainMachine(fn) + clk := state.NewFakeClock(epoch) + id := durable.InstanceID("fetch-1") + opts := []durable.Option[string, string, svcCtx]{ + durable.WithRunnerClock[string, string, svcCtx](clk), + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn)), + } + store := durable.NewMemStore() + startFetchChain(t, m, store, id, opts) + _, tail, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + settle, _ := olderSettle(tail[1]) + older := durable.NewMemStore() + if _, err = durable.NewRunner(m, older, opts...).Start(ctx, id, svcCtx{}, state.WithInitialState("idle")); err != nil { + t.Fatalf("Start: %v", err) + } + appendAll(t, older, id, tail[0], settle) + + clk.Advance(3 * time.Second) + first, err := durable.Recover(ctx, m, older, id, opts...) + if err != nil { + t.Fatalf("first Recover: %v", err) + } + clk.Advance(6 * time.Second) // due at 8s, from the recovery's read + if _, err = first.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + if got := first.Instance().Current(); got != "next" { + t.Fatalf("live state %q, want next", got) + } + second, err := durable.Recover(ctx, m, older, id, opts...) + if err != nil { + t.Fatalf("second Recover: %v", err) + } + if got := second.Instance().Current(); got != "next" { + t.Fatalf("recovered state %q, want next", got) + } +} + +// TestRecover_OlderJournal_LeftoverReadStaysOutOfTheLiveClock proves a reading no step +// consumed at the end of a tail doesn't reach the live clock, so a timer armed after +// recovery measures from the real time. +func TestRecover_OlderJournal_LeftoverReadStaysOutOfTheLiveClock(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := fetchChainMachine(fn) + clk := state.NewFakeClock(epoch) + id := durable.InstanceID("fetch-1") + opts := []durable.Option[string, string, svcCtx]{ + durable.WithRunnerClock[string, string, svcCtx](clk), + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn)), + } + store := durable.NewMemStore() + h, err := durable.NewRunner(m, store, opts...).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: %v", err) + } + _, tail, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + fire := tail[0] + fire.Entries = append([]state.JournalEntry{clockReadAt(0)}, fire.Entries...) // no Absorb used it + older := durable.NewMemStore() + if _, err = durable.NewRunner(m, older, opts...).Start(ctx, id, svcCtx{}, state.WithInitialState("idle")); err != nil { + t.Fatalf("Start: %v", err) + } + appendAll(t, older, id, fire) + + clk.Advance(94 * time.Second) + recovered, err := durable.Recover(ctx, m, older, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + mustRunService(t, recovered, state.InvokeID("fetchChain", "loading", 0)) // arms 5s, due at 99s + clk.Advance(time.Second) + fired, err := recovered.Tick(ctx) + if err != nil { + t.Fatalf("Tick: %v", err) + } + if len(fired) != 0 { + t.Fatalf("Tick fired %d timers before the deadline, want 0", len(fired)) + } +} + +// TestRecover_OlderJournal_SettleTimerReachesTheNextCheckpoint proves a timer armed by an +// older settle Record, which holds no clock read, still reaches the Handle's table, so the +// next checkpoint persists it and a second recovery fires it. +func TestRecover_OlderJournal_SettleTimerReachesTheNextCheckpoint(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := fetchChainMachine(fn) + clk := state.NewFakeClock(epoch) + id := durable.InstanceID("fetch-1") + opts := []durable.Option[string, string, svcCtx]{ + durable.WithRunnerClock[string, string, svcCtx](clk), + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn)), + durable.WithCheckpointEvery[string, string, svcCtx](6), + } + store := durable.NewMemStore() + startFetchChain(t, m, store, id, opts) + _, tail, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + + settle, settleRead := olderSettle(tail[1]) + older := durable.NewMemStore() + if _, err = durable.NewRunner(m, older, opts...).Start(ctx, id, svcCtx{}, state.WithInitialState("idle")); err != nil { + t.Fatalf("Start: %v", err) + } + appendAll(t, older, id, tail[0], settle, + durable.Record{Step: 2, Tick: true, Entries: []state.JournalEntry{settleRead, clockReadAt(time.Second)}}) + + first, err := durable.Recover(ctx, m, older, id, opts...) + if err != nil { + t.Fatalf("first Recover: %v", err) + } + for range 3 { // cross a checkpoint boundary after the settle has replayed + if _, err = first.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire(poke): %v", err) + } + } + second, err := durable.Recover(ctx, m, older, id, opts...) + if err != nil { + t.Fatalf("second Recover: %v", err) + } + clk.Advance(6 * time.Second) + if _, err = second.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + if got := second.Instance().Current(); got != "next" { + t.Fatalf("state %q after the second recovery, want next", 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. diff --git a/durable/store.go b/durable/store.go index b57254f7..cbeda795 100644 --- a/durable/store.go +++ b/durable/store.go @@ -80,7 +80,7 @@ type Record struct { Tick bool // TickSteps is the number of timer transitions the tick at this barrier fired - // (zero for a tick that found nothing due). It lets replay advance its step + // (zero only in a journal from before idle ticks stopped recording). It lets replay advance its step // accounting across a re-derived tick without a recorded Event per timer. TickSteps int } diff --git a/durable/timer_checkpoint_test.go b/durable/timer_checkpoint_test.go index 3834a216..dee74767 100644 --- a/durable/timer_checkpoint_test.go +++ b/durable/timer_checkpoint_test.go @@ -381,3 +381,542 @@ 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) + } + } + _, 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)) + } + 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) + } + } + + if got := h.Instance().Current(); got != "done" { + t.Fatalf("live state %q, want done", got) + } + + // A clock still at the start proves the timer replays at its recorded instant. + recovered, err := durable.Recover(ctx, m, store, id, append(opts, durable.WithRunnerClock[string, string, *timerCtx](state.NewFakeClock(epoch)))...) + 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) + } +} + +// TestStateAt_TimerArmedAtStart_Fires proves a historical read re-arms the timers a +// baseline carries, so a timer armed by Start fires in the replayed window. +func TestStateAt_TimerArmedAtStart_Fires(t *testing.T) { + ctx := context.Background() + m := timerMachine() + store := durable.NewMemStore(durable.WithHistory()) + id := durable.InstanceID("timer-1") + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *timerCtx]{durable.WithRunnerClock[string, string, *timerCtx](clk)} + h, err := durable.NewRunner(m, store, opts...).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) + } + view, err := durable.StateAt(ctx, m, store, id, 0, opts...) + if err != nil { + t.Fatalf("StateAt: %v", err) + } + if got := view.Instance().Current(); got != "done" { + t.Fatalf("StateAt state %q, want done", got) + } +} + +// TestRecover_OlderIdleTickRecord_FiresNothing proves an idle tick Record from an older +// journal replays as nothing fired, even with a due re-armed timer. +func TestRecover_OlderIdleTickRecord_FiresNothing(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), + } + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, &timerCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + for _, ev := range []string{"arm", "poke", "poke"} { // the checkpoint carries the timer + if _, err = h.Fire(ctx, ev); err != nil { + t.Fatalf("Fire(%s): %v", ev, err) + } + } + // An older build recorded this idle tick with a reading past the timer's deadline. + late := epoch.Add(time.Hour).UnixNano() + idle := durable.Record{Step: 3, Tick: true, Entries: []state.JournalEntry{{Kind: state.JournalClockRead, ClockUnixNano: late}}} + if _, err = store.Append(ctx, id, idle); err != nil { + t.Fatalf("Append: %v", err) + } + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if got := recovered.Instance().Current(); got != "waiting" { + t.Fatalf("recovered in %q, want waiting: the Record says nothing fired", got) + } +} + +// driftingClock moves by step on every read, as a real clock moves during a tick. +type driftingClock struct { + now time.Time + step time.Duration +} + +func (c *driftingClock) Now() time.Time { + t := c.now + c.now = c.now.Add(c.step) + return t +} + +func (c *driftingClock) After(time.Duration) <-chan time.Time { return nil } + +// twoTimerMachine holds two timers on a parent state. Each fires into one of parent's +// children, so firing one cancels neither. +func twoTimerMachine() *state.Machine[string, string, *timerCtx] { + return state.ForgeFor[*timerCtx]("twoTimers"). + State("idle"). + SuperState("parent"). + Initial("p1"). + SubState("p1"). + SubState("p2"). + SubState("p3"). + EndSuperState(). + Initial("idle"). + Transition("idle").On("go").GoTo("parent"). + Transition("parent").On("poke").GoTo("p1"). + Transition("parent").After(10 * time.Second).On("first").GoTo("p2"). + Transition("parent").After(10*time.Second + 10*time.Millisecond).On("second").GoTo("p3"). + Quench() +} + +// tickAcrossTwoDeadlines starts a twoTimerMachine and ticks it so the tick's first read +// finds one timer due and its second read finds the other. +func tickAcrossTwoDeadlines( + t *testing.T, + m *state.Machine[string, string, *timerCtx], + store durable.Store, + id durable.InstanceID, + clk *driftingClock, + opts []durable.Option[string, string, *timerCtx], +) { + t.Helper() + ctx := context.Background() + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, &timerCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "go"); err != nil { + t.Fatalf("Fire: %v", err) + } + clk.now, clk.step = epoch.Add(10*time.Second+5*time.Millisecond), 20*time.Millisecond + fired, err := h.Tick(ctx) + if err != nil { + t.Fatalf("Tick: %v", err) + } + if len(fired) != 2 { + t.Fatalf("Tick fired %d timers, want 2", len(fired)) + } + clk.step = 0 +} + +// slowChainMachine has a chained timer armed after a slow action, and a second region +// whose timer fires between the chained timer's start and its real deadline. +func slowChainMachine(clk *driftingClock, other time.Duration) *state.Machine[string, string, *timerCtx] { + return state.ForgeFor[*timerCtx]("slowChain"). + Action("slow", func(state.ActionCtx[*timerCtx]) (state.Effect, error) { + clk.now = clk.now.Add(200 * time.Millisecond) + return nil, nil + }). + State("idle"). + SuperState("live"). + Region("first"). + Initial("a1"). + SubState("a1"). + SubState("a2"). + SubState("a3"). + EndRegion(). + Region("second"). + Initial("b1"). + SubState("b1"). + SubState("b2"). + EndRegion(). + EndSuperState(). + Initial("idle"). + Transition("idle").On("arm").GoTo("live"). + Transition("a1").After(10 * time.Second).On("x").GoTo("a2").Do("slow"). + Transition("a2").After(5 * time.Second).On("c").GoTo("a3"). + Transition("b1").After(other).On("y").GoTo("b2"). + Transition("b1").On("poke").GoTo("b1"). + Quench() +} + +// TestTick_ChainedTimerAfterSlowAction_SurvivesRecovery proves the Handle holds a chained +// timer at the deadline the scheduler armed, after the action ran. An earlier deadline +// would drop it when a later tick passes that instant, and recovery would lose it. +func TestTick_ChainedTimerAfterSlowAction_SurvivesRecovery(t *testing.T) { + ctx := context.Background() + clk := &driftingClock{now: epoch, step: 100 * time.Microsecond} + other := 15*time.Second - 100*time.Microsecond + m := slowChainMachine(clk, other) + store := durable.NewMemStore() + id := durable.InstanceID("timer-1") + opts := []durable.Option[string, string, *timerCtx]{ + durable.WithRunnerClock[string, string, *timerCtx](clk), + durable.WithCheckpointEvery[string, string, *timerCtx](1), + } + h, err := durable.NewRunner(m, store, opts...).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: %v", err) + } + for _, at := range []time.Duration{10 * time.Second, other} { + clk.now = epoch.Add(at) + 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) + } + clk.now = epoch.Add(16 * time.Second) + fired, err := recovered.Tick(ctx) + if err != nil { + t.Fatalf("recovered Tick: %v", err) + } + if len(fired) != 1 { + t.Fatalf("recovered Tick fired %d timers, want the chained timer", len(fired)) + } +} + +// TestRecover_ChainedTimerAfterSlowAction_NotFiredEarly proves a checkpointed chained +// timer keeps its real deadline, so a tail tick doesn't fire it before live did. +func TestRecover_ChainedTimerAfterSlowAction_NotFiredEarly(t *testing.T) { + ctx := context.Background() + clk := &driftingClock{now: epoch, step: 100 * time.Microsecond} + other := 15*time.Second - 100*time.Microsecond + m := slowChainMachine(clk, other) + store := durable.NewMemStore() + id := durable.InstanceID("timer-1") + opts := []durable.Option[string, string, *timerCtx]{ + durable.WithRunnerClock[string, string, *timerCtx](clk), + durable.WithCheckpointEvery[string, string, *timerCtx](2), + } + h, err := durable.NewRunner(m, store, opts...).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: %v", err) + } + clk.now = epoch.Add(10 * time.Second) + 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) + } + 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) + } +} + +// TestRecover_ReplayedChainedTimer_KeepsItsDeadline proves a replayed tick mirrors a +// chained timer at the deadline live armed, so the next checkpoint persists that deadline. +func TestRecover_ReplayedChainedTimer_KeepsItsDeadline(t *testing.T) { + ctx := context.Background() + clk := &driftingClock{now: epoch, step: 100 * time.Microsecond} + m := slowChainMachine(clk, time.Minute) + store := durable.NewMemStore() + id := durable.InstanceID("timer-1") + opts := []durable.Option[string, string, *timerCtx]{ + durable.WithRunnerClock[string, string, *timerCtx](clk), + durable.WithCheckpointEvery[string, string, *timerCtx](6), + } + h, err := durable.NewRunner(m, store, opts...).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: %v", err) + } + clk.now = epoch.Add(10 * time.Second) + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + first, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("first Recover: %v", err) + } + for range 3 { // cross a checkpoint boundary after the tick has replayed + if _, err = first.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire(poke): %v", err) + } + } + second, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("second Recover: %v", err) + } + clk.now = epoch.Add(15*time.Second + 100*time.Millisecond) // before 15.2s, the real deadline + fired, err := second.Tick(ctx) + if err != nil { + t.Fatalf("recovered Tick: %v", err) + } + if len(fired) != 0 { + t.Fatalf("recovered Tick fired %d timers before the deadline, want 0", len(fired)) + } +} + +// TestRecover_OlderJournal_TailEndingOnIdleTicks_LeavesNoReadings proves a tail ending on +// idle tick Records leaves none of their readings for the live clock, so the first Tick +// after recovery reads the real time and fires what is due. +func TestRecover_OlderJournal_TailEndingOnIdleTicks_LeavesNoReadings(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)} + h, err := durable.NewRunner(m, store, opts...).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: %v", err) + } + for step := 1; step <= 3; step++ { + read := state.JournalEntry{Kind: state.JournalClockRead, ClockUnixNano: epoch.Add(time.Duration(step) * time.Second).UnixNano()} + idle := durable.Record{Step: step, Tick: true, Entries: []state.JournalEntry{read}} + if _, err = store.Append(ctx, id, idle); err != nil { + t.Fatalf("Append: %v", err) + } + } + + clk.Advance(20 * time.Second) + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + fired, err := recovered.Tick(ctx) + if err != nil { + t.Fatalf("Tick: %v", err) + } + if len(fired) != 1 { + t.Fatalf("first Tick after recovery fired %d timers, want 1", len(fired)) + } +} + +// TestRecover_OlderJournal_FireArmsFromItsOwnRead proves a replayed Fire arms its timer +// from its own reading, the last in its Record, past a read an older journal left in front. +func TestRecover_OlderJournal_FireArmsFromItsOwnRead(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)} + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, &timerCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + clk.Advance(time.Second) + if _, err = h.Fire(ctx, "arm"); err != nil { + t.Fatalf("Fire: %v", err) + } + _, tail, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + + stale := state.JournalEntry{Kind: state.JournalClockRead, ClockUnixNano: epoch.UnixNano()} + fire := tail[0] + fire.Entries = append([]state.JournalEntry{stale}, fire.Entries...) + older := durable.NewMemStore() + if _, err = durable.NewRunner(m, older, opts...).Start(ctx, id, &timerCtx{}, state.WithInitialState("idle")); err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = older.Append(ctx, id, fire); err != nil { + t.Fatalf("Append: %v", err) + } + + recovered, err := durable.Recover(ctx, m, older, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + clk.Advance(9*time.Second + 500*time.Millisecond) // the timer is due at 11s + for range 2 { // a misaligned replay leaves a reading for the first + fired, err := recovered.Tick(ctx) + if err != nil { + t.Fatalf("Tick: %v", err) + } + if len(fired) != 0 { + t.Fatalf("Tick fired %d timers before the deadline, want 0", len(fired)) + } + } +} + +// TestTick_TimerArmedAndFiredInOneTick_NotPersisted proves a timer a tick both arms and +// fires leaves the table, so a checkpoint doesn't persist it for recovery to fire again. +func TestTick_TimerArmedAndFiredInOneTick_NotPersisted(t *testing.T) { + ctx := context.Background() + m := state.ForgeFor[*timerCtx]("zeroDelay"). + State("idle"). + State("wait"). + SuperState("parent"). + Initial("p1"). + SubState("p1"). + SubState("p2"). + SubState("p3"). + EndSuperState(). + Initial("idle"). + Transition("idle").On("arm").GoTo("wait"). + Transition("wait").After(10 * time.Second).On("go").GoTo("parent"). + Transition("parent").After(0).On("zero").GoTo("p2"). + Transition("p2").On("poke").GoTo("p3"). + Quench() + 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](1), + } + h, err := durable.NewRunner(m, store, opts...).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: %v", err) + } + clk.Advance(10 * time.Second) + 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) + } + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + fired, err := recovered.Tick(ctx) + if err != nil { + t.Fatalf("recovered Tick: %v", err) + } + if len(fired) != 0 { + t.Fatalf("recovered Tick fired %d timers, want 0", len(fired)) + } +} + +// TestTick_TimerDueAfterTheFirstRead_NotPersisted proves a timer that comes due between a +// tick's clock reads leaves the table with the others it fired, so a checkpoint doesn't +// persist it and recovery doesn't fire it again. +func TestTick_TimerDueAfterTheFirstRead_NotPersisted(t *testing.T) { + ctx := context.Background() + m := twoTimerMachine() + store := durable.NewMemStore() + id := durable.InstanceID("timer-1") + clk := &driftingClock{now: epoch} + opts := []durable.Option[string, string, *timerCtx]{ + durable.WithRunnerClock[string, string, *timerCtx](clk), + durable.WithCheckpointEvery[string, string, *timerCtx](1), + } + tickAcrossTwoDeadlines(t, m, store, id, clk, opts) + + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + again, err := recovered.Tick(ctx) + if err != nil { + t.Fatalf("recovered Tick: %v", err) + } + if len(again) != 0 { + t.Fatalf("recovered Tick fired %d timers, want 0", len(again)) + } +} + +// TestRecover_TimerDueAfterTheFirstRead_NotPersisted proves the same when the tick is +// replayed: the replayed tick drops the timer, so the next checkpoint doesn't persist it. +func TestRecover_TimerDueAfterTheFirstRead_NotPersisted(t *testing.T) { + ctx := context.Background() + m := twoTimerMachine() + store := durable.NewMemStore() + id := durable.InstanceID("timer-1") + clk := &driftingClock{now: epoch} + opts := []durable.Option[string, string, *timerCtx]{ + durable.WithRunnerClock[string, string, *timerCtx](clk), + durable.WithCheckpointEvery[string, string, *timerCtx](6), + } + tickAcrossTwoDeadlines(t, m, store, id, clk, opts) + + first, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("first Recover: %v", err) + } + for range 3 { // cross a checkpoint boundary after the tick has replayed + if _, err = first.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire(poke): %v", err) + } + } + second, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("second Recover: %v", err) + } + again, err := second.Tick(ctx) + if err != nil { + t.Fatalf("recovered Tick: %v", err) + } + if len(again) != 0 { + t.Fatalf("recovered Tick fired %d timers, want 0", len(again)) + } +} diff --git a/durable/timetravel.go b/durable/timetravel.go index 24a4eea1..354b54f4 100644 --- a/durable/timetravel.go +++ b/durable/timetravel.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "time" "github.com/stablekernel/crucible/state" ) @@ -133,12 +134,12 @@ func StateAt[S comparable, E comparable, C any]( bounded := boundRecords(records, step) repClock := newReplayClock(clockReadings(bounded), cfg.clock) - inst, err := restoreBaseline(m, baseline, repClock) + inst, timers, err := restoreBaseline(m, baseline, repClock) if err != nil { return nil, fmt.Errorf("durable: reconstructing %q at step %d: %w", id, step, err) } - if err := replayThrough(ctx, inst, &cfg, bounded); err != nil { + if err := replayThrough(ctx, inst, &cfg, bounded, timers, repClock); err != nil { return nil, fmt.Errorf("durable: reconstructing %q at step %d: %w", id, step, err) } return &TimeTravelView[S, E, C]{inst: inst, step: step}, nil @@ -195,30 +196,29 @@ func lastStep(records []Record) int { // restoreBaseline unwraps the baseline checkpoint envelope and restores a detached // kernel instance from it under the bounded replay clock, so the reconstruction -// reads recorded instants rather than the wall clock. The baseline carries no armed -// timers (it is the pre-event cast), so its envelope's timer table is empty and only -// the snapshot matters here. +// reads recorded instants rather than the wall clock. It also returns the timers the +// baseline carries, which the replay must re-arm before it runs. func restoreBaseline[S comparable, E comparable, C any]( m *state.Machine[S, E, C], baseline []byte, repClock state.Clock, -) (*state.Instance[S, E, C], error) { - kernelSnap, _, _, err := unmarshalCheckpoint(baseline) +) (*state.Instance[S, E, C], []pendingTimerWire, error) { + kernelSnap, timers, _, err := unmarshalCheckpoint(baseline) if err != nil { - return nil, fmt.Errorf("unmarshaling baseline envelope: %w", err) + return nil, nil, fmt.Errorf("unmarshaling baseline envelope: %w", err) } snap, err := state.UnmarshalSnapshot[S, E, C](kernelSnap) if err != nil { - return nil, fmt.Errorf("unmarshaling baseline snapshot: %w", err) + return nil, nil, fmt.Errorf("unmarshaling baseline snapshot: %w", err) } // Reconstruct in the same full-trace, unbounded-history mode the runner records // in, so a time-travel reconstruction's Trace history is byte-identical to the // live run it replays. inst, err := m.Restore(snap, state.WithRestoreClock[S](repClock), state.WithRestoreUnboundedHistory[S]()) if err != nil { - return nil, fmt.Errorf("restoring baseline snapshot: %w", err) + return nil, nil, fmt.Errorf("restoring baseline snapshot: %w", err) } - return inst, nil + return inst, timers, nil } // replayThrough replays the bounded record window forward into inst, reproducing the @@ -233,14 +233,32 @@ func replayThrough[S comparable, E comparable, C any]( inst *state.Instance[S, E, C], cfg *runnerConfig[S, E, C], bounded []Record, + timers []pendingTimerWire, + repClock *replayClock, ) error { sched := state.NewScheduler(inst) + // Re-arm the baseline's timers before the window replays, because it can fire them. + // As in Recover, the deadlines are absolute, so a held instant keeps this read from + // consuming a recorded reading. + armedAt := time.Unix(0, 0).UTC() + reEffects, _, err := reArmEffects(cfg.eventCodec, timers, armedAt) + if err != nil { + return fmt.Errorf("re-arming baseline timers: %w", err) + } + if len(reEffects) > 0 { + repClock.hold(armedAt) + sched.Absorb(ctx, reEffects) + repClock.release() + } + svc := newReplayServiceRunner(inst, cfg) if svc != nil { svc.Absorb(ctx, inst.StartEffects()) } + reads := 0 // clock readings through the current Record + settles := settleClock{clock: repClock} for i := range bounded { // A canceled context can stop a replayed settle or tick with no error, leaving // state short of the journal, so check it before each Record. @@ -248,6 +266,8 @@ func replayThrough[S comparable, E comparable, C any]( return err } rec := &bounded[i] + reads += clockReadCount(rec) + settles.next(rec) switch { case hasActorEntry(rec): for _, entry := range actorEntries(rec) { @@ -257,11 +277,19 @@ func replayThrough[S comparable, E comparable, C any]( } case hasServiceEntry(rec): for _, entry := range serviceEntries(rec) { - if _, err := replayService(ctx, svc, entry); err != nil { + effects, err := replayService(ctx, svc, entry) + if err != nil { return fmt.Errorf("at step %d: %w", rec.Step, err) } + if hasTimerEffect(effects) { + settles.absorb(rec, reads, func() { sched.Absorb(ctx, effects) }) + } } + case rec.Tick && rec.TickSteps == 0: + // Only an older journal holds a tick that fired nothing. Don't tick: a timer + // re-armed from a checkpoint could be due by now, and the Record says otherwise. case rec.Tick: + repClock.skipTo(reads - tickReads(rec.TickSteps)) sched.Tick(ctx) case len(rec.Event) == 0: continue // a checkpoint-only Record drives no event @@ -279,6 +307,7 @@ func replayThrough[S comparable, E comparable, C any]( return fmt.Errorf("replaying step %d: %w", rec.Step, res.Err) } if hasTimerEffect(res.Effects) { + repClock.skipTo(reads - 1) sched.Absorb(ctx, res.Effects) } if svc != nil {