From 2c304a00223dd2b7c26a755464dc93379c57e06d Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Tue, 27 Oct 2026 15:23:52 +0100 Subject: [PATCH] Make a timed batch write on every tick (closes #192) The prefix, AS and peer handlers skipped a timed write unless one interval had passed since the last write ended. The next tick came a little less than one interval after that, so every other tick was skipped. Each tick now writes. A full batch's write restarts the timer, so the next timed write comes one interval after that write began. Comparing clock readings would not do: two ticks one interval apart can read a hair under it. flushBatch no longer reports whether it wrote a batch, since nothing reads that. A new test runs the loop on a fake clock and fails on the old code. Model: opus-5-5 --- TODO.md | 4 ++ internal/routewatch/ashandler.go | 8 ++- internal/routewatch/batchwrite.go | 28 +++++----- .../routewatch/batchwrite_internal_test.go | 53 +++++++++++++++++++ internal/routewatch/peerhandler.go | 8 ++- internal/routewatch/prefixhandler.go | 7 +-- 6 files changed, 78 insertions(+), 30 deletions(-) diff --git a/TODO.md b/TODO.md index 59637e2..f0538d9 100644 --- a/TODO.md +++ b/TODO.md @@ -33,6 +33,10 @@ tested. # Completed Steps +- 2026-10-27: the prefix, AS and peer handlers make a timed write on every tick + of their timers, every 1, 2 and 2 seconds. They used to skip every other tick. + A full batch's write restarts the timer, so the next timed write comes one + interval after that write began (closes #192) - 2026-10-27: once the stop deadline has passed, the AS, peer and peering handlers start no write either. A full batch or a tick of the timer that waited while a write ran past the deadline could still start one. The flush diff --git a/internal/routewatch/ashandler.go b/internal/routewatch/ashandler.go index cee6699..5370bb0 100644 --- a/internal/routewatch/ashandler.go +++ b/internal/routewatch/ashandler.go @@ -202,11 +202,11 @@ func (h *ASHandler) takeBatch() []asnOp { } // flushBatch takes the pending batch and writes it to the database, without -// holding the mutex during the write. It reports whether there was a batch. -func (h *ASHandler) flushBatch() bool { +// holding the mutex during the write. +func (h *ASHandler) flushBatch() { batch := h.takeBatch() if len(batch) == 0 { - return false + return } // The earliest and the latest time the batch saw each AS. The batch holds @@ -229,6 +229,4 @@ func (h *ASHandler) flushBatch() bool { h.mu.Lock() h.writing = 0 h.mu.Unlock() - - return true } diff --git a/internal/routewatch/batchwrite.go b/internal/routewatch/batchwrite.go index 53ef900..6388342 100644 --- a/internal/routewatch/batchwrite.go +++ b/internal/routewatch/batchwrite.go @@ -6,27 +6,26 @@ import ( ) // flushLoop is the flush goroutine of the prefix, AS and peer handlers. It is -// the only goroutine that writes a handler's batch. flushBatch takes the batch, -// writes it, and reports whether there was one. +// the only goroutine that writes a handler's batch. flushBatch takes the batch +// and writes it. // -// flushLoop calls flushBatch when batchFullCh wakes it. It also calls it every -// batchTimeout, unless it wrote a batch less than batchTimeout ago. When stopCh -// hands it Stop's context, it calls flushBatch once more, unless the stop -// deadline has passed. Then it closes done and returns. +// flushLoop calls flushBatch on each tick of a timer, every batchTimeout. It +// also calls it at once when batchFullCh wakes it. A full batch's write +// restarts the timer, so the next tick comes batchTimeout after that write +// began. When stopCh hands it Stop's context, it calls flushBatch once more, +// unless the stop deadline has passed. Then it closes done and returns. func flushLoop( batchTimeout time.Duration, batchFullCh <-chan struct{}, stopCh <-chan context.Context, done chan<- struct{}, - flushBatch func() bool, + flushBatch func(), ) { defer close(done) ticker := time.NewTicker(batchTimeout) defer ticker.Stop() - lastFlush := time.Now() - for { // Stop goes first. A write can run past the stop deadline while a full // batch's wake-up or a tick waits beside Stop's context. The select @@ -45,13 +44,12 @@ func flushLoop( select { case <-ticker.C: - if time.Since(lastFlush) >= batchTimeout && flushBatch() { - lastFlush = time.Now() - } + flushBatch() case <-batchFullCh: - if flushBatch() { - lastFlush = time.Now() - } + // Reset also drops a tick that waits in ticker.C, so no timed + // write follows this one at once. + ticker.Reset(batchTimeout) + flushBatch() case ctx := <-stopCh: // Final flush, unless the stop deadline has passed if ctx.Err() == nil { diff --git a/internal/routewatch/batchwrite_internal_test.go b/internal/routewatch/batchwrite_internal_test.go index f659a0f..3a09aa6 100644 --- a/internal/routewatch/batchwrite_internal_test.go +++ b/internal/routewatch/batchwrite_internal_test.go @@ -9,6 +9,7 @@ import ( "slices" "sync" "testing" + "testing/synctest" "time" "git.eeqj.de/sneak/routewatch/internal/database" @@ -27,6 +28,58 @@ const ( stillWaitingAfter = 100 * time.Millisecond ) +// TestFlushLoopWritesOnEveryTick runs flushLoop in a synctest bubble, whose +// clock is fake: time.Sleep there moves the clock on and returns at once. Each +// write takes a tenth of the interval. Two timed writes must start one interval +// apart. A full batch's write, which starts as the second timed write ends, +// must be followed by a timed write one interval after it began. +func TestFlushLoopWritesOnEveryTick(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + const ( + interval = time.Second + writeTime = 100 * time.Millisecond + ) + + writeStarts := make(chan time.Time) + batchFullCh := make(chan struct{}, 1) + stopCh := make(chan context.Context, 1) + done := make(chan struct{}) + + go func() { + flushLoop(interval, batchFullCh, stopCh, done, func() { + writeStarts <- time.Now() + + time.Sleep(writeTime) + }) + }() + + first := <-writeStarts + second := <-writeStarts + + if gap := second.Sub(first); gap != interval { + t.Errorf("the second timed write started %s after the first, "+ + "want %s", gap, interval) + } + + batchFullCh <- struct{}{} + + full := <-writeStarts + timed := <-writeStarts + + if gap := timed.Sub(full); gap != interval { + t.Errorf("the timed write after a full batch's started %s after "+ + "it, want %s", gap, interval) + } + + stopCh <- t.Context() + + <-writeStarts // Stop's last write + <-done + }) +} + // blockingStore is a mockStore whose batch writes for the AS, peer and prefix // handlers, delete of a peer's routes and peering write wait until the test // calls releaseWrites. writeStarted is closed when the first of them starts. It diff --git a/internal/routewatch/peerhandler.go b/internal/routewatch/peerhandler.go index 5800474..b15a58e 100644 --- a/internal/routewatch/peerhandler.go +++ b/internal/routewatch/peerhandler.go @@ -200,11 +200,11 @@ func (h *PeerHandler) takeBatch() []peerUpdate { } // flushBatch takes the pending batch and writes it to the database, without -// holding the mutex during the write. It reports whether there was a batch. -func (h *PeerHandler) flushBatch() bool { +// holding the mutex during the write. +func (h *PeerHandler) flushBatch() { batch := h.takeBatch() if len(batch) == 0 { - return false + return } // The earliest and the latest time the batch saw each peer, with the peer @@ -234,6 +234,4 @@ func (h *PeerHandler) flushBatch() bool { h.mu.Lock() h.writing = 0 h.mu.Unlock() - - return true } diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 6835699..5fa550f 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -352,13 +352,12 @@ func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) { // flushBatch writes the routing table's changes, then takes the pending batch // and writes it to the database, without holding the mutex during the writes. -// It reports whether there was a batch, or a peer down to delete the routes of. -func (h *PrefixHandler) flushBatch() bool { +func (h *PrefixHandler) flushBatch() { h.writeTableChanges() batch, peersDown := h.takeBatch() if len(batch) == 0 && len(peersDown) == 0 { - return false + return } startTime := time.Now() @@ -441,8 +440,6 @@ func (h *PrefixHandler) flushBatch() bool { "success", successCount, "duration_ms", elapsed.Milliseconds(), ) - - return true } // applyToTable applies update to the routing table in memory: as an