Start no prefix flush after the stop deadline (closes #188)

When a write ran past the stop deadline, a full batch's wake-up or a tick
could wait beside Stop's context. The flush goroutine's select picked one
of them at random, so about half the time it wrote that batch after the
deadline. That was the intermittent failure of
TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline. The flush
goroutine now takes a waiting stop before anything else. A new test hands
over a full batch while a write is held; it fails about half its runs on
the old code.

Model: opus-5-5

internal/routewatch/batchwrite_internal_test.go

@@ -317,9 +317,7 @@ func stopWhileAWriteIsBlocked(
 
 // TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline also checks the
 // routing table's changes: the flush whose delete is held wrote those of the
-// fill, so Stop must log the one of the extra prefix as not written. The flush
+// fill, so Stop must log the one of the extra prefix as not written.
-// timer must not fire. A flush it started before the fill would be the one
-// held, and it would not have written the fill's change.
 func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	t.Parallel()
 
@@ -328,8 +326,8 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	var logs bytes.Buffer
 
 	store := newBlockingStore()
-	h := newPrefixHandler(store, newErrorLogger(&logs),
+	h := NewPrefixHandler(store, newErrorLogger(&logs),
-		routetable.New(routetable.Params{}), longerThanAnyTest)
+		routetable.New(routetable.Params{}))
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	// The peer's session goes down first, so the write held is the delete of
 	// its routes, and Stop must log that peer down too.

internal/routewatch/prefixhandler.go

@@ -130,16 +130,6 @@ func (u prefixUpdate) announcesRoute() bool {
 // the updates to table, the routing table read from db at startup.
 func NewPrefixHandler(
 	db database.Store, logger *logger.Logger, table *routetable.Table,
-) *PrefixHandler {
-	return newPrefixHandler(db, logger, table, prefixBatchTimeout)
-}
-
-// newPrefixHandler is NewPrefixHandler with the flush timer's interval,
-// batchTimeout, in place of prefixBatchTimeout. A test that the timer must not
-// disturb gives an interval longer than the test runs.
-func newPrefixHandler(
-	db database.Store, logger *logger.Logger, table *routetable.Table,
-	batchTimeout time.Duration,
 ) *PrefixHandler {
 	h := &PrefixHandler{
 		db:          db,
@@ -154,7 +144,7 @@ func newPrefixHandler(
 	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
-	go h.flushLoop(batchTimeout)
+	go h.flushLoop()
 
 	return h
 }
@@ -325,11 +315,11 @@ func (h *PrefixHandler) peerSessionDown(peer string) {
 }
 
 // flushLoop runs in a goroutine, the only one that writes the batch: it
-// flushes when the batch is full, every batchTimeout, and once more on Stop.
+// flushes when the batch is full, periodically, and once more on Stop.
-func (h *PrefixHandler) flushLoop(batchTimeout time.Duration) {
+func (h *PrefixHandler) flushLoop() {
 	defer close(h.done)
 
-	ticker := time.NewTicker(batchTimeout)
+	ticker := time.NewTicker(prefixBatchTimeout)
 	defer ticker.Stop()
 
 	for {
@@ -350,7 +340,7 @@ func (h *PrefixHandler) flushLoop(batchTimeout time.Duration) {
 
 		select {
 		case <-ticker.C:
-			if time.Since(h.lastFlush) >= batchTimeout {
+			if time.Since(h.lastFlush) >= prefixBatchTimeout {
 				h.flushBatch()
 			}
 		case <-h.batchFullCh:

internal/routewatch/prefixhandler_internal_test.go

@@ -980,11 +980,6 @@ func TestPrefixHandlerStopLogsNothingAfterADeleteThatSucceeds(t *testing.T) {
 	}
 }
 
-// longerThanAnyTest is a flush timer interval that no test lasts. A test whose
-// prefix handler must flush only when its batch is full and on Stop gives it to
-// newPrefixHandler.
-const longerThanAnyTest = time.Hour
-
 // prefixHandlerWithoutTimer returns a prefix handler that writes to store and
 // applies the updates to an empty routing table. It is built without
 // NewPrefixHandler's flush goroutine, so the test flushes itself.

internal/routewatch/prefixhandler_table_internal_test.go

@@ -341,14 +341,15 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
 }
 
 // TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten stops a prefix handler,
-// with no deadline, whose routing table holds more changed pairs than one write
+// with no deadline, whose routing table holds more changed pairs than two
-// takes. Its last flush writes one write's worth, so Stop must log the pairs
+// writes take. Its last flush writes one write's worth, so Stop must log the
-// that no write took. The flush timer must not fire, or its writes could leave
+// pairs that no write took.
-// no pairs for Stop to log.
 func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
 	t.Parallel()
 
-	const pairs = maxTableChangesPerWrite + 1
+	// More than two writes take, so that pairs are left even when the flush
+	// timer writes once before Stop.
+	const pairs = 2*maxTableChangesPerWrite + 1
 
 	var logs bytes.Buffer
 
@@ -356,7 +357,7 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
 	table := routetable.New(routetable.Params{})
 	announceManyPrefixes(table, pairs)
 
-	h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest)
+	h := NewPrefixHandler(store, newErrorLogger(&logs), table)
 	h.Stop(t.Context())
 
 	written := 0