fix(durable): keep timers across recovery and skip idle ticks

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

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)

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)
+		}
 	}
 }
 

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

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)
+	}
+}