diff --git a/durable/clock_internal_test.go b/durable/clock_internal_test.go index a0a0124..208cae6 100644 --- a/durable/clock_internal_test.go +++ b/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") } } diff --git a/state/CHANGELOG.md b/state/CHANGELOG.md index e4894a8..9f136e2 100644 --- a/state/CHANGELOG.md +++ b/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 diff --git a/state/driver.go b/state/driver.go index 52c7eeb..76d8166 100644 --- a/state/driver.go +++ b/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. diff --git a/state/fakeclock_test.go b/state/fakeclock_test.go new file mode 100644 index 0000000..d8fed32 --- /dev/null +++ b/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) + } + } +}