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

@@ -35,6 +35,12 @@ tested.
 
 # Completed Steps
 
+- 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
+  limit and of the prefix handler's limits give the handler small limits, so the
+  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/app_integration_internal_test.go

@@ -509,8 +509,6 @@ func announcementFeed(prefixes []string) string {
 func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
 	t.Parallel()
 
-	const maxWait = 10 * time.Second
-
 	prefixes := []string{"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"}
 	feed := announcementFeed(prefixes)
 
@@ -556,9 +554,9 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
 	for range cap(linesSeen) {
 		select {
 		case <-linesSeen:
-		case <-time.After(maxWait):
+		case <-time.After(stepTimeout):
 			t.Fatalf("the streamer did not read %d lines within %s",
-				cap(linesSeen), maxWait)
+				cap(linesSeen), stepTimeout)
 		}
 	}
 
@@ -571,8 +569,8 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
 
 	select {
 	case <-shutDown:
-	case <-time.After(maxWait):
-		t.Fatalf("Shutdown did not return within %s", maxWait)
+	case <-time.After(stepTimeout):
+		t.Fatalf("Shutdown did not return within %s", stepTimeout)
 	}
 
 	store.mu.Lock()
@@ -609,21 +607,16 @@ func freePort(t *testing.T) int {
 	return addr.Port
 }
 
-// TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked starts the HTTP server
-// and has the AS handler start a write that the store holds until the test
-// ends, past Shutdown's deadline. Shutdown must return within its deadline
-// plus a small margin, with the HTTP server stopped, and must neither
-// checkpoint nor close the database: the real one would wait for the write.
+// TestShutdownDoesNotWaitForABlockedWrite starts the HTTP server and has the
+// AS handler start a write that the store holds until the test ends, past
+// Shutdown's deadline. Shutdown must return without waiting for the write, with
+// the HTTP server stopped, and must neither checkpoint nor close the database:
+// the real one would wait for the write.
 // Before the fix Shutdown waited for the write.
-func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
+func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) {
 	t.Parallel()
 
-	const (
-		deadline = 100 * time.Millisecond
-		// margin is how long Shutdown may take after its deadline, to stop
-		// the HTTP server.
-		margin = 2 * time.Second
-	)
+	const deadline = 100 * time.Millisecond
 
 	blocking := newBlockingStore()
 	defer blocking.releaseWrites()
@@ -667,9 +660,9 @@ func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
 
 	select {
 	case <-shutDown:
-	case <-time.After(deadline + margin):
+	case <-time.After(deadline + stepTimeout):
 		t.Fatalf("Shutdown did not return within %s of its deadline while "+
-			"a write was blocked", margin)
+			"a write was blocked", stepTimeout)
 	}
 
 	var dialer net.Dialer
@@ -744,8 +737,6 @@ func (s *closeOrderStore) record(call string) {
 func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
 	t.Parallel()
 
-	const deadline = 10 * time.Second
-
 	store := &closeOrderStore{Store: newMockStore()}
 	log := logger.New()
 	cfg := &config.Config{}
@@ -763,7 +754,7 @@ func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
 		Path: ristypes.ASPath{Sequence: []int{testASNA}},
 	})
 
-	ctx, cancel := context.WithTimeout(t.Context(), deadline)
+	ctx, cancel := context.WithTimeout(t.Context(), stepTimeout)
 	defer cancel()
 
 	rw.Shutdown(ctx)

internal/routewatch/asnfetcher_internal_test.go

@@ -224,7 +224,7 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
 
 	select {
 	case <-lookupStarted:
-	case <-time.After(handleMessageTimeout):
+	case <-time.After(stepTimeout):
 		t.Fatal("the fetcher did not start the lookup")
 	}
 
@@ -240,7 +240,7 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
 
 	select {
 	case <-f.done:
-	case <-time.After(handleMessageTimeout):
+	case <-time.After(stepTimeout):
 		t.Fatal("the fetcher goroutine did not return after the lookup ended")
 	}
 }

internal/routewatch/batchwrite_internal_test.go

@@ -19,9 +19,15 @@ import (
 )
 
 const (
-	// handleMessageTimeout is how long a test waits for HandleMessage to
-	// return, or for a write to start, before it fails.
-	handleMessageTimeout = 2 * time.Second
+	// stepTimeout is how long a test waits for a step it expects, such as
+	// HandleMessage returning, a write starting or Shutdown returning, before
+	// it fails. Only a step that never comes should reach it. On a loaded
+	// host, under the race detector, HandleMessage takes seconds to apply a
+	// full prefix batch. The helpers below bound every wait with it, Stop
+	// included. A test that hangs in one of them waits out at most two of
+	// these bounds, its step's and Stop's, and then fails with its own
+	// message. That leaves time before go test's 90-second timeout.
+	stepTimeout = 15 * time.Second
 
 	// stillWaitingAfter is how long a test watches a HandleMessage that must
 	// wait. One that does not wait returns well within it.
@@ -198,7 +204,7 @@ func lastLogged(t *testing.T, logs *bytes.Buffer) stopLog {
 	return logged
 }
 
-// returnsInTime reports whether f returns within handleMessageTimeout.
+// returnsInTime reports whether f returns within stepTimeout.
 func returnsInTime(f func()) bool {
 	done := make(chan struct{})
 
@@ -210,11 +216,22 @@ func returnsInTime(f func()) bool {
 	select {
 	case <-done:
 		return true
-	case <-time.After(handleMessageTimeout):
+	case <-time.After(stepTimeout):
 		return false
 	}
 }
 
+// stopInTime stops h, with no deadline, and fails the test if Stop does not
+// return within stepTimeout. It only marks the test failed, so it can be
+// deferred.
+func stopInTime(t *testing.T, h interface{ Stop(ctx context.Context) }) {
+	t.Helper()
+
+	if !returnsInTime(func() { h.Stop(t.Context()) }) {
+		t.Errorf("Stop did not return within %s", stepTimeout)
+	}
+}
+
 // startBlockedWrite hands h the messages in fill and waits until h starts to
 // write them to store. The prefix, AS and peer handlers start when fill fills
 // their batch, and the peering handler on its timer. store holds the write
@@ -240,7 +257,7 @@ func startBlockedWrite(
 
 	select {
 	case <-store.writeStarted:
-	case <-time.After(handleMessageTimeout):
+	case <-time.After(stepTimeout):
 		t.Fatal("the handler did not start to write the messages")
 	}
 }
@@ -261,7 +278,7 @@ func handleWhileAWriteIsBlocked(
 ) {
 	t.Helper()
 
-	defer h.Stop(t.Context())
+	defer stopInTime(t, h)
 	defer store.releaseWrites()
 
 	startBlockedWrite(t, store, h, fill)
@@ -288,13 +305,18 @@ func handleAtMaxBatchSize(
 ) {
 	t.Helper()
 
-	defer h.Stop(t.Context())
+	defer stopInTime(t, h)
 	defer store.releaseWrites()
 
 	startBlockedWrite(t, store, h, fill)
 
-	for _, msg := range toMax {
-		h.HandleMessage(msg)
+	if !returnsInTime(func() {
+		for _, msg := range toMax {
+			h.HandleMessage(msg)
+		}
+	}) {
+		t.Fatal("HandleMessage did not return while the batch grew to its " +
+			"maximum size")
 	}
 
 	returned := make(chan struct{})
@@ -315,7 +337,7 @@ func handleAtMaxBatchSize(
 
 	select {
 	case <-returned:
-	case <-time.After(handleMessageTimeout):
+	case <-time.After(stepTimeout):
 		t.Fatal("HandleMessage did not return after the database write finished")
 	}
 }
@@ -345,8 +367,12 @@ func stopWhileAWriteIsBlocked(
 
 	startBlockedWrite(t, store, h, fill)
 
-	for _, msg := range extra {
-		h.HandleMessage(msg)
+	if !returnsInTime(func() {
+		for _, msg := range extra {
+			h.HandleMessage(msg)
+		}
+	}) {
+		t.Fatal("HandleMessage did not return while a database write was blocked")
 	}
 
 	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
@@ -361,7 +387,7 @@ func stopWhileAWriteIsBlocked(
 
 	select {
 	case <-done:
-	case <-time.After(handleMessageTimeout):
+	case <-time.After(stepTimeout):
 		t.Fatal("the flush goroutine did not return after the write finished")
 	}
 }
@@ -380,7 +406,8 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 
 	store := newBlockingStore()
 	h := newPrefixHandler(store, newErrorLogger(&logs),
-		routetable.New(routetable.Params{}), longerThanAnyTest)
+		routetable.New(routetable.Params{}), longerThanAnyTest,
+		prefixBatchMaxSize, maxTableChangesPerWrite)
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
 		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
@@ -597,7 +624,7 @@ func TestPeeringHandlerStopCountsTheWriteInProgress(t *testing.T) {
 
 	select {
 	case <-h.done:
-	case <-time.After(handleMessageTimeout):
+	case <-time.After(stepTimeout):
 		t.Fatal("the processing goroutine did not return after the write " +
 			"finished")
 	}
@@ -618,7 +645,7 @@ func TestPeeringHandlerStartsNoWriteAfterTheStopDeadline(t *testing.T) {
 	const processInterval = 10 * time.Millisecond
 
 	store := newBlockingStore()
-	h := newPeeringHandler(store, logger.New(), processInterval)
+	h := newPeeringHandler(store, logger.New(), processInterval, maxTrackedPaths)
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
 		[]*ristypes.RISMessage{
 			{Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}},
@@ -708,15 +735,20 @@ 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()
 
 	const extraPrefix = "198.51.100.0/24"
 
 	store := newBlockingStore()
-	h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))
+	h := newPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}),
+		longerThanAnyTest, prefixBatchSize, maxTableChangesPerWrite)
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
-	toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchMaxSize)
+	toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	handleAtMaxBatchSize(t, store, h,
 		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
 		[]*ristypes.RISMessage{prefixAnnouncement(toMax)},
@@ -796,8 +828,9 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			batchSize: prefixBatchSize,
 			handle: func(n int) chan struct{} {
 				h := &PrefixHandler{
-					batchFullCh: make(chan struct{}, 1),
-					table:       routetable.New(routetable.Params{}),
+					batchMaxSize: prefixBatchMaxSize,
+					batchFullCh:  make(chan struct{}, 1),
+					table:        routetable.New(routetable.Params{}),
 				}
 				h.HandleMessage(prefixAnnouncement(
 					slices.Repeat([]string{testV4Prefix}, n)))

internal/routewatch/dbmaintainer_internal_test.go

@@ -97,7 +97,7 @@ func TestDBMaintainerStopReturnsByItsDeadline(t *testing.T) {
 
 	select {
 	case <-store.started:
-	case <-time.After(handleMessageTimeout):
+	case <-time.After(stepTimeout):
 		t.Fatal("the maintainer did not start the checkpoint")
 	}
 
@@ -113,7 +113,7 @@ func TestDBMaintainerStopReturnsByItsDeadline(t *testing.T) {
 
 	select {
 	case <-m.done:
-	case <-time.After(handleMessageTimeout):
+	case <-time.After(stepTimeout):
 		t.Fatal("the maintenance goroutine did not return after the " +
 			"checkpoint ended")
 	}

internal/routewatch/peeringhandler.go

@@ -42,6 +42,10 @@ type PeeringHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
+	// maxPaths is the most paths asPaths holds: maxTrackedPaths, except in
+	// tests that give a smaller one.
+	maxPaths int
+
 	// In-memory AS path tracking: each path, JSON-encoded, with the earliest
 	// and the latest time an update carried it.
 	mu           sync.Mutex
@@ -62,21 +66,24 @@ type PeeringHandler struct {
 // periodically processes accumulated AS paths into peering records. The
 // handler begins processing immediately upon creation.
 func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler {
-	return newPeeringHandler(db, logger, peeringProcessInterval)
+	return newPeeringHandler(db, logger, peeringProcessInterval, maxTrackedPaths)
 }
 
 // newPeeringHandler is NewPeeringHandler with the processing timer's interval,
-// processInterval, in place of peeringProcessInterval. A test gives a short
-// one to have the timer start a write.
+// processInterval, in place of peeringProcessInterval, and the most paths it
+// holds, maxPaths, in place of maxTrackedPaths. A test gives a short interval to
+// have the timer start a write, and a small maxPaths to fill the map quickly.
 func newPeeringHandler(
 	db database.Store, logger *logger.Logger, processInterval time.Duration,
+	maxPaths int,
 ) *PeeringHandler {
 	h := &PeeringHandler{
-		db:      db,
-		logger:  logger,
-		asPaths: make(map[string]database.SeenTimes),
-		stopCh:  make(chan context.Context, 1),
-		done:    make(chan struct{}),
+		db:       db,
+		logger:   logger,
+		asPaths:  make(map[string]database.SeenTimes),
+		maxPaths: maxPaths,
+		stopCh:   make(chan context.Context, 1),
+		done:     make(chan struct{}),
 	}
 
 	// Start the periodic processing goroutine
@@ -131,7 +138,7 @@ func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) {
 	defer h.mu.Unlock()
 
 	seen, exists := h.asPaths[key]
-	if !exists && len(h.asPaths) >= maxTrackedPaths {
+	if !exists && len(h.asPaths) >= h.maxPaths {
 		// Map is full; drop this new path and count it.
 		h.droppedPaths++
 

internal/routewatch/peeringhandler_internal_test.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"maps"
 	"slices"
-	"strconv"
 	"sync"
 	"testing"
 	"time"
@@ -20,6 +19,10 @@ const (
 	testASNB = 64501
 	testASNC = 64502
 	testASND = 64503
+
+	// testMaxTrackedPaths is a limit of the path map that a test fills
+	// quickly, in place of maxTrackedPaths.
+	testMaxTrackedPaths = 10
 )
 
 // recordingStore wraps mockStore to count every RecordPeering call, so a
@@ -52,9 +55,10 @@ func (r *recordingStore) callCount() int {
 // processing goroutine, so tests drive processing explicitly.
 func newTestHandler(db database.Store) *PeeringHandler {
 	return &PeeringHandler{
-		db:      db,
-		logger:  logger.New(),
-		asPaths: make(map[string]database.SeenTimes),
+		db:       db,
+		logger:   logger.New(),
+		asPaths:  make(map[string]database.SeenTimes),
+		maxPaths: maxTrackedPaths,
 	}
 }
 
@@ -244,25 +248,28 @@ func TestPeeringHandlerStopWritesTheRemainingPaths(t *testing.T) {
 
 // TestPeeringHandlerCapDropsAndCounts verifies that a full map drops new paths
 // and counts them, while a path already tracked is refreshed rather than
-// dropped.
+// dropped. The handler's limit is small, and its timer does not fire.
 func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 	t.Parallel()
 
 	store := &recordingStore{mockStore: newMockStore()}
-	h := newTestHandler(store)
+	h := newPeeringHandler(store, logger.New(), longerThanAnyTest,
+		testMaxTrackedPaths)
+
+	defer stopInTime(t, h)
 
 	now := time.Now().UTC()
 
-	// Fill the map to exactly maxTrackedPaths, including one real path key so
-	// the "already tracked" branch can be exercised. The filler keys are never
-	// processed in this test, so their contents do not matter.
+	// Fill the map to exactly its limit, including one real path key so the
+	// "already tracked" branch can be exercised. The other paths are of one
+	// AS, so they carry no peering.
 	existing := pathKey(t, testASNA, testASNB)
 
 	h.mu.Lock()
 	h.asPaths[existing] = database.SeenTimes{First: now, Last: now}
 
-	for i := 0; len(h.asPaths) < maxTrackedPaths; i++ {
-		h.asPaths[strconv.Itoa(i)] = database.SeenTimes{First: now, Last: now}
+	for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ {
+		h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now}
 	}
 	h.mu.Unlock()
 
@@ -270,8 +277,9 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 	handle(h, now.Add(time.Second), testASNA, testASNC)
 
 	tracked, dropped := snapshot(h)
-	if tracked != maxTrackedPaths {
-		t.Fatalf("want map size %d after drop, got %d", maxTrackedPaths, tracked)
+	if tracked != testMaxTrackedPaths {
+		t.Fatalf("want map size %d after drop, got %d", testMaxTrackedPaths,
+			tracked)
 	}
 
 	if dropped != 1 {
@@ -283,8 +291,9 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 	handle(h, refreshed, testASNA, testASNB)
 
 	tracked, dropped = snapshot(h)
-	if tracked != maxTrackedPaths {
-		t.Fatalf("want map size %d after refresh, got %d", maxTrackedPaths, tracked)
+	if tracked != testMaxTrackedPaths {
+		t.Fatalf("want map size %d after refresh, got %d", testMaxTrackedPaths,
+			tracked)
 	}
 
 	if dropped != 1 {
@@ -303,17 +312,21 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 
 // TestPeeringHandlerRunResetsTheDroppedCount fills the map, has the handler
 // drop one path, and runs the processing. The count of dropped paths is the
-// run's own, so it is 0 again after the run.
+// run's own, so it is 0 again after the run. The handler's limit is small, and
+// its timer does not fire.
 func TestPeeringHandlerRunResetsTheDroppedCount(t *testing.T) {
 	t.Parallel()
 
-	h := newTestHandler(newMockStore())
+	h := newPeeringHandler(newMockStore(), logger.New(), longerThanAnyTest,
+		testMaxTrackedPaths)
+
+	defer stopInTime(t, h)
 
 	now := time.Now().UTC()
 
 	// Paths of one AS carry no peering, so the run records nothing.
 	h.mu.Lock()
-	for asn := 1; len(h.asPaths) < maxTrackedPaths; asn++ {
+	for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ {
 		h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now}
 	}
 	h.mu.Unlock()

internal/routewatch/prefixhandler.go

@@ -60,10 +60,15 @@ type PrefixHandler struct {
 	metrics *metrics.Tracker
 	table   *routetable.Table
 
+	// batchMaxSize is prefixBatchMaxSize, and tableChangesPerWrite is
+	// maxTableChangesPerWrite, except in tests that give smaller ones.
+	batchMaxSize         int
+	tableChangesPerWrite int
+
 	// Batching. mu guards batch, writing and writingTableChanges only: the
 	// flush goroutine takes them under mu and writes them after releasing
 	// mu, so HandleMessage waits for a database write only while batch is at
-	// prefixBatchMaxSize.
+	// batchMaxSize.
 	mu    sync.Mutex
 	batch []seenPrefix
 	// writing is how many updates the flush goroutine took and has not
@@ -119,24 +124,29 @@ func (u prefixUpdate) announcesRoute() bool {
 func NewPrefixHandler(
 	db database.Store, logger *logger.Logger, table *routetable.Table,
 ) *PrefixHandler {
-	return newPrefixHandler(db, logger, table, prefixBatchTimeout)
+	return newPrefixHandler(db, logger, table, prefixBatchTimeout,
+		prefixBatchMaxSize, maxTableChangesPerWrite)
 }
 
 // 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.
+// batchTimeout, in place of prefixBatchTimeout. batchMaxSize and
+// tableChangesPerWrite replace prefixBatchMaxSize and maxTableChangesPerWrite.
+// A test that the timer must not disturb gives an interval longer than the test
+// runs. A test of either limit gives a smaller one, so that it reaches it fast.
 func newPrefixHandler(
 	db database.Store, logger *logger.Logger, table *routetable.Table,
-	batchTimeout time.Duration,
+	batchTimeout time.Duration, batchMaxSize, tableChangesPerWrite int,
 ) *PrefixHandler {
 	h := &PrefixHandler{
-		db:          db,
-		logger:      logger,
-		table:       table,
-		batch:       make([]seenPrefix, 0, prefixBatchSize),
-		batchFullCh: make(chan struct{}, 1),
-		stopCh:      make(chan context.Context, 1),
-		done:        make(chan struct{}),
+		db:                   db,
+		logger:               logger,
+		table:                table,
+		batch:                make([]seenPrefix, 0, prefixBatchSize),
+		batchMaxSize:         batchMaxSize,
+		tableChangesPerWrite: tableChangesPerWrite,
+		batchFullCh:          make(chan struct{}, 1),
+		stopCh:               make(chan context.Context, 1),
+		done:                 make(chan struct{}),
 	}
 	h.batchTaken.L = &h.mu
 
@@ -191,7 +201,7 @@ func (h *PrefixHandler) 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) >= prefixBatchMaxSize {
+	for len(h.batch) >= h.batchMaxSize {
 		h.batchTaken.Wait()
 	}
 
@@ -239,7 +249,7 @@ func (h *PrefixHandler) 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 prefixBatchMaxSize.
+	// or the batch reaches batchMaxSize.
 	if len(h.batch) >= prefixBatchSize {
 		select {
 		case h.batchFullCh <- struct{}{}:
@@ -250,7 +260,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 
 // Stop stops the flush goroutine, which makes one last flush before it returns,
 // and waits for it until ctx ends. That flush writes at most
-// maxTableChangesPerWrite of the routing table's changes, and the prefixes of
+// tableChangesPerWrite of the routing table's changes, and the prefixes of
 // the updates the handler holds. Once ctx has ended the flush goroutine skips
 // that last flush, and a write still running is lost when the process exits. A
 // failed write of the table's changes is not retried either. Stop logs how many
@@ -378,14 +388,13 @@ func (h *PrefixHandler) applyToTable(update prefixUpdate) {
 	})
 }
 
-// writeTableChanges writes at most maxTableChangesPerWrite changes of the
-// routing table in memory, and every peer the table numbered since the last
-// write.
+// writeTableChanges writes at most tableChangesPerWrite changes of the routing
+// table in memory, and every peer the table numbered since the last write.
 // When the write fails, they go back to the table, so that a later write writes
 // them again, each pair with its content then.
 func (h *PrefixHandler) writeTableChanges() {
 	h.mu.Lock()
-	changes, newPeers := h.table.TakeChanges(maxTableChangesPerWrite)
+	changes, newPeers := h.table.TakeChanges(h.tableChangesPerWrite)
 	h.writingTableChanges = len(changes)
 	h.mu.Unlock()
 

internal/routewatch/prefixhandler_internal_test.go

@@ -237,9 +237,10 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
 	}
 }
 
-// longerThanAnyTest is a flush timer interval that no test lasts. A test whose
-// prefix, AS or peer handler must flush only when its batch is full and on Stop
-// gives it to newPrefixHandler, newASHandler or newPeerHandler.
+// longerThanAnyTest is a timer interval that no test lasts. A test gives it to
+// newPrefixHandler, newASHandler, newPeerHandler or newPeeringHandler when the
+// handler must write only when its batch is full, when the test says, or on
+// Stop.
 const longerThanAnyTest = time.Hour
 
 // prefixHandlerWithoutTimer returns a prefix handler that writes to store and
@@ -247,9 +248,11 @@ const longerThanAnyTest = time.Hour
 // NewPrefixHandler's flush goroutine, so the test flushes itself.
 func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
 	return &PrefixHandler{
-		db:     store,
-		logger: logger.New(),
-		table:  routetable.New(routetable.Params{}),
+		db:                   store,
+		logger:               logger.New(),
+		table:                routetable.New(routetable.Params{}),
+		batchMaxSize:         prefixBatchMaxSize,
+		tableChangesPerWrite: maxTableChangesPerWrite,
 	}
 }
 

internal/routewatch/prefixhandler_table_internal_test.go

@@ -30,11 +30,12 @@ const (
 
 	// firstManyPrefix is the first prefix announceManyPrefixes announces.
 	firstManyPrefix = "10.0.0.0/32"
-)
 
-// tableWriteTimeout is how long a test waits for the flush timer to write the
-// routing table's changes: a few ticks of it.
-const tableWriteTimeout = 5 * prefixBatchTimeout
+	// testMaxTableChangesPerWrite is a limit of one write of the routing
+	// table's changes that a test exceeds quickly, in place of
+	// maxTableChangesPerWrite.
+	testMaxTableChangesPerWrite = 10
+)
 
 // tableCase is one test of the routing table in memory: the messages handed to
 // the prefix handler, and for each prefix the peers that carry it with each
@@ -276,9 +277,9 @@ func TestPrefixHandlerWritesTheTableChangesEachSecond(t *testing.T) {
 				t.Fatalf("wrote %+v, want the pair of %s and AS%d with "+
 					"removed %t", changes, prefix, testASNB, step.wantRemoved)
 			}
-		case <-time.After(tableWriteTimeout):
+		case <-time.After(stepTimeout):
 			t.Fatalf("the routing table's change was not written within %s",
-				tableWriteTimeout)
+				stepTimeout)
 		}
 	}
 }
@@ -375,25 +376,32 @@ func (s *tableWriteSizesStore) WriteRouteChanges(
 // takes, then flushes three times. The first write must hold as many pairs as
 // one write takes, and the peer. The second must hold the other pairs, and the
 // third flush must write nothing. The database must then hold the table in
-// memory.
+// memory. The handler's limit is small, and its timer does not fire.
 func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
 	t.Parallel()
 
 	const (
-		pairs   = maxTableChangesPerWrite * 3 / 2
+		pairs   = testMaxTableChangesPerWrite * 3 / 2
 		flushes = 3
 	)
 
 	db := newTestDatabase(t)
 	store := &tableWriteSizesStore{Database: db}
-	h := prefixHandlerWithoutTimer(store)
-	announceManyPrefixes(h.table, pairs)
+	table := routetable.New(routetable.Params{})
+	announceManyPrefixes(table, pairs)
+
+	h := newPrefixHandler(store, logger.New(), table, longerThanAnyTest,
+		prefixBatchMaxSize, testMaxTableChangesPerWrite)
+
+	defer stopInTime(t, h)
 
 	for range flushes {
 		h.flushBatch()
 	}
 
-	wantChanges := []int{maxTableChangesPerWrite, pairs - maxTableChangesPerWrite}
+	wantChanges := []int{
+		testMaxTableChangesPerWrite, pairs - testMaxTableChangesPerWrite,
+	}
 	wantNewPeers := []int{1, 0}
 
 	if !slices.Equal(store.changes, wantChanges) ||
@@ -415,11 +423,11 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
 // takes. Its last flush writes one write's worth, so Stop must log the pairs
 // that no write took, and must not say that the stop deadline passed. The
 // flush timer must not fire, or its writes could leave no pairs for Stop to
-// log.
+// log. The handler's limit of one write is small.
 func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
 	t.Parallel()
 
-	const pairs = maxTableChangesPerWrite + 1
+	const pairs = testMaxTableChangesPerWrite + 1
 
 	var logs bytes.Buffer
 
@@ -427,7 +435,8 @@ 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, longerThanAnyTest,
+		prefixBatchMaxSize, testMaxTableChangesPerWrite)
 	h.Stop(t.Context())
 
 	written := 0