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. Under the race detector on a loaded host that
takes longer, so they failed although nothing waited. A test now waits up
to stepTimeout, 15 seconds, for each step it expects.

The slowest tests filled production limits: 500,000 paths in the peering
handler, and 30,000 table changes or 250,000 updates in the prefix handler.
The handlers' unexported constructors now take these limits, as they take
their timer interval, and those tests give small ones. 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 handlers' limits give
+  15 seconds for each step it expects. The tests of the peering handler's path
-  the handlers small ones, so the package no longer runs past its 90-second
+  limit and of the prefix handler's limits give the handler small limits, so the
-  limit (closes #213, #215)
+  package no longer runs past its 90-second 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,13 +40,9 @@ 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 batchMaxSize.
+	// waits for a database write only while batch is at asnBatchMaxSize.
 	mu    sync.Mutex
 	batch []asnOp
 	// writing is how many operations the flush goroutine took from batch
@@ -83,13 +79,12 @@ func newASHandler(
 	db database.Store, logger *logger.Logger, batchTimeout time.Duration,
 ) *ASHandler {
 	h := &ASHandler{
-		db:           db,
+		db:          db,
-		logger:       logger,
+		logger:      logger,
-		batchMaxSize: asnBatchMaxSize,
+		batch:       make([]asnOp, 0, asnBatchSize),
-		batch:        make([]asnOp, 0, asnBatchSize),
+		batchFullCh: make(chan struct{}, 1),
-		batchFullCh:  make(chan struct{}, 1),
+		stopCh:      make(chan context.Context, 1),
-		stopCh:       make(chan context.Context, 1),
+		done:        make(chan struct{}),
-		done:         make(chan struct{}),
 	}
 	h.batchTaken.L = &h.mu
 
@@ -132,7 +127,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) >= h.batchMaxSize {
+	for len(h.batch) >= asnBatchMaxSize {
 		h.batchTaken.Wait()
 	}
 
@@ -156,7 +151,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 batchMaxSize.
+	// or the batch reaches asnBatchMaxSize.
 	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(), batchMaxSize: asnBatchMaxSize}
+	h := &ASHandler{db: db, logger: logger.New()}
 
 	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(), batchMaxSize: asnBatchMaxSize}
+	h := &ASHandler{db: store, logger: logger.New()}
 
 	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, for three times statsWindow, and never
+// second, by a clock the test sets, far longer than 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,9 +292,7 @@ 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. The tests lower h's maximum batch
+// it stops h, which writes what it holds.
-// 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,
@@ -409,7 +407,7 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	store := newBlockingStore()
 	h := newPrefixHandler(store, newErrorLogger(&logs),
 		routetable.New(routetable.Params{}), longerThanAnyTest,
-		maxTableChangesPerWrite)
+		prefixBatchMaxSize, maxTableChangesPerWrite)
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
 		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
@@ -737,6 +735,10 @@ 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()
 
@@ -744,8 +746,7 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 
 	store := newBlockingStore()
 	h := newPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}),
-		longerThanAnyTest, maxTableChangesPerWrite)
+		longerThanAnyTest, prefixBatchSize, maxTableChangesPerWrite)
-	h.batchMaxSize = prefixBatchSize
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	handleAtMaxBatchSize(t, store, h,
@@ -767,9 +768,8 @@ 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}, asnBatchSize)}
+	toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchMaxSize)}
 	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 fail.
+// would hang.
 func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	t.Parallel()
 
@@ -795,11 +795,10 @@ 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, peerBatchSize),
+		slices.Repeat(message, peerBatchMaxSize),
 		&ristypes.RISMessage{Peer: otherPeerIP})
 
 	store.mu.Lock()
@@ -843,10 +842,7 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			name:      "AS",
 			batchSize: asnBatchSize,
 			handle: func(n int) chan struct{} {
-				h := &ASHandler{
+				h := &ASHandler{batchFullCh: make(chan struct{}, 1)}
-					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)},
@@ -859,10 +855,7 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			name:      "peer",
 			batchSize: peerBatchSize,
 			handle: func(n int) chan struct{} {
-				h := &PeerHandler{
+				h := &PeerHandler{batchFullCh: make(chan struct{}, 1)}
-					batchMaxSize: peerBatchMaxSize,
-					batchFullCh:  make(chan struct{}, 1),
-				}
 				for range n {
 					h.HandleMessage(&ristypes.RISMessage{Peer: testPeerIP})
 				}

internal/routewatch/peerhandler.go

@@ -43,14 +43,10 @@ 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
-	// batchMaxSize.
+	// peerBatchMaxSize.
 	mu        sync.Mutex
 	peerBatch []peerUpdate
 	// writing is how many updates the flush goroutine took from peerBatch
@@ -88,13 +84,12 @@ func newPeerHandler(
 	db database.Store, logger *logger.Logger, batchTimeout time.Duration,
 ) *PeerHandler {
 	h := &PeerHandler{
-		db:           db,
+		db:          db,
-		logger:       logger,
+		logger:      logger,
-		batchMaxSize: peerBatchMaxSize,
+		peerBatch:   make([]peerUpdate, 0, peerBatchSize),
-		peerBatch:    make([]peerUpdate, 0, peerBatchSize),
+		batchFullCh: make(chan struct{}, 1),
-		batchFullCh:  make(chan struct{}, 1),
+		stopCh:      make(chan context.Context, 1),
-		stopCh:       make(chan context.Context, 1),
+		done:        make(chan struct{}),
-		done:         make(chan struct{}),
 	}
 	h.batchTaken.L = &h.mu
 
@@ -140,7 +135,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) >= h.batchMaxSize {
+	for len(h.peerBatch) >= peerBatchMaxSize {
 		h.batchTaken.Wait()
 	}
 
@@ -154,7 +149,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 batchMaxSize.
+	// or the batch reaches peerBatchMaxSize.
 	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(), batchMaxSize: peerBatchMaxSize}
+	h := &PeerHandler{db: store, logger: logger.New()}
 
 	first := peerAnnouncement(testPeerIP, testV4Prefix)
 	second := peerAnnouncement(testPeerIP, testV4Prefix)

internal/routewatch/prefixhandler.go

@@ -60,11 +60,9 @@ type PrefixHandler struct {
 	metrics *metrics.Tracker
 	table   *routetable.Table
 
-	// batchMaxSize is prefixBatchMaxSize. Only HandleMessage reads it, so a
+	// batchMaxSize is prefixBatchMaxSize, and tableChangesPerWrite is
-	// test may lower it before it hands the handler a message.
+	// maxTableChangesPerWrite, except in tests that give smaller ones.
-	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
@@ -127,25 +125,24 @@ func NewPrefixHandler(
 	db database.Store, logger *logger.Logger, table *routetable.Table,
 ) *PrefixHandler {
 	return newPrefixHandler(db, logger, table, prefixBatchTimeout,
-		maxTableChangesPerWrite)
+		prefixBatchMaxSize, maxTableChangesPerWrite)
 }
 
 // newPrefixHandler is NewPrefixHandler with the flush timer's interval,
-// batchTimeout, in place of prefixBatchTimeout, and tableChangesPerWrite in
+// batchTimeout, in place of prefixBatchTimeout. batchMaxSize and
-// place of maxTableChangesPerWrite. A test that the timer must not disturb
+// tableChangesPerWrite replace prefixBatchMaxSize and maxTableChangesPerWrite.
-// gives an interval longer than the test runs. A test of writes split over
+// A test that the timer must not disturb gives an interval longer than the test
-// several flushes gives a small tableChangesPerWrite, so that it exceeds it
+// runs. A test of either limit gives a smaller one, so that it reaches it fast.
-// with few changes.
 func newPrefixHandler(
 	db database.Store, logger *logger.Logger, table *routetable.Table,
-	batchTimeout time.Duration, tableChangesPerWrite int,
+	batchTimeout time.Duration, batchMaxSize, tableChangesPerWrite int,
 ) *PrefixHandler {
 	h := &PrefixHandler{
 		db:                   db,
 		logger:               logger,
 		table:                table,
 		batch:                make([]seenPrefix, 0, prefixBatchSize),
-		batchMaxSize:         prefixBatchMaxSize,
+		batchMaxSize:         batchMaxSize,
 		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,
-		testMaxTableChangesPerWrite)
+		prefixBatchMaxSize, testMaxTableChangesPerWrite)
 
 	defer stopInTime(t, h)
 
@@ -436,7 +436,7 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
 	announceManyPrefixes(table, pairs)
 
 	h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest,
-		testMaxTableChangesPerWrite)
+		prefixBatchMaxSize, testMaxTableChangesPerWrite)
 	h.Stop(t.Context())
 
 	written := 0