fix(state): fire FakeClock.After only once Advance reaches the deadline

durable/clock_internal_test.go

@@ -39,13 +39,20 @@ func TestRecordingClock_AfterDelegates(t *testing.T) {
 	real := state.NewFakeClock(clockEpoch)
 	c := newRecordingClock(real, &buf)
 
+	ch := c.After(time.Second)
 	select {
-	case got := <-c.After(time.Second):
+	case <-ch:
+		t.Fatal("After fired before the wrapped clock advanced")
+	default:
+	}
+	real.Advance(time.Second)
+	select {
+	case got := <-ch:
 		if !got.Equal(clockEpoch.Add(time.Second)) {
 			t.Fatalf("After: want %v, got %v", clockEpoch.Add(time.Second), got)
 		}
 	default:
-		t.Fatal("After channel did not deliver (FakeClock.After fires immediately)")
+		t.Fatal("After channel did not deliver once the wrapped clock advanced")
 	}
 	if len(buf) != 0 {
 		t.Fatalf("After must record nothing, recorded %d", len(buf))
@@ -78,14 +85,22 @@ func TestReplayClock_ReturnsRecordedThenFallsThrough(t *testing.T) {
 // TestReplayClock_AfterDelegatesToFallback confirms After delegates to the
 // fallback clock.
 func TestReplayClock_AfterDelegatesToFallback(t *testing.T) {
-	c := newReplayClock(nil, state.NewFakeClock(clockEpoch))
+	fallback := state.NewFakeClock(clockEpoch)
+	c := newReplayClock(nil, fallback)
+	ch := c.After(time.Second)
+	select {
+	case <-ch:
+		t.Fatal("After fired before the fallback clock advanced")
+	default:
+	}
+	fallback.Advance(time.Second)
 	select {
-	case got := <-c.After(time.Second):
+	case got := <-ch:
 		if !got.Equal(clockEpoch.Add(time.Second)) {
 			t.Fatalf("After: want %v, got %v", clockEpoch.Add(time.Second), got)
 		}
 	default:
-		t.Fatal("After channel did not deliver")
+		t.Fatal("After channel did not deliver once the fallback clock advanced")
 	}
 }
 

state/CHANGELOG.md

@@ -20,6 +20,16 @@ counts as an additive (minor) versus breaking (major) change. Use the
 
 - Requires Go 1.27.3.
 
+### Fixed
+
+- **`FakeClock.After`** now fires only once `Advance` moves the clock to or past
+  its deadline, sending the deadline as before. It used to fire at once, so a
+  test that selected on it saw every timeout elapse with no advance. A test that
+  received from it without calling `Advance` now blocks: advance the clock to
+  the deadline first. `After(d)` with `d` of zero or less still fires at once
+  with the current instant, as `time.After` does; a negative `d` used to send an
+  instant in the past.
+
 ## [1.0.0] — 2026-06-13
 
 The first stable release. It freezes the data model, the serialized IR, the

state/driver.go

@@ -156,9 +156,26 @@ func (s *Scheduler[S, E, C]) dueLocked(now time.Time) (string, pending[E], bool)
 // FakeClock is a deterministic Clock for tests: time advances only when Advance
 // is called. It implements Clock; pair it with a Scheduler (via WithClock at
 // Cast) to drive `after` transitions with no real waiting. It is concurrency-safe.
+//
+// After behaves like time.After on the fake timeline: its channel stays empty
+// until Advance moves the clock to or past the deadline. A host that selects on
+// After (a silence timeout, a reconnect grace) sees it fire only when the test
+// advances far enough. An unfired waiter stays registered on the clock until
+// an Advance reaches it; there is no Stop to cancel it.
 type FakeClock struct {
-	mu  sync.Mutex
-	now time.Time
+	mu      sync.Mutex
+	now     time.Time
+	seq     uint64
+	waiters []fakeWaiter
+}
+
+// fakeWaiter is one channel handed out by FakeClock.After that hasn't fired yet.
+// seq records the order the waiters were armed in, to break ties between equal
+// deadlines.
+type fakeWaiter struct {
+	due time.Time
+	seq uint64
+	ch  chan time.Time
 }
 
 // NewFakeClock returns a FakeClock starting at the given instant. The zero
@@ -174,24 +191,56 @@ func (c *FakeClock) Now() time.Time {
 	return c.now
 }
 
-// After returns a channel that fires once the fake clock has advanced by at
-// least d from the call. It is provided for Clock conformance; the Scheduler
-// drives elapses through Now + Tick rather than this channel, so a test never
-// blocks on it.
+// After returns a channel that receives once the fake clock reaches the
+// instant of the call plus d. The value sent is that deadline. A d of zero or
+// less fires at once with the current instant, as time.After does. With no
+// Advance far enough, a receive on the channel blocks forever. The Scheduler
+// doesn't use After: it drives elapses through Now and Tick.
 func (c *FakeClock) After(d time.Duration) <-chan time.Time {
 	ch := make(chan time.Time, 1)
 	c.mu.Lock()
-	ch <- c.now.Add(d)
-	c.mu.Unlock()
+	defer c.mu.Unlock()
+	if d <= 0 {
+		ch <- c.now
+		return ch
+	}
+	c.seq++
+	c.waiters = append(c.waiters, fakeWaiter{due: c.now.Add(d), seq: c.seq, ch: ch})
 	return ch
 }
 
-// Advance moves the fake clock forward by d. After advancing, call Scheduler.Tick
-// to fire any now-due timers.
+// Advance moves the fake clock forward by d, then fires every After channel
+// whose deadline the clock has now reached, earliest deadline first, and in
+// the order they were armed when deadlines tie. A goroutine already blocked
+// selecting on several of them receives from the earliest; a receiver that
+// arrives after Advance returns finds them all ready, and select picks at
+// random. After advancing, call Scheduler.Tick to fire any now-due timers.
 func (c *FakeClock) Advance(d time.Duration) {
 	c.mu.Lock()
+	defer c.mu.Unlock()
 	c.now = c.now.Add(d)
-	c.mu.Unlock()
+	var due []fakeWaiter
+	kept := c.waiters[:0]
+	for _, w := range c.waiters {
+		if w.due.After(c.now) {
+			kept = append(kept, w)
+			continue
+		}
+		due = append(due, w)
+	}
+	clear(c.waiters[len(kept):])
+	c.waiters = kept
+	sort.Slice(due, func(i, j int) bool {
+		if !due[i].due.Equal(due[j].due) {
+			return due[i].due.Before(due[j].due)
+		}
+		return due[i].seq < due[j].seq
+	})
+	// Each channel is buffered and receives exactly once, so these sends never
+	// block while the lock is held.
+	for _, w := range due {
+		w.ch <- w.due
+	}
 }
 
 // keep context imported for the driver contract's Fire symmetry.

state/fakeclock_test.go

@@ -0,0 +1,195 @@
+package state_test
+
+import (
+	"sync"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/stablekernel/crucible/state"
+)
+
+var fakeEpoch = time.Unix(0, 0)
+
+// fired reports whether ch holds a value, without blocking, and returns it.
+func fired(ch <-chan time.Time) (time.Time, bool) {
+	select {
+	case v := <-ch:
+		return v, true
+	default:
+		return time.Time{}, false
+	}
+}
+
+// TestFakeClock_AfterWaitsForAdvance checks that After fires only once the clock
+// reaches its deadline: not with no advance, not one second short, and then
+// exactly once, with the deadline as its value.
+func TestFakeClock_AfterWaitsForAdvance(t *testing.T) {
+	clk := state.NewFakeClock(fakeEpoch)
+	ch := clk.After(5 * time.Second)
+
+	if _, ok := fired(ch); ok {
+		t.Fatal("After(5s) fired with no advance")
+	}
+	clk.Advance(4 * time.Second)
+	if _, ok := fired(ch); ok {
+		t.Fatal("After(5s) fired after Advance(4s)")
+	}
+	clk.Advance(time.Second)
+	got, ok := fired(ch)
+	if !ok {
+		t.Fatal("After(5s) did not fire once the clock reached 5s")
+	}
+	if want := fakeEpoch.Add(5 * time.Second); !got.Equal(want) {
+		t.Fatalf("After value: want %v, got %v", want, got)
+	}
+	clk.Advance(time.Hour)
+	if _, ok := fired(ch); ok {
+		t.Fatal("After fired a second time")
+	}
+}
+
+// TestFakeClock_AfterSingleAdvance checks that one Advance of exactly the
+// duration fires the channel.
+func TestFakeClock_AfterSingleAdvance(t *testing.T) {
+	clk := state.NewFakeClock(fakeEpoch)
+	ch := clk.After(5 * time.Second)
+	clk.Advance(5 * time.Second)
+	if _, ok := fired(ch); !ok {
+		t.Fatal("After(5s) did not fire after Advance(5s)")
+	}
+}
+
+// TestFakeClock_AfterNonPositiveFiresAtOnce checks that After(0) and a negative
+// duration fire immediately with the current instant, as time.After does.
+func TestFakeClock_AfterNonPositiveFiresAtOnce(t *testing.T) {
+	clk := state.NewFakeClock(fakeEpoch)
+	clk.Advance(3 * time.Second)
+	for _, d := range []time.Duration{0, -time.Second} {
+		got, ok := fired(clk.After(d))
+		if !ok {
+			t.Fatalf("After(%v) did not fire at once", d)
+		}
+		if want := fakeEpoch.Add(3 * time.Second); !got.Equal(want) {
+			t.Fatalf("After(%v) value: want %v, got %v", d, want, got)
+		}
+	}
+}
+
+// TestFakeClock_AdvanceFiresEveryReachedWaiter checks that one large Advance
+// fires every waiter it reaches, each with its own deadline, and none past it.
+func TestFakeClock_AdvanceFiresEveryReachedWaiter(t *testing.T) {
+	clk := state.NewFakeClock(fakeEpoch)
+	waiters := []struct {
+		d  time.Duration
+		ch <-chan time.Time
+	}{
+		{9 * time.Second, clk.After(9 * time.Second)},
+		{2 * time.Second, clk.After(2 * time.Second)},
+		{5 * time.Second, clk.After(5 * time.Second)},
+		{5 * time.Second, clk.After(5 * time.Second)},
+	}
+	never := clk.After(time.Minute)
+
+	clk.Advance(10 * time.Second)
+
+	for _, w := range waiters {
+		got, ok := fired(w.ch)
+		if !ok {
+			t.Fatalf("After(%v) did not fire after Advance(10s)", w.d)
+		}
+		if want := fakeEpoch.Add(w.d); !got.Equal(want) {
+			t.Fatalf("After(%v) value: want %v, got %v", w.d, want, got)
+		}
+	}
+	if _, ok := fired(never); ok {
+		t.Fatal("After(1m) fired after Advance(10s)")
+	}
+	clk.Advance(50 * time.Second)
+	if _, ok := fired(never); !ok {
+		t.Fatal("After(1m) did not fire once the clock reached 1m")
+	}
+}
+
+// TestFakeClock_AdvanceFiresInDeadlineOrder checks the order one Advance fires
+// waiters in, as a goroutine already blocked on them sees it: the earliest
+// deadline first, and the first armed when deadlines tie. synctest.Wait makes
+// sure the receiver is blocked in its select before the clock moves.
+func TestFakeClock_AdvanceFiresInDeadlineOrder(t *testing.T) {
+	cases := []struct {
+		name         string
+		older, newer time.Duration // older is armed first
+		wantOlder    bool
+	}{
+		{"earlier deadline armed second wins", 5 * time.Second, 2 * time.Second, false},
+		{"earlier deadline armed first wins", 2 * time.Second, 5 * time.Second, true},
+		{"tie goes to the first armed", 5 * time.Second, 5 * time.Second, true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				clk := state.NewFakeClock(fakeEpoch)
+				older := clk.After(tc.older)
+				newer := clk.After(tc.newer)
+				gotOlder := make(chan bool, 1)
+				go func() {
+					select {
+					case <-older:
+						gotOlder <- true
+					case <-newer:
+						gotOlder <- false
+					}
+				}()
+				synctest.Wait()
+				clk.Advance(10 * time.Second)
+				if got := <-gotOlder; got != tc.wantOlder {
+					t.Fatalf("receiver woke on the %s waiter, want the %s", armed(got), armed(tc.wantOlder))
+				}
+			})
+		})
+	}
+}
+
+// armed names a waiter by when it was armed, for failure messages.
+func armed(older bool) string {
+	if older {
+		return "first-armed"
+	}
+	return "second-armed"
+}
+
+// TestFakeClock_ConcurrentAfterAndAdvance arms waiters from many goroutines
+// while others advance the clock, for the race detector. Every waiter armed must
+// fire once the clock has moved past all deadlines.
+func TestFakeClock_ConcurrentAfterAndAdvance(t *testing.T) {
+	const arms, advances = 64, 16
+	clk := state.NewFakeClock(fakeEpoch)
+
+	chans := make([]<-chan time.Time, arms)
+	var wg sync.WaitGroup
+	for i := range arms {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			chans[i] = clk.After(time.Duration(i+1) * time.Millisecond)
+		}()
+	}
+	for range advances {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			clk.Advance(time.Millisecond)
+			_ = clk.Now()
+		}()
+	}
+	wg.Wait()
+
+	// The last waiter armed may have started as late as the final advance, so
+	// move past the longest deadline from there.
+	clk.Advance(time.Duration(arms+1) * time.Millisecond)
+	for i, ch := range chans {
+		if _, ok := fired(ch); !ok {
+			t.Fatalf("waiter %d did not fire", i)
+		}
+	}
+}