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
-  messages until that peer's `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. At that number, `HandleMessage` waits until one of them is applied.
-  `PeerDown` lets the table's lock go after every 10,000 prefixes, so the IP
-  lookup and the pages no longer wait for the whole visit. When `Stop` runs past
-  its deadline, it logs how many held-back messages were not applied (closes
-  #216)
+  time, and goes on taking messages. `PeerDown` holds the table's lock for
+  10,000 prefixes at a time, then lets the lock go for as long again. The
+  handler holds back the down peer's later messages until that peer's `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. At that number,
+  `HandleMessage` waits until one of them is applied. When `Stop` runs past 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,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
-// visits every prefix in the table. It lets the table's lock go after every
-// walkBatch prefixes, so that waiting reads and updates get in. An
-// announcement by the peer made meanwhile may be dropped too. So the caller
-// applies the peer's later updates only once PeerDown has returned.
+// set, and every route of the peer is dropped. When the peer carries any
+// prefix, PeerDown visits every prefix in the table, in batches of walkBatch.
+// After each batch, it lets the table's lock go for as long as it held the
+// lock for the batch. During a walk over many batches, reads and updates can
+// 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
-// PeerDown returns. Then it applies them in order, one at a time, and takes
-// new messages in between. It holds back at most maxHeldBackMessages
-// messages.
+// meanwhile. PeerDown lets the table's lock go about half of the time, so the
+// handler applies other peers' messages while PeerDown runs. The handler holds
+// back the down peer's later messages until PeerDown returns. Then the handler
+// 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
-// takes mu for one message at a time, so HandleMessage takes new messages in
-// between. The peer's new messages join the end of the held-back ones. A
-// held-back message that says the peer's session went down again starts a new
-// peerDown, which applies the rest.
+// running. Then peerDown applies the messages held back from the peer, in
+// order. It takes mu for one message at a time, so HandleMessage takes new
+// messages in between. The peer's new messages join the end of the held-back
+// ones. A held-back message that says the peer's session went down again
+// 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
-// session going down again, and a second announcement. Once the PeerDowns
-// have returned, the down peer must carry only the prefix of the second
-// announcement.
+// at once. It holds back these messages of the down peer, in this order:
+//   - an announcement;
+//   - its session going down again;
+//   - 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
-// takes a millisecond. While the handler applies the held-back messages, it
-// must take another peer's announcement.
+// not parse, so applying it logs an error. Each log write takes a
+// millisecond. While the handler applies the held-back messages, it must take
+// 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
-// checks its routing table once the PeerDowns it started have returned, with
-// no flush.
+// checkTableCases hands the messages of each case to a new prefix handler. It
+// checks the handler's routing table once the handler's PeerDowns have
+// returned, with no flush.
 func checkTableCases(t *testing.T, tests []tableCase) {
 	t.Helper()