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 holds the lock for 10,000 prefixes at a time, then lets
the lock go for as long again. The handler runs PeerDown on a separate
goroutine, one peer at a time. The handler holds back the down peer's
later messages until that PeerDown returns. Then the handler 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,14 +37,13 @@ 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. It holds back the down peer's later
+  time, and goes on taking messages. `PeerDown` holds the table's lock for
-  messages until that peer's `PeerDown` returns. Then it applies them in order,
+  10,000 prefixes at a time, then lets the lock go for as long again. The
-  one at a time, and takes new messages in between. It holds back at most 20,000
+  handler holds back the down peer's later messages until that peer's `PeerDown`
-  messages. At that number, `HandleMessage` waits until one of them is applied.
+  returns. Then the handler applies them in order, one at a time, and takes new
-  `PeerDown` lets the table's lock go after every 10,000 prefixes, so the IP
+  messages in between. It holds back at most 20,000 messages. At that number,
-  lookup and the pages no longer wait for the whole visit. When `Stop` runs past
+  `HandleMessage` waits until one of them is applied. When `Stop` runs past its
-  its deadline, it logs how many held-back messages were not applied (closes
+  deadline, it logs how many held-back messages were not applied (closes #216)
-  #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,11 +206,13 @@ 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 it is dropped. When the peer carries any prefix, it
+// set, and every route of the peer is dropped. When the peer carries any
-// visits every prefix in the table. It lets the table's lock go after every
+// prefix, PeerDown visits every prefix in the table, in batches of walkBatch.
-// walkBatch prefixes, so that waiting reads and updates get in. An
+// After each batch, it lets the table's lock go for as long as it held the
-// announcement by the peer made meanwhile may be dropped too. So the caller
+// lock for the batch. During a walk over many batches, reads and updates can
-// applies the peer's later updates only once PeerDown has returned.
+// then take the lock about half of the time. An announcement by the peer
+// 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()
@@ -221,6 +223,7 @@ func (t *Table) PeerDown(peer string) {
 	}
 
 	visited := 0
+	batchStart := time.Now()
 
 	for _, f := range []*family{&t.ipv4, &t.ipv6} {
 		for _, prefixes := range f.byLength {
@@ -234,8 +237,13 @@ 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,12 +59,13 @@ const (
 // whose BGP session goes down leaves every prefix.
 //
 // The table's PeerDown applies a session going down. It visits every prefix,
-// which takes over a second on a full table. So the handler runs PeerDown on
+// which takes seconds on a full table. So the handler runs PeerDown on
 // a separate goroutine, one peer at a time, and goes on taking messages
-// meanwhile. The handler holds back the down peer's later messages until
+// meanwhile. PeerDown lets the table's lock go about half of the time, so the
-// PeerDown returns. Then it applies them in order, one at a time, and takes
+// handler applies other peers' messages while PeerDown runs. The handler holds
-// new messages in between. It holds back at most maxHeldBackMessages
+// back the down peer's later messages until PeerDown returns. Then the handler
-// messages.
+// applies them in order, one at a time, and takes new messages in between. It
+// 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.
@@ -385,11 +386,11 @@ func (h *PrefixHandler) startPeerDown(peer string) {
 }
 
 // peerDown runs the table's PeerDown for peer, once no other PeerDown is
-// running. Then it applies the messages held back from the peer, in order. It
+// running. Then peerDown applies the messages held back from the peer, in
-// takes mu for one message at a time, so HandleMessage takes new messages in
+// order. It takes mu for one message at a time, so HandleMessage takes new
-// between. The peer's new messages join the end of the held-back ones. A
+// messages in between. The peer's new messages join the end of the held-back
-// held-back message that says the peer's session went down again starts a new
+// ones. A held-back message that says the peer's session went down again
-// peerDown, which applies the rest.
+// starts a new peerDown, which applies the rest.
 func (h *PrefixHandler) peerDown(peer string) {
 	defer h.peerDowns.Done()
 

internal/routewatch/prefixhandler_table_internal_test.go

@@ -7,6 +7,7 @@ import (
 	"log/slog"
 	"net/netip"
 	"reflect"
+	"runtime"
 	"slices"
 	"strings"
 	"sync"
@@ -105,10 +106,13 @@ 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 the down peer's own messages: an announcement, its
+// at once. It holds back these messages of the down peer, in this order:
-// session going down again, and a second announcement. Once the PeerDowns
+//   - an announcement;
-// have returned, the down peer must carry only the prefix of the second
+//   - its session going down again;
-// announcement.
+//   - a second 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()
 
@@ -154,11 +158,69 @@ 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 the handler logs an error as it applies it. Each log write
+// not parse, so applying it logs an error. Each log write takes a
-// takes a millisecond. While the handler applies the held-back messages, it
+// millisecond. While the handler applies the held-back messages, it must take
-// must take another peer's announcement.
+// another peer's announcement.
 func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
 	t.Parallel()
 
@@ -361,9 +423,9 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
 	}
 }
 
-// checkTableCases hands the messages of each case to a new prefix handler and
+// checkTableCases hands the messages of each case to a new prefix handler. It
-// checks its routing table once the PeerDowns it started have returned, with
+// checks the handler's routing table once the handler's PeerDowns have
-// no flush.
+// returned, with no flush.
 func checkTableCases(t *testing.T, tests []tableCase) {
 	t.Helper()