Keep taking messages while a peer's session goes down (closes #216)

The table's PeerDown visits every prefix under the write lock, which
takes over a second on a full table. The prefix handler called it
inline, so its queue filled and the streamer dropped updates.

PeerDown now lets the lock go after every 10,000 prefixes. The handler
runs it on a separate goroutine, one peer at a time. It holds back the
down peer's later messages until that PeerDown returns. Then it applies
them in order, one at a time, and takes new messages in between. It
holds back at most 20,000 messages. Stop waits for all of this before
its last flush.

Model: opus-5-5

TODO.md

@@ -37,13 +37,14 @@ tested.
 
 - 2026-11-02: a peer's session going down no longer stops the prefix handler.
   The handler runs the table's `PeerDown` on a separate goroutine, one peer at a
-  time, and goes on taking messages. `PeerDown` holds the table's lock for
+  time, and goes on taking messages. It holds back the down peer's later
-  10,000 prefixes at a time, then lets the lock go for as long again. The
+  messages until that peer's `PeerDown` returns. Then it applies them in order,
-  handler holds back the down peer's later messages until that peer's `PeerDown`
+  one at a time, and takes new messages in between. It holds back at most 20,000
-  returns. Then the handler applies them in order, one at a time, and takes new
+  messages. At that number, `HandleMessage` waits until one of them is applied.
-  messages in between. It holds back at most 20,000 messages. At that number,
+  `PeerDown` lets the table's lock go after every 10,000 prefixes, so the IP
-  `HandleMessage` waits until one of them is applied. When `Stop` runs past its
+  lookup and the pages no longer wait for the whole visit. When `Stop` runs past
-  deadline, it logs how many held-back messages were not applied (closes #216)
+  its deadline, it logs how many held-back messages were not applied (closes
+  #216)
 - 2026-11-02: the AS handler never waits for a database write. Its batch holds
   one entry per AS, with the earliest and the latest time of its updates. So the
   batch no longer grows with the feed's rate. It used to hold one entry for each

internal/routetable/routetable.go

@@ -206,13 +206,11 @@ func (t *Table) Withdraw(prefix netip.Prefix, peer string) {
 }
 
 // PeerDown applies a peer's session going down: the peer leaves every peer
-// set, and every route of the peer is dropped. When the peer carries any
+// set, and every route of it is dropped. When the peer carries any prefix, it
-// prefix, PeerDown visits every prefix in the table, in batches of walkBatch.
+// visits every prefix in the table. It lets the table's lock go after every
-// After each batch, it lets the table's lock go for as long as it held the
+// walkBatch prefixes, so that waiting reads and updates get in. An
-// lock for the batch. During a walk over many batches, reads and updates can
+// announcement by the peer made meanwhile may be dropped too. So the caller
-// then take the lock about half of the time. An announcement by the peer
+// applies the peer's later updates only once PeerDown has returned.
-// applied during the walk may be dropped too. So the caller applies the peer's
-// later updates only once PeerDown has returned.
 func (t *Table) PeerDown(peer string) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
@@ -223,7 +221,6 @@ func (t *Table) PeerDown(peer string) {
 	}
 
 	visited := 0
-	batchStart := time.Now()
 
 	for _, f := range []*family{&t.ipv4, &t.ipv6} {
 		for _, prefixes := range f.byLength {
@@ -237,13 +234,8 @@ func (t *Table) PeerDown(peer string) {
 				// the range over it.
 				visited++
 				if visited%walkBatch == 0 {
-					held := time.Since(batchStart)
-
 					t.mu.Unlock()
-					time.Sleep(held)
 					t.mu.Lock()
-
-					batchStart = time.Now()
 				}
 			}
 		}

internal/routewatch/prefixhandler.go

@@ -59,13 +59,12 @@ const (
 // whose BGP session goes down leaves every prefix.
 //
 // The table's PeerDown applies a session going down. It visits every prefix,
-// which takes seconds on a full table. So the handler runs PeerDown on
+// which takes over a second on a full table. So the handler runs PeerDown on
 // a separate goroutine, one peer at a time, and goes on taking messages
-// meanwhile. PeerDown lets the table's lock go about half of the time, so the
+// meanwhile. The handler holds back the down peer's later messages until
-// handler applies other peers' messages while PeerDown runs. The handler holds
+// PeerDown returns. Then it applies them in order, one at a time, and takes
-// back the down peer's later messages until PeerDown returns. Then the handler
+// new messages in between. It holds back at most maxHeldBackMessages
-// applies them in order, one at a time, and takes new messages in between. It
+// messages.
-// holds back at most maxHeldBackMessages messages.
 //
 // Each flush writes the table's changes to the database. It also stores each
 // prefix the updates name with the time it was first seen.
@@ -386,11 +385,11 @@ func (h *PrefixHandler) startPeerDown(peer string) {
 }
 
 // peerDown runs the table's PeerDown for peer, once no other PeerDown is
-// running. Then peerDown applies the messages held back from the peer, in
+// running. Then it applies the messages held back from the peer, in order. It
-// order. It takes mu for one message at a time, so HandleMessage takes new
+// takes mu for one message at a time, so HandleMessage takes new messages in
-// messages in between. The peer's new messages join the end of the held-back
+// between. The peer's new messages join the end of the held-back ones. A
-// ones. A held-back message that says the peer's session went down again
+// held-back message that says the peer's session went down again starts a new
-// starts a new peerDown, which applies the rest.
+// peerDown, which applies the rest.
 func (h *PrefixHandler) peerDown(peer string) {
 	defer h.peerDowns.Done()
 

internal/routewatch/prefixhandler_table_internal_test.go

@@ -7,7 +7,6 @@ import (
 	"log/slog"
 	"net/netip"
 	"reflect"
-	"runtime"
 	"slices"
 	"strings"
 	"sync"
@@ -106,13 +105,10 @@ func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) {
 // TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeerDown of a
 // peer whose session went down, as another peer's long PeerDown would. The
 // handler must go on taking messages. It applies another peer's announcement
-// at once. It holds back these messages of the down peer, in this order:
+// at once. It holds back the down peer's own messages: an announcement, its
-//   - an announcement;
+// session going down again, and a second announcement. Once the PeerDowns
-//   - its session going down again;
+// have returned, the down peer must carry only the prefix of the second
-//   - a second announcement.
+// announcement.
-//
-// Once the PeerDowns have returned, the down peer must carry only the prefix
-// of the second announcement.
 func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
 	t.Parallel()
 
@@ -158,69 +154,11 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
 	})
 }
 
-// TestPrefixHandlerAppliesMessagesWhilePeerDownRuns gives a peer 50,000
-// prefixes and hands the prefix handler that peer's session going down. The
-// table's PeerDown then visits the prefixes in five batches. Once PeerDown has
-// begun, the handler is handed 1,000 announcements of another peer. The
-// handler must apply them all before PeerDown ends, while the down peer still
-// carries some of its prefixes.
-func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
-	t.Parallel()
-
-	const (
-		// downPeerPrefixes is five times the 10,000 prefixes PeerDown visits
-		// in one hold of the table's lock.
-		downPeerPrefixes = 50000
-		// otherPeerMessages is far more messages than the handler applies
-		// while PeerDown runs, if PeerDown takes the lock back at once after
-		// each batch. The handler then applies only a few for each batch.
-		otherPeerMessages = 1000
-		// firstOtherAddr is the address of the other peer's first prefix.
-		firstOtherAddr = "2001:db8:2::"
-	)
-
-	h := prefixHandlerWithoutTimer(newMockStore())
-	announceManyPrefixes(h.table, downPeerPrefixes)
-
-	h.HandleMessage(sessionDown())
-
-	// Wait until PeerDown has begun: the down peer has left some of its
-	// prefixes. The down peer's prefixes are IPv4 and the other peer's are
-	// IPv6, so the counts tell them apart.
-	for h.table.Counts().IPv4Routes == downPeerPrefixes {
-		runtime.Gosched()
-	}
-
-	// Each announcement names the /128 of the next address.
-	addr := netip.MustParseAddr(firstOtherAddr)
-	for range otherPeerMessages {
-		h.HandleMessage(peerAnnouncement(secondPeerIP,
-			netip.PrefixFrom(addr, addr.BitLen()).String()))
-
-		addr = addr.Next()
-	}
-
-	counts := h.table.Counts()
-	if counts.IPv6Routes != otherPeerMessages {
-		t.Errorf("the other peer carries %d prefixes, want %d",
-			counts.IPv6Routes, otherPeerMessages)
-	}
-
-	if counts.IPv4Routes == 0 {
-		t.Errorf("PeerDown ended before the handler applied the other peer's "+
-			"%d announcements", otherPeerMessages)
-	}
-
-	if !returnsInTime(h.peerDowns.Wait) {
-		t.Fatalf("the PeerDown did not return within %s", stepTimeout)
-	}
-}
-
 // TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many
 // messages of a peer whose session went down. Each names a prefix that does
-// not parse, so applying it logs an error. Each log write takes a
+// not parse, so the handler logs an error as it applies it. Each log write
-// millisecond. While the handler applies the held-back messages, it must take
+// takes a millisecond. While the handler applies the held-back messages, it
-// another peer's announcement.
+// must take another peer's announcement.
 func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
 	t.Parallel()
 
@@ -423,9 +361,9 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
 	}
 }
 
-// checkTableCases hands the messages of each case to a new prefix handler. It
+// checkTableCases hands the messages of each case to a new prefix handler and
-// checks the handler's routing table once the handler's PeerDowns have
+// checks its routing table once the PeerDowns it started have returned, with
-// returned, with no flush.
+// no flush.
 func checkTableCases(t *testing.T, tests []tableCase) {
 	t.Helper()