From cf3231d5b059e33aea2aae8af91b6390b4e5ae0a Mon Sep 17 00:00:00 2001 From: Jordan Coin Jackson Date: Sun, 11 Oct 2026 23:04:33 +0200 Subject: [PATCH] test(watch): burst detector takes its clock from the caller; storm test asserts exact counts on a synthetic clock TestCheckoutStormCoalescesIntoOneBatch fed 500 events in 200 ms against 250 ms wall-clock windows and asserted at most 17 git calls; on a loaded CI runner the storm split into two bursts and 27 were observed (#219's ubuntu 1.26 run). The detection logic moves out of the event loop into burstDetector, whose observe, enqueue and take all take `now`, so the loop passes time.Now() and tests pass a fixed clock. The exact counts (16 per-file, then one batch of 292 files; git, topology and refresh each exactly once) are asserted on the synthetic clock, and the one test that still goes through the real loop asserts only that batching happened, with the bound derived from burstThreshold. Refs #201 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01NRaBgKRAZMX9hjbkkefA7T --- watch/burst.go | 67 ++++++++++++ watch/burst_test.go | 245 ++++++++++++++++++++++++++++++-------------- watch/events.go | 58 ++--------- 3 files changed, 248 insertions(+), 122 deletions(-) create mode 100644 watch/burst.go diff --git a/watch/burst.go b/watch/burst.go new file mode 100644 index 0000000..50716b3 --- /dev/null +++ b/watch/burst.go @@ -0,0 +1,67 @@ +package watch + +import ( + "time" + + "github.com/fsnotify/fsnotify" +) + +// burstDetector decides when the event loop is inside a checkout storm and +// holds the storm's events until it settles. It never reads the wall clock: +// every method takes `now`, so the loop passes time.Now() and tests pass a +// synthetic clock and get the same decisions every run (#201). +type burstDetector struct { + arrivals []time.Time // source-event times inside burstDetectWindow + queue []fsnotify.Event // the storm's events, first-seen order, last op per path + queued map[string]int // path -> index into queue + start time.Time // when the current burst opened +} + +// observe records one source-event arrival and reports whether the loop is +// in a burst: either one is already open, or more than burstThreshold +// arrivals fall inside burstDetectWindow ending now. +func (b *burstDetector) observe(now time.Time) bool { + if b.queued != nil { + return true + } + cut := 0 + for cut < len(b.arrivals) && now.Sub(b.arrivals[cut]) > burstDetectWindow { + cut++ + } + b.arrivals = append(b.arrivals[cut:], now) + return len(b.arrivals) > burstThreshold +} + +// enqueue adds an event to the open burst, keeping the last op per path, and +// returns how long the loop should wait before flushing: burstSettle of +// quiet, clipped so the burst never stays open past burstCeiling. +func (b *burstDetector) enqueue(event fsnotify.Event, now time.Time) time.Duration { + if b.queued == nil { + b.queued = make(map[string]int) + b.start = now + } + if i, ok := b.queued[event.Name]; ok { + b.queue[i] = event + } else { + b.queued[event.Name] = len(b.queue) + b.queue = append(b.queue, event) + } + if remaining := burstCeiling - now.Sub(b.start); remaining < burstSettle { + return max(remaining, 0) + } + return burstSettle +} + +// take closes the burst and returns its events; the next observe starts a +// fresh detection window. +func (b *burstDetector) take() []fsnotify.Event { + events := b.queue + b.queue, b.queued = nil, nil + b.arrivals = b.arrivals[:0] + return events +} + +// pending reports whether a burst is open with events waiting. +func (b *burstDetector) pending() bool { + return len(b.queue) > 0 +} diff --git a/watch/burst_test.go b/watch/burst_test.go index 1625e32..0fcb490 100644 --- a/watch/burst_test.go +++ b/watch/burst_test.go @@ -11,24 +11,93 @@ import ( "github.com/fsnotify/fsnotify" ) -// A rebase or branch switch writes every touched file through a temp file and -// a rename, so a 300-file checkout used to cost 300 git subprocesses, 300 -// state writes and 300 topology recomputes on the event loop (#201). The -// storm must coalesce into one batch. -func TestCheckoutStormCoalescesIntoOneBatch(t *testing.T) { - root := t.TempDir() - if err := os.Mkdir(filepath.Join(root, ".git"), 0o755); err != nil { - t.Fatal(err) - } +// stormEvents is a 300-file checkout the way git writes one: every file +// CREATEd, the first 200 also WRITTEN, 500 events in all. +func stormEvents(t *testing.T, root string) []fsnotify.Event { + t.Helper() const files = 300 - paths := make([]string, files) - for i := range paths { - paths[i] = filepath.Join(root, fmt.Sprintf("f%03d.go", i)) - if err := os.WriteFile(paths[i], []byte("package x\n"), 0o644); err != nil { + events := make([]fsnotify.Event, 0, 500) + for i := 0; i < files; i++ { + path := filepath.Join(root, fmt.Sprintf("f%03d.go", i)) + if err := os.WriteFile(path, []byte("package x\n"), 0o644); err != nil { t.Fatal(err) } + events = append(events, fsnotify.Event{Name: path, Op: fsnotify.Create}) + if i < 200 { + events = append(events, fsnotify.Event{Name: path, Op: fsnotify.Write}) + } } + return events +} + +// The detector takes its clock from the caller, so the same 500 arrivals +// 0.4 ms apart make the same decisions every run: the first burstThreshold +// are handled one by one, everything after is queued, and one settle +// closes the burst with one event per path. +func TestBurstDetectorOpensOnceForAStorm(t *testing.T) { + events := stormEvents(t, t.TempDir()) + var b burstDetector + clock := time.Unix(1_700_000_000, 0) + const gap = 400 * time.Microsecond // 500 events in 200 ms + perFile := 0 + for _, event := range events { + if !b.observe(clock) { + perFile++ + } else { + if delay := b.enqueue(event, clock); delay != burstSettle { + t.Fatalf("enqueue delay = %v, want burstSettle %v", delay, burstSettle) + } + } + clock = clock.Add(gap) + } + if perFile != burstThreshold { + t.Fatalf("events handled before the burst opened = %d, want burstThreshold %d", perFile, burstThreshold) + } + batch := b.take() + // The first 16 events are f000..f007 (CREATE+WRITE each); the batch holds + // the other 292 files once each, last op wins. + if len(batch) != 300-burstThreshold/2 { + t.Fatalf("batch = %d events, want %d (one per remaining file)", len(batch), 300-burstThreshold/2) + } + if got := batch[0]; filepath.Base(got.Name) != "f008.go" || got.Op != fsnotify.Write { + t.Fatalf("batch[0] = %s %v, want f008.go WRITE (last op for the path)", filepath.Base(got.Name), got.Op) + } + if b.pending() { + t.Fatal("take must close the burst") + } + // After the burst, a lone event starts a fresh window and is not batched. + if b.observe(clock.Add(time.Second)) { + t.Fatal("a single event after a flush must not open a burst") + } +} + +// The ceiling clips the settle delay so a storm that never goes quiet still +// flushes after burstCeiling. +func TestBurstDetectorCeilingClipsSettle(t *testing.T) { + var b burstDetector + clock := time.Unix(1_700_000_000, 0) + for i := 0; i <= burstThreshold; i++ { + b.observe(clock) + } + event := fsnotify.Event{Name: "a.go", Op: fsnotify.Create} + if delay := b.enqueue(event, clock); delay != burstSettle { + t.Fatalf("first enqueue delay = %v, want %v", delay, burstSettle) + } + late := clock.Add(burstCeiling - 100*time.Millisecond) + if delay := b.enqueue(event, late); delay != 100*time.Millisecond { + t.Fatalf("delay near the ceiling = %v, want 100ms", delay) + } + if delay := b.enqueue(event, clock.Add(burstCeiling+time.Second)); delay != 0 { + t.Fatalf("delay past the ceiling = %v, want 0", delay) + } +} + +func newBurstTestDaemon(t *testing.T, root string) *Daemon { + t.Helper() + if err := os.Mkdir(filepath.Join(root, ".git"), 0o755); err != nil { + t.Fatal(err) + } d, err := NewDaemon(root, false) if err != nil { t.Fatal(err) @@ -41,69 +110,70 @@ func TestCheckoutStormCoalescesIntoOneBatch(t *testing.T) { if err := os.MkdirAll(d.runtimeDir, 0o755); err != nil { t.Fatal(err) } - var gitCalls, topology, refreshes atomic.Int32 + return d +} + +type burstCounters struct { + git, topology, refreshes atomic.Int32 +} + +// stubBurstSeams counts the shared costs. The daemon must be stopped before +// the seams are restored, so callers register Stop after this. +func stubBurstSeams(t *testing.T) *burstCounters { + t.Helper() + c := &burstCounters{} origDirty, origDirtyPaths := isFileDirtyFn, dirtyPathsFn origTopology, origRefresh := daemonComputeTopology, daemonRefreshDependencies t.Cleanup(func() { isFileDirtyFn, dirtyPathsFn = origDirty, origDirtyPaths daemonComputeTopology, daemonRefreshDependencies = origTopology, origRefresh }) - // Registered after the seam restore so it runs first: the loop must be - // stopped before the seams change under it. - t.Cleanup(func() { d.Stop() }) - isFileDirtyFn = func(string, string) bool { gitCalls.Add(1); return false } + isFileDirtyFn = func(string, string) bool { c.git.Add(1); return false } dirtyPathsFn = func(_ string, rel []string) map[string]bool { - gitCalls.Add(1) + c.git.Add(1) + if len(rel) == 0 { + return nil + } return map[string]bool{rel[0]: true} } - daemonComputeTopology = func(*Daemon) { topology.Add(1) } - daemonRefreshDependencies = func(*Daemon) { refreshes.Add(1) } + daemonComputeTopology = func(*Daemon) { c.topology.Add(1) } + daemonRefreshDependencies = func(*Daemon) { c.refreshes.Add(1) } + return c +} - d.eventLoopWG.Add(1) - go func() { - defer d.eventLoopWG.Done() - d.eventLoop() - }() +// A flushed batch pays each shared cost exactly once, however many files it +// holds, and still records one event per file with the batched dirty flag. +func TestHandleBurstPaysEachSharedCostOnce(t *testing.T) { + root := t.TempDir() + events := stormEvents(t, root) + d := newBurstTestDaemon(t, root) + c := stubBurstSeams(t) + t.Cleanup(func() { d.Stop() }) - // 500 events for 300 files inside 200 ms: every file CREATEd, the first - // 200 also RENAMEd into place, the way git writes a checkout. - start := time.Now() - for i := range paths { - d.watcher.Events <- fsnotify.Event{Name: paths[i], Op: fsnotify.Create} - if i < 200 { - d.watcher.Events <- fsnotify.Event{Name: paths[i], Op: fsnotify.Write} - } - } - if elapsed := time.Since(start); elapsed > 200*time.Millisecond { - t.Fatalf("feeding the storm took %v, want under 200ms for the test to mean anything", elapsed) + var b burstDetector + clock := time.Unix(1_700_000_000, 0) + for _, event := range events { + b.observe(clock) + b.enqueue(event, clock) + clock = clock.Add(400 * time.Microsecond) } + d.handleBurst(b.take()) - waitForWatchCondition(t, 5*time.Second, func() bool { - return len(d.GetEvents(files+1)) >= files - }) - // Let any trailing per-file work surface before counting. - time.Sleep(burstSettle + 50*time.Millisecond) - - // The first burstThreshold events are handled one by one before the storm - // is recognised; everything after that is one batch. - const leading = burstThreshold - if got := gitCalls.Load(); got > leading+1 { - t.Fatalf("git subprocesses = %d, want at most %d (per-file before detection + one batch)", got, leading+1) + if got := c.git.Load(); got != 1 { + t.Fatalf("git subprocesses = %d, want exactly 1", got) } - if got := topology.Load(); got > leading+1 { - t.Fatalf("topology recomputes = %d, want at most %d", got, leading+1) + if got := c.topology.Load(); got != 1 { + t.Fatalf("topology recomputes = %d, want exactly 1", got) } - if got := refreshes.Load(); got > leading+1 { - t.Fatalf("dependency refreshes = %d, want at most %d", got, leading+1) + if got := c.refreshes.Load(); got != 1 { + t.Fatalf("dependency refreshes = %d, want exactly 1", got) } - // Files handled before detection may log both their CREATE and WRITE; - // every file in the batch logs exactly once, last op wins. - events := d.GetEvents(files + leading + 1) - if len(events) < files || len(events) > files+leading { - t.Fatalf("events recorded = %d, want between %d and %d", len(events), files, files+leading) + recorded := d.GetEvents(1000) + if len(recorded) != 300 { + t.Fatalf("events recorded = %d, want 300 (one per file)", len(recorded)) } dirty := 0 - for _, event := range events { + for _, event := range recorded { if event.Dirty { dirty++ } @@ -113,29 +183,54 @@ func TestCheckoutStormCoalescesIntoOneBatch(t *testing.T) { } } -// Below the threshold nothing changes: a handful of events are still handled -// one at a time with their own git call each. -func TestFewEventsAreNotBatched(t *testing.T) { +// End to end through the real event loop on the wall clock. CI load can +// split one storm into several bursts, so this asserts only that batching +// happened: per-file costs stay far below one per file. Exact counts live in +// the synthetic-clock tests above. +func TestCheckoutStormIsBatchedThroughTheEventLoop(t *testing.T) { root := t.TempDir() - if err := os.Mkdir(filepath.Join(root, ".git"), 0o755); err != nil { - t.Fatal(err) + events := stormEvents(t, root) + d := newBurstTestDaemon(t, root) + c := stubBurstSeams(t) + t.Cleanup(func() { d.Stop() }) + + d.eventLoopWG.Add(1) + go func() { + defer d.eventLoopWG.Done() + d.eventLoop() + }() + for _, event := range events { + d.watcher.Events <- event } - path := filepath.Join(root, "a.go") - if err := os.WriteFile(path, []byte("package x\n"), 0o644); err != nil { - t.Fatal(err) + waitForWatchCondition(t, 10*time.Second, func() bool { + return len(d.GetEvents(1000)) >= 300 + }) + time.Sleep(burstSettle + 50*time.Millisecond) + + // Each burst costs at most burstThreshold per-file calls before it is + // recognised plus one batched call; allow a slow runner to split the + // storm four ways. Unbatched, the count would be 500. + const bound = 4 * (burstThreshold + 1) + if got := c.git.Load(); got > bound { + t.Fatalf("git subprocesses = %d, want at most %d", got, bound) } - d, err := NewDaemon(root, false) - if err != nil { - t.Fatal(err) + if got := c.topology.Load(); got > bound { + t.Fatalf("topology recomputes = %d, want at most %d", got, bound) } - if err := d.watcher.Close(); err != nil { - t.Fatal(err) + if got := c.refreshes.Load(); got > bound { + t.Fatalf("dependency refreshes = %d, want at most %d", got, bound) } - d.watcher.Events = make(chan fsnotify.Event, 8) - d.watcher.Errors = make(chan error) - if err := os.MkdirAll(d.runtimeDir, 0o755); err != nil { +} + +// Below the threshold nothing changes: a lone event is handled at once with +// its own git call. +func TestFewEventsAreNotBatched(t *testing.T) { + root := t.TempDir() + path := filepath.Join(root, "a.go") + if err := os.WriteFile(path, []byte("package x\n"), 0o644); err != nil { t.Fatal(err) } + d := newBurstTestDaemon(t, root) var perFile, batched atomic.Int32 origDirty, origDirtyPaths := isFileDirtyFn, dirtyPathsFn origTopology, origRefresh := daemonComputeTopology, daemonRefreshDependencies @@ -143,11 +238,11 @@ func TestFewEventsAreNotBatched(t *testing.T) { isFileDirtyFn, dirtyPathsFn = origDirty, origDirtyPaths daemonComputeTopology, daemonRefreshDependencies = origTopology, origRefresh }) - t.Cleanup(func() { d.Stop() }) isFileDirtyFn = func(string, string) bool { perFile.Add(1); return false } dirtyPathsFn = func(string, []string) map[string]bool { batched.Add(1); return nil } daemonComputeTopology = func(*Daemon) {} daemonRefreshDependencies = func(*Daemon) {} + t.Cleanup(func() { d.Stop() }) d.eventLoopWG.Add(1) go func() { diff --git a/watch/events.go b/watch/events.go index 5470a9b..4a9821f 100644 --- a/watch/events.go +++ b/watch/events.go @@ -292,19 +292,14 @@ func (d *Daemon) eventLoop() { } }() - // Burst state: arrivals is the recent source-event timestamps used to - // detect a storm; queue holds the storm's events, last op per path, in - // first-seen order, until the burst settles or hits its ceiling. - var ( - arrivals []time.Time - queue []fsnotify.Event - queued map[string]int - burstStart time.Time - burstTimer = time.NewTimer(time.Hour) - burstC <-chan time.Time - ) + // A checkout storm is detected and queued by burstDetector, which takes + // the clock from this loop so tests can drive it deterministically. The + // loop owns only the timer that fires the flush. + var burst burstDetector + burstTimer := time.NewTimer(time.Hour) burstTimer.Stop() defer burstTimer.Stop() + var burstC <-chan time.Time armBurst := func(delay time.Duration) { if !burstTimer.Stop() { select { @@ -315,44 +310,13 @@ func (d *Daemon) eventLoop() { burstTimer.Reset(delay) burstC = burstTimer.C } - enqueue := func(event fsnotify.Event, now time.Time) { - if queued == nil { - queued = make(map[string]int) - burstStart = now - } - if i, ok := queued[event.Name]; ok { - queue[i] = event - } else { - queued[event.Name] = len(queue) - queue = append(queue, event) - } - if remaining := burstCeiling - now.Sub(burstStart); remaining < burstSettle { - armBurst(max(remaining, 0)) - } else { - armBurst(burstSettle) - } - } flushBurst := func() { burstC = nil - events := queue - queue, queued = nil, nil - arrivals = arrivals[:0] - d.handleBurst(events) - } - inBurst := func(now time.Time) bool { - if queued != nil { - return true - } - cut := 0 - for cut < len(arrivals) && now.Sub(arrivals[cut]) > burstDetectWindow { - cut++ - } - arrivals = append(arrivals[cut:], now) - return len(arrivals) > burstThreshold + d.handleBurst(burst.take()) } defer func() { - if len(queue) > 0 { - d.handleBurst(queue) + if burst.pending() { + d.handleBurst(burst.take()) } }() @@ -432,9 +396,9 @@ func (d *Daemon) eventLoop() { continue } - if inBurst(now) { + if burst.observe(now) { debouncer.cancelPending(event.Name) - enqueue(event, now) + armBurst(burst.enqueue(event, now)) continue }