TODO.md

@@ -33,6 +33,9 @@ tested.
 
 # Completed Steps
 
+- 2026-10-27: once the stop deadline has passed, the prefix handler starts no
+  flush. A batch that filled while a write ran past the deadline could still be
+  written after it (closes #188)
 - 2026-10-27: the prefix handler applies each update to the routing table in
   memory, and writes the table's changes to the database every second. A write
   takes at most 20,000 changed pairs, and the rest follow in the next writes. A

internal/routewatch/batchwrite_internal_test.go

@@ -317,7 +317,9 @@ func stopWhileAWriteIsBlocked(
 
 // TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline also checks the
 // routing table's changes: the flush whose delete is held wrote those of the
-// fill, so Stop must log the one of the extra prefix as not written.
+// fill, so Stop must log the one of the extra prefix as not written. The flush
+// timer must not fire. A flush it started before the fill would be the one
+// held, and it would not have written the fill's change.
 func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	t.Parallel()
 
@@ -326,8 +328,8 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	var logs bytes.Buffer
 
 	store := newBlockingStore()
-	h := NewPrefixHandler(store, newErrorLogger(&logs),
-		routetable.New(routetable.Params{}))
+	h := newPrefixHandler(store, newErrorLogger(&logs),
+		routetable.New(routetable.Params{}), longerThanAnyTest)
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	// The peer's session goes down first, so the write held is the delete of
 	// its routes, and Stop must log that peer down too.
@@ -351,6 +353,31 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	}
 }
 
+// TestPrefixHandlerSkipsAFullBatchAfterTheStopDeadline hands the prefix 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 TestPrefixHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	const extraPrefix = "198.51.100.0/24"
+
+	store := newBlockingStore()
+	h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))
+	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
+	full := slices.Repeat([]string{extraPrefix}, prefixBatchSize)
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
+		prefixAnnouncement(full))
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+
+	if store.prefixes[extraPrefix] {
+		t.Errorf("%s was written after the stop deadline", extraPrefix)
+	}
+}
+
 // tableWriteBlockingStore is a blockingStore whose writes of the routing
 // table's changes also wait until the test calls releaseWrites.
 type tableWriteBlockingStore struct {

internal/routewatch/prefixhandler.go

@@ -130,6 +130,16 @@ func (u prefixUpdate) announcesRoute() bool {
 // the updates to table, the routing table read from db at startup.
 func NewPrefixHandler(
 	db database.Store, logger *logger.Logger, table *routetable.Table,
+) *PrefixHandler {
+	return newPrefixHandler(db, logger, table, prefixBatchTimeout)
+}
+
+// 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.
+func newPrefixHandler(
+	db database.Store, logger *logger.Logger, table *routetable.Table,
+	batchTimeout time.Duration,
 ) *PrefixHandler {
 	h := &PrefixHandler{
 		db:          db,
@@ -144,7 +154,7 @@ func NewPrefixHandler(
 	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
-	go h.flushLoop()
+	go h.flushLoop(batchTimeout)
 
 	return h
 }
@@ -315,17 +325,32 @@ func (h *PrefixHandler) peerSessionDown(peer string) {
 }
 
 // flushLoop runs in a goroutine, the only one that writes the batch: it
-// flushes when the batch is full, periodically, and once more on Stop.
-func (h *PrefixHandler) flushLoop() {
+// flushes when the batch is full, every batchTimeout, and once more on Stop.
+func (h *PrefixHandler) flushLoop(batchTimeout time.Duration) {
 	defer close(h.done)
 
-	ticker := time.NewTicker(prefixBatchTimeout)
+	ticker := time.NewTicker(batchTimeout)
 	defer ticker.Stop()
 
 	for {
+		// Stop goes first. 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
+		// below picks one of them at random, but past the deadline no flush
+		// may start.
+		select {
+		case ctx := <-h.stopCh:
+			// Final flush, unless the stop deadline has passed
+			if ctx.Err() == nil {
+				h.flushBatch()
+			}
+
+			return
+		default:
+		}
+
 		select {
 		case <-ticker.C:
-			if time.Since(h.lastFlush) >= prefixBatchTimeout {
+			if time.Since(h.lastFlush) >= batchTimeout {
 				h.flushBatch()
 			}
 		case <-h.batchFullCh:

internal/routewatch/prefixhandler_internal_test.go

@@ -980,6 +980,11 @@ func TestPrefixHandlerStopLogsNothingAfterADeleteThatSucceeds(t *testing.T) {
 	}
 }
 
+// longerThanAnyTest is a flush timer interval that no test lasts. A test whose
+// prefix handler must flush only when its batch is full and on Stop gives it to
+// newPrefixHandler.
+const longerThanAnyTest = time.Hour
+
 // prefixHandlerWithoutTimer returns a prefix handler that writes to store and
 // applies the updates to an empty routing table. It is built without
 // NewPrefixHandler's flush goroutine, so the test flushes itself.

internal/routewatch/prefixhandler_table_internal_test.go

@@ -341,15 +341,14 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
 }
 
 // TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten stops a prefix handler,
-// with no deadline, whose routing table holds more changed pairs than two
-// writes take. Its last flush writes one write's worth, so Stop must log the
-// pairs that no write took.
+// with no deadline, whose routing table holds more changed pairs than one write
+// takes. Its last flush writes one write's worth, so Stop must log the pairs
+// that no write took. The flush timer must not fire, or its writes could leave
+// no pairs for Stop to log.
 func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
 	t.Parallel()
 
-	// More than two writes take, so that pairs are left even when the flush
-	// timer writes once before Stop.
-	const pairs = 2*maxTableChangesPerWrite + 1
+	const pairs = maxTableChangesPerWrite + 1
 
 	var logs bytes.Buffer
 
@@ -357,7 +356,7 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
 	table := routetable.New(routetable.Params{})
 	announceManyPrefixes(table, pairs)
 
-	h := NewPrefixHandler(store, newErrorLogger(&logs), table)
+	h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest)
 	h.Stop(t.Context())
 
 	written := 0