Start no AS, peer or peering write after the stop deadline (closes #189)

A write can run past the stop deadline while a full batch's wake-up or a
tick waits beside Stop's context. The select picked one at random, so about
half the time the AS, peer or peering handler started another write after
the deadline. Each loop now takes a waiting stop first, as the prefix
handler's does. The prefix, AS and peer handlers' flush loops were
near-copies, so they are now one function, flushLoop. A new test for each
handler fails about half its runs on the old code.

The AS and peer handler tests give the flush timer an interval longer than
the test. A timer flush during the fill can no longer hang
TestPeerHandlerWaitsAtTheMaximumBatchSize.

Model: opus-5-5

internal/routewatch/batchwrite_internal_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"fmt"
 	"log/slog"
 	"slices"
 	"sync"
@@ -168,9 +169,10 @@ func returnsInTime(f func()) bool {
 	}
 }
 
-// startBlockedWrite hands h the messages in fill, which fill its batch so that
-// the handler writes it to store, and waits until that write starts. store
-// holds it until the test calls releaseWrites.
+// 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
+// until the test calls releaseWrites.
 func startBlockedWrite(
 	t *testing.T,
 	store *blockingStore,
@@ -193,7 +195,7 @@ func startBlockedWrite(
 	select {
 	case <-store.writeStarted:
 	case <-time.After(handleMessageTimeout):
-		t.Fatal("the handler did not write its full batch")
+		t.Fatal("the handler did not start to write the messages")
 	}
 }
 
@@ -273,7 +275,7 @@ func handleAtMaxBatchSize(
 }
 
 // stopWhileAWriteIsBlocked starts a blocked write with the messages in fill,
-// hands h the message extra, and stops h with a deadline that passes while
+// hands h the messages in extra, and stops h with a deadline that passes while
 // store holds the write. Stop must return all the same. Then it lets the write
 // finish and waits until done, closed when h's flush goroutine returns. Past
 // the deadline that goroutine skips its last write, so extra is not written,
@@ -287,7 +289,7 @@ func stopWhileAWriteIsBlocked(
 	},
 	done <-chan struct{},
 	fill []*ristypes.RISMessage,
-	extra *ristypes.RISMessage,
+	extra ...*ristypes.RISMessage,
 ) {
 	t.Helper()
 
@@ -296,7 +298,10 @@ func stopWhileAWriteIsBlocked(
 	defer store.releaseWrites()
 
 	startBlockedWrite(t, store, h, fill)
-	h.HandleMessage(extra)
+
+	for _, msg := range extra {
+		h.HandleMessage(msg)
+	}
 
 	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
 	defer cancel()
@@ -428,7 +433,7 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	var logs bytes.Buffer
 
 	store := newBlockingStore()
-	h := NewASHandler(store, newErrorLogger(&logs))
+	h := newASHandler(store, newErrorLogger(&logs), longerThanAnyTest)
 	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
 		[]*ristypes.RISMessage{{Path: fill}},
@@ -447,6 +452,29 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	}
 }
 
+// TestASHandlerSkipsAFullBatchAfterTheStopDeadline hands the AS handler a full
+// batch while a write is held, so the batch's wake-up of the flush goroutine
+// waits. The stop deadline passes meanwhile. When the write ends, the flush
+// goroutine must stop instead of writing the batch.
+func TestASHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	store := newBlockingStore()
+	h := newASHandler(store, logger.New(), longerThanAnyTest)
+	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
+	full := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)}
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		[]*ristypes.RISMessage{{Path: fill}},
+		&ristypes.RISMessage{Path: full})
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+
+	if store.asns[testASNB] {
+		t.Errorf("AS%d was written after the stop deadline", testASNB)
+	}
+}
+
 func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	t.Parallel()
 
@@ -455,7 +483,7 @@ func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	var logs bytes.Buffer
 
 	store := newBlockingStore()
-	h := NewPeerHandler(store, newErrorLogger(&logs))
+	h := newPeerHandler(store, newErrorLogger(&logs), longerThanAnyTest)
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
 		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
 		&ristypes.RISMessage{Peer: otherPeerIP})
@@ -473,6 +501,30 @@ func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	}
 }
 
+// TestPeerHandlerSkipsAFullBatchAfterTheStopDeadline hands the peer handler a
+// full batch while a write is held, so the batch's wake-up of the flush
+// goroutine waits. The stop deadline passes meanwhile. When the write ends, the
+// flush goroutine must stop instead of writing the batch.
+func TestPeerHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	const otherPeerIP = "2001:db8:ffff::2"
+
+	store := newBlockingStore()
+	h := newPeerHandler(store, logger.New(), longerThanAnyTest)
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
+		slices.Repeat([]*ristypes.RISMessage{{Peer: otherPeerIP}},
+			peerBatchSize)...)
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+
+	if store.peers[otherPeerIP] {
+		t.Errorf("peer %s was written after the stop deadline", otherPeerIP)
+	}
+}
+
 // TestPeeringHandlerStopCountsTheWriteInProgress hands a running peering
 // handler one AS path and stops it with a deadline that passes while store
 // holds Stop's last write, of that path's peering. Stop must log that path as
@@ -512,6 +564,35 @@ func TestPeeringHandlerStopCountsTheWriteInProgress(t *testing.T) {
 	}
 }
 
+// TestPeeringHandlerStartsNoWriteAfterTheStopDeadline has the peering handler's
+// timer start a write that store holds, and hands the handler one more AS path
+// meanwhile. The stop deadline passes, and a tick of the timer waits. When the
+// write ends, the processing goroutine must stop instead of writing the path.
+func TestPeeringHandlerStartsNoWriteAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	// The timer fires many times while Stop waits for its deadline.
+	const processInterval = 10 * time.Millisecond
+
+	store := newBlockingStore()
+	h := newPeeringHandler(store, logger.New(), processInterval)
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		[]*ristypes.RISMessage{
+			{Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}},
+		},
+		&ristypes.RISMessage{
+			Path: ristypes.ASPath{Sequence: []int{testASNC, testASND}},
+		})
+
+	store.mockStore.mu.Lock()
+	defer store.mockStore.mu.Unlock()
+
+	if store.Peerings[fmt.Sprintf("%d_%d", testASNC, testASND)] {
+		t.Errorf("the peering of AS%d and AS%d was written after the stop "+
+			"deadline", testASNC, testASND)
+	}
+}
+
 // prefixAnnouncement returns a message in which testPeerIP announces prefixes
 // with origin AS testASNA.
 func prefixAnnouncement(prefixes []string) *ristypes.RISMessage {
@@ -549,7 +630,7 @@ func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
 	t.Parallel()
 
 	store := newBlockingStore()
-	h := NewASHandler(store, logger.New())
+	h := newASHandler(store, logger.New(), longerThanAnyTest)
 	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
 	handleWhileAWriteIsBlocked(t, store, h,
 		[]*ristypes.RISMessage{{Path: fill}},
@@ -570,7 +651,7 @@ func TestPeerHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
 	const otherPeerIP = "2001:db8:ffff::2"
 
 	store := newBlockingStore()
-	h := NewPeerHandler(store, logger.New())
+	h := newPeerHandler(store, logger.New(), longerThanAnyTest)
 	handleWhileAWriteIsBlocked(t, store, h,
 		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
 		&ristypes.RISMessage{Peer: otherPeerIP})
@@ -611,7 +692,7 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	t.Parallel()
 
 	store := newBlockingStore()
-	h := NewASHandler(store, logger.New())
+	h := newASHandler(store, logger.New(), longerThanAnyTest)
 	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
 	toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchMaxSize)}
 	handleAtMaxBatchSize(t, store, h,
@@ -628,13 +709,17 @@ 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.
 func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	t.Parallel()
 
 	const otherPeerIP = "2001:db8:ffff::2"
 
 	store := newBlockingStore()
-	h := NewPeerHandler(store, logger.New())
+	h := newPeerHandler(store, logger.New(), longerThanAnyTest)
 	message := []*ristypes.RISMessage{{Peer: testPeerIP}}
 	handleAtMaxBatchSize(t, store, h,
 		slices.Repeat(message, peerBatchSize),