Make the routewatch tests pass on a loaded host (closes #213)

Five prefix handler tests gave HandleMessage two seconds to apply a full
batch of 25,000 prefixes, which under the race detector on a loaded host
takes longer. A test now waits up to stepTimeout, 15 seconds, for each step
it expects.

The slowest tests filled production limits, such as the peering handler's
500,000 paths. The peering handler now takes its path limit, and the prefix
handler its limit on one write of the table's changes, as they take their
timer interval. Those tests give small limits, and the tests of the maximum
batch size lower it to one batch. This also fixes
#215.

The test helpers bound every wait, Stop included, so a test that hangs in
one fails with its own message.

Model: opus-5-5

TODO.md

@@ -38,9 +38,9 @@ tested.
 - 2026-10-27: the tests of `internal/routewatch` no longer fail on a loaded
   host. Five of them gave `HandleMessage` two seconds to apply a full prefix
   batch, which under the race detector can take longer. A test now waits up to
-  15 seconds for each step it expects. The tests of the peering handler's path
+  15 seconds for each step it expects. The tests of the handlers' limits give
-  limit and of the prefix handler's limits give the handler small limits, so the
+  the handlers small ones, so the package no longer runs past its 90-second
-  package no longer runs past its 90-second limit (closes #213, #215)
+  limit (closes #213, #215)
 - 2026-10-27: the server works out the home page's cards every minute until it
   stops. It used to work them out only once, at startup: the end of the context
   that `Start` is given stopped the refreshes (closes #211)

internal/routewatch/ashandler.go

@@ -40,9 +40,13 @@ type ASHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
+	// batchMaxSize is asnBatchMaxSize. Only HandleMessage reads it, so a test
+	// may lower it before it hands the handler a message.
+	batchMaxSize int
+
 	// Batching. mu guards batch and writing only: the flush goroutine takes
 	// the batch under mu and writes it after releasing mu, so HandleMessage
-	// waits for a database write only while batch is at asnBatchMaxSize.
+	// waits for a database write only while batch is at batchMaxSize.
 	mu    sync.Mutex
 	batch []asnOp
 	// writing is how many operations the flush goroutine took from batch
@@ -79,12 +83,13 @@ func newASHandler(
 	db database.Store, logger *logger.Logger, batchTimeout time.Duration,
 ) *ASHandler {
 	h := &ASHandler{
-		db:          db,
+		db:           db,
-		logger:      logger,
+		logger:       logger,
-		batch:       make([]asnOp, 0, asnBatchSize),
+		batchMaxSize: asnBatchMaxSize,
-		batchFullCh: make(chan struct{}, 1),
+		batch:        make([]asnOp, 0, asnBatchSize),
-		stopCh:      make(chan context.Context, 1),
+		batchFullCh:  make(chan struct{}, 1),
-		done:        make(chan struct{}),
+		stopCh:       make(chan context.Context, 1),
+		done:         make(chan struct{}),
 	}
 	h.batchTaken.L = &h.mu
 
@@ -127,7 +132,7 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// While the batch is at its maximum size, wait until the flush goroutine
 	// takes it. Meanwhile the handler takes nothing off its queue, which
 	// fills, and the streamer drops messages for it.
-	for len(h.batch) >= asnBatchMaxSize {
+	for len(h.batch) >= h.batchMaxSize {
 		h.batchTaken.Wait()
 	}
 
@@ -151,7 +156,7 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 
 	// A full batch wakes the flush goroutine. If it is writing, the wake-up
 	// waits in batchFullCh and the batch keeps growing until the write ends
-	// or the batch reaches asnBatchMaxSize.
+	// or the batch reaches batchMaxSize.
 	if len(h.batch) >= asnBatchSize {
 		select {
 		case h.batchFullCh <- struct{}{}:

internal/routewatch/ashandler_internal_test.go

@@ -19,7 +19,7 @@ func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
 	db := newTestDatabase(t)
 
 	// Built without NewASHandler's flush goroutine, so the test flushes itself.
-	h := &ASHandler{db: db, logger: logger.New()}
+	h := &ASHandler{db: db, logger: logger.New(), batchMaxSize: asnBatchMaxSize}
 
 	first := peerAnnouncement(testPeerIP, testV4Prefix)
 	second := peerAnnouncement(testPeerIP, testV4Prefix)
@@ -54,7 +54,7 @@ func TestASHandlerRecordsNothingFromAPathWithASZero(t *testing.T) {
 	store := newMockStore()
 
 	// Built without NewASHandler's flush goroutine, so the test flushes itself.
-	h := &ASHandler{db: store, logger: logger.New()}
+	h := &ASHandler{db: store, logger: logger.New(), batchMaxSize: asnBatchMaxSize}
 
 	msg := peerAnnouncement(testPeerIP, testV4Prefix)
 	msg.Path = decodePath(t, testASNA, 0, testASNB)

internal/routewatch/asnfetcher_internal_test.go

@@ -246,16 +246,16 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
 }
 
 // TestRecordKeepsOnlyTheStatsWindow records a success and a failure every
-// second, by a clock the test sets, far longer than statsWindow, and never
+// second, by a clock the test sets, for three times statsWindow, and never
 // asks for the stats. The fetcher must keep only the times within the last
 // statsWindow.
 func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) {
 	t.Parallel()
 
 	const (
-		records = 100_000
 		step    = time.Second
 		want    = int(statsWindow / step)
+		records = 3 * want
 	)
 
 	clock := time.Now()

internal/routewatch/batchwrite_internal_test.go

@@ -292,7 +292,9 @@ func handleWhileAWriteIsBlocked(
 // while store holds the write, hands h the messages in toMax, which bring its
 // next batch to its maximum size, and the message extra. HandleMessage for
 // extra must wait until the test lets the write finish, and return after. Then
-// it stops h, which writes what it holds.
+// it stops h, which writes what it holds. The tests lower h's maximum batch
+// size to one batch, so toMax is one batch: the real maximum of ten batches is
+// slow to reach under the race detector.
 func handleAtMaxBatchSize(
 	t *testing.T,
 	store *blockingStore,
@@ -407,7 +409,7 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	store := newBlockingStore()
 	h := newPrefixHandler(store, newErrorLogger(&logs),
 		routetable.New(routetable.Params{}), longerThanAnyTest,
-		prefixBatchMaxSize, maxTableChangesPerWrite)
+		maxTableChangesPerWrite)
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
 		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
@@ -735,10 +737,6 @@ func TestPeerHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
 	}
 }
 
-// TestPrefixHandlerWaitsAtTheMaximumBatchSize gives the prefix handler a
-// maximum batch size of one batch, so the test reaches it with one batch more.
-// Each update is applied to the routing table, which makes the real maximum of
-// ten batches slow to reach under the race detector.
 func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	t.Parallel()
 
@@ -746,7 +744,8 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 
 	store := newBlockingStore()
 	h := newPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}),
-		longerThanAnyTest, prefixBatchSize, maxTableChangesPerWrite)
+		longerThanAnyTest, maxTableChangesPerWrite)
+	h.batchMaxSize = prefixBatchSize
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	handleAtMaxBatchSize(t, store, h,
@@ -768,8 +767,9 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 
 	store := newBlockingStore()
 	h := newASHandler(store, logger.New(), longerThanAnyTest)
+	h.batchMaxSize = asnBatchSize
 	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
-	toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchMaxSize)}
+	toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
 	handleAtMaxBatchSize(t, store, h,
 		[]*ristypes.RISMessage{{Path: fill}},
 		[]*ristypes.RISMessage{{Path: toMax}},
@@ -787,7 +787,7 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 // TestPeerHandlerWaitsAtTheMaximumBatchSize must not have its flush timer fire.
 // A timer flush during the fill would take only part of it. The batch would
 // then reach its maximum size before the test's last message, and the test
-// would hang.
+// would fail.
 func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	t.Parallel()
 
@@ -795,10 +795,11 @@ func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 
 	store := newBlockingStore()
 	h := newPeerHandler(store, logger.New(), longerThanAnyTest)
+	h.batchMaxSize = peerBatchSize
 	message := []*ristypes.RISMessage{{Peer: testPeerIP}}
 	handleAtMaxBatchSize(t, store, h,
 		slices.Repeat(message, peerBatchSize),
-		slices.Repeat(message, peerBatchMaxSize),
+		slices.Repeat(message, peerBatchSize),
 		&ristypes.RISMessage{Peer: otherPeerIP})
 
 	store.mu.Lock()
@@ -842,7 +843,10 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			name:      "AS",
 			batchSize: asnBatchSize,
 			handle: func(n int) chan struct{} {
-				h := &ASHandler{batchFullCh: make(chan struct{}, 1)}
+				h := &ASHandler{
+					batchMaxSize: asnBatchMaxSize,
+					batchFullCh:  make(chan struct{}, 1),
+				}
 				// The origin AS and each AS in the path are one update each.
 				h.HandleMessage(&ristypes.RISMessage{
 					Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)},
@@ -855,7 +859,10 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			name:      "peer",
 			batchSize: peerBatchSize,
 			handle: func(n int) chan struct{} {
-				h := &PeerHandler{batchFullCh: make(chan struct{}, 1)}
+				h := &PeerHandler{
+					batchMaxSize: peerBatchMaxSize,
+					batchFullCh:  make(chan struct{}, 1),
+				}
 				for range n {
 					h.HandleMessage(&ristypes.RISMessage{Peer: testPeerIP})
 				}

internal/routewatch/peerhandler.go

@@ -43,10 +43,14 @@ type PeerHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
+	// batchMaxSize is peerBatchMaxSize. Only HandleMessage reads it, so a test
+	// may lower it before it hands the handler a message.
+	batchMaxSize int
+
 	// Batching. mu guards peerBatch and writing only: the flush goroutine
 	// takes the batch under mu and writes it after releasing mu, so
 	// HandleMessage waits for a database write only while peerBatch is at
-	// peerBatchMaxSize.
+	// batchMaxSize.
 	mu        sync.Mutex
 	peerBatch []peerUpdate
 	// writing is how many updates the flush goroutine took from peerBatch
@@ -84,12 +88,13 @@ func newPeerHandler(
 	db database.Store, logger *logger.Logger, batchTimeout time.Duration,
 ) *PeerHandler {
 	h := &PeerHandler{
-		db:          db,
+		db:           db,
-		logger:      logger,
+		logger:       logger,
-		peerBatch:   make([]peerUpdate, 0, peerBatchSize),
+		batchMaxSize: peerBatchMaxSize,
-		batchFullCh: make(chan struct{}, 1),
+		peerBatch:    make([]peerUpdate, 0, peerBatchSize),
-		stopCh:      make(chan context.Context, 1),
+		batchFullCh:  make(chan struct{}, 1),
-		done:        make(chan struct{}),
+		stopCh:       make(chan context.Context, 1),
+		done:         make(chan struct{}),
 	}
 	h.batchTaken.L = &h.mu
 
@@ -135,7 +140,7 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// While the batch is at its maximum size, wait until the flush goroutine
 	// takes it. Meanwhile the handler takes nothing off its queue, which
 	// fills, and the streamer drops messages for it.
-	for len(h.peerBatch) >= peerBatchMaxSize {
+	for len(h.peerBatch) >= h.batchMaxSize {
 		h.batchTaken.Wait()
 	}
 
@@ -149,7 +154,7 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 
 	// A full batch wakes the flush goroutine. If it is writing, the wake-up
 	// waits in batchFullCh and the batch keeps growing until the write ends
-	// or the batch reaches peerBatchMaxSize.
+	// or the batch reaches batchMaxSize.
 	if len(h.peerBatch) >= peerBatchSize {
 		select {
 		case h.batchFullCh <- struct{}{}:

internal/routewatch/peerhandler_internal_test.go

@@ -35,7 +35,7 @@ func TestPeerHandlerWritesTheEarliestAndLatestTimesOfANewPeer(t *testing.T) {
 
 	// Built without NewPeerHandler's flush goroutine, so the test flushes
 	// itself.
-	h := &PeerHandler{db: store, logger: logger.New()}
+	h := &PeerHandler{db: store, logger: logger.New(), batchMaxSize: peerBatchMaxSize}
 
 	first := peerAnnouncement(testPeerIP, testV4Prefix)
 	second := peerAnnouncement(testPeerIP, testV4Prefix)

internal/routewatch/prefixhandler.go

@@ -60,9 +60,11 @@ type PrefixHandler struct {
 	metrics *metrics.Tracker
 	table   *routetable.Table
 
-	// batchMaxSize is prefixBatchMaxSize, and tableChangesPerWrite is
+	// batchMaxSize is prefixBatchMaxSize. Only HandleMessage reads it, so a
-	// maxTableChangesPerWrite, except in tests that give smaller ones.
+	// test may lower it before it hands the handler a message.
-	batchMaxSize         int
+	batchMaxSize int
+	// tableChangesPerWrite is maxTableChangesPerWrite, except in tests that
+	// give a smaller one.
 	tableChangesPerWrite int
 
 	// Batching. mu guards batch, writing and writingTableChanges only: the
@@ -125,24 +127,25 @@ func NewPrefixHandler(
 	db database.Store, logger *logger.Logger, table *routetable.Table,
 ) *PrefixHandler {
 	return newPrefixHandler(db, logger, table, prefixBatchTimeout,
-		prefixBatchMaxSize, maxTableChangesPerWrite)
+		maxTableChangesPerWrite)
 }
 
 // newPrefixHandler is NewPrefixHandler with the flush timer's interval,
-// batchTimeout, in place of prefixBatchTimeout. batchMaxSize and
+// batchTimeout, in place of prefixBatchTimeout, and tableChangesPerWrite in
-// tableChangesPerWrite replace prefixBatchMaxSize and maxTableChangesPerWrite.
+// place of maxTableChangesPerWrite. A test that the timer must not disturb
-// A test that the timer must not disturb gives an interval longer than the test
+// gives an interval longer than the test runs. A test of writes split over
-// runs. A test of either limit gives a smaller one, so that it reaches it fast.
+// several flushes gives a small tableChangesPerWrite, so that it exceeds it
+// with few changes.
 func newPrefixHandler(
 	db database.Store, logger *logger.Logger, table *routetable.Table,
-	batchTimeout time.Duration, batchMaxSize, tableChangesPerWrite int,
+	batchTimeout time.Duration, tableChangesPerWrite int,
 ) *PrefixHandler {
 	h := &PrefixHandler{
 		db:                   db,
 		logger:               logger,
 		table:                table,
 		batch:                make([]seenPrefix, 0, prefixBatchSize),
-		batchMaxSize:         batchMaxSize,
+		batchMaxSize:         prefixBatchMaxSize,
 		tableChangesPerWrite: tableChangesPerWrite,
 		batchFullCh:          make(chan struct{}, 1),
 		stopCh:               make(chan context.Context, 1),

internal/routewatch/prefixhandler_table_internal_test.go

@@ -391,7 +391,7 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
 	announceManyPrefixes(table, pairs)
 
 	h := newPrefixHandler(store, logger.New(), table, longerThanAnyTest,
-		prefixBatchMaxSize, testMaxTableChangesPerWrite)
+		testMaxTableChangesPerWrite)
 
 	defer stopInTime(t, h)
 
@@ -436,7 +436,7 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
 	announceManyPrefixes(table, pairs)
 
 	h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest,
-		prefixBatchMaxSize, testMaxTableChangesPerWrite)
+		testMaxTableChangesPerWrite)
 	h.Stop(t.Context())
 
 	written := 0