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

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)
@@ -275,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,
@@ -288,13 +307,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 +339,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 +369,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 +389,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 +408,8 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 
 	store := newBlockingStore()
 	h := newPrefixHandler(store, newErrorLogger(&logs),
-		routetable.New(routetable.Params{}), longerThanAnyTest)
+		routetable.New(routetable.Params{}), longerThanAnyTest,
+		maxTableChangesPerWrite)
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
 		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
@@ -597,7 +626,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 +647,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}}},
@@ -714,9 +743,11 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	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, maxTableChangesPerWrite)
+	h.batchMaxSize = prefixBatchSize
 	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)},
@@ -736,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}},
@@ -755,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()
 
@@ -763,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()
@@ -796,8 +829,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)))
@@ -809,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)},
@@ -822,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})
 				}