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 goroutine of its own, one peer at a time, and holds back
the down peer's later messages until it returns, then applies them in
order. Stop waits for this before its last flush, and logs the
held-back messages it could not apply.

Model: opus-5-5

README.md

@@ -442,7 +442,9 @@ the routing table, and everything else from the database.
         - An announcement gives the peer a route to the prefix.
         - A withdrawal removes the peer's route. So does an announcement whose
           AS path is empty or holds AS 0 anywhere.
-        - A peer whose BGP session goes down leaves every prefix.
+        - A peer whose BGP session goes down leaves every prefix. The handler
+          holds back the peer's later updates until it has left them all, then
+          applies them in order.
 
         Every second it writes the changed prefix and origin AS pairs to the
         `routes` table. A write takes at most 20,000 pairs, and the rest follow

TODO.md

@@ -35,6 +35,13 @@ tested.
 
 # Completed Steps
 
+- 2026-11-02: a peer's session going down no longer stops the prefix handler.
+  The handler runs the table's `PeerDown` on a goroutine of its own, one peer at
+  a time, and goes on taking messages. It holds back the down peer's later
+  messages until its `PeerDown` returns, then applies them in order. `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. A stop past its deadline logs the
+  held-back messages it did not apply (closes #216)
 - 2026-10-27: the tests of `internal/routewatch` no longer fail on a loaded
   host. Five of them gave `HandleMessage` two seconds to apply a full prefix
   batch, which under the race detector can take longer. A test now waits up to

internal/routetable/overview.go

@@ -10,10 +10,6 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/ristypes"
 )
 
-// walkBatch is how many prefixes the walk over the whole table reads in one
-// hold of the read lock. The updates waiting for the lock get it in between.
-const walkBatch = 10000
-
 // Overview holds figures about the whole table. Working them out takes a walk
 // over every prefix, so a caller asks for them now and then and keeps them.
 type Overview struct {

internal/routetable/routetable.go

@@ -27,6 +27,11 @@ const (
 	// kept routes in order: each of its nodes holds up to 63 times.
 	keptTimesDegree = 32
 
+	// walkBatch is how many prefixes a walk over the whole table visits in
+	// one hold of the lock: Overview's walk and PeerDown's. The others
+	// waiting for the lock get it in between.
+	walkBatch = 10000
+
 	ipv4Bits = 32
 	ipv6Bits = 128
 
@@ -202,7 +207,10 @@ 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.
+// visits every prefix in the table. It lets the lock go after every walkBatch
+// prefixes, so the reads and updates waiting for it get in. An announcement of
+// the peer made meanwhile 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()
@@ -212,6 +220,8 @@ func (t *Table) PeerDown(peer string) {
 		return
 	}
 
+	visited := 0
+
 	for _, f := range []*family{&t.ipv4, &t.ipv6} {
 		for _, prefixes := range f.byLength {
 			for prefix, origins := range prefixes {
@@ -219,6 +229,14 @@ func (t *Table) PeerDown(peer string) {
 				if o != nil {
 					t.leave(prefix, o, number)
 				}
+
+				// Go allows the map to change before the next step of
+				// the range over it.
+				visited++
+				if visited%walkBatch == 0 {
+					t.mu.Unlock()
+					t.mu.Lock()
+				}
 			}
 		}
 	}

internal/routetable/routetable_internal_test.go

@@ -0,0 +1,69 @@
+package routetable
+
+import (
+	"net/netip"
+	"runtime"
+	"testing"
+	"time"
+)
+
+// TestPeerDownLetsReadsIn reads the table's counts again and again while a peer
+// that carries more than two batches of prefixes goes down. PeerDown lets the
+// lock go after every walkBatch prefixes, so a read must get in while the peer
+// still carries some of its prefixes but no longer all.
+func TestPeerDownLetsReadsIn(t *testing.T) {
+	t.Parallel()
+
+	const (
+		asn  = 1000
+		peer = "192.0.2.1"
+		// prefixCount is more prefixes than two of walkBatch.
+		prefixCount = 2*walkBatch + 1
+	)
+
+	routeTime := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
+	table := New(Params{})
+	addr := netip.MustParseAddr("1.0.0.0")
+
+	for range prefixCount {
+		prefix := netip.PrefixFrom(addr, addr.BitLen())
+		table.Announce(prefix, asn, Route{Peer: peer, Time: routeTime})
+
+		addr = addr.Next()
+	}
+
+	// Hold a read lock until PeerDown waits for the lock, so that the first
+	// read below starts once PeerDown holds it. TryRLock fails once a writer
+	// waits.
+	table.mu.RLock()
+
+	peerDownDone := make(chan struct{})
+
+	go func() {
+		defer close(peerDownDone)
+
+		table.PeerDown(peer)
+	}()
+
+	for table.mu.TryRLock() {
+		table.mu.RUnlock()
+		runtime.Gosched()
+	}
+
+	table.mu.RUnlock()
+
+	for {
+		routes := table.Counts().IPv4Routes
+		if routes > 0 && routes < prefixCount {
+			break
+		}
+
+		select {
+		case <-peerDownDone:
+			t.Fatal("no read got in while PeerDown ran")
+		default:
+		}
+	}
+
+	<-peerDownDone
+}

internal/routewatch/batchwrite_internal_test.go

@@ -177,6 +177,7 @@ type stopLog struct {
 	Msg          string `json:"msg"`
 	Updates      int    `json:"updates"`
 	TableChanges int    `json:"table_changes"`
+	HeldMessages int    `json:"held_messages"`
 	Paths        int    `json:"paths"`
 }
 

internal/routewatch/prefixhandler.go

@@ -52,6 +52,12 @@ const (
 // does an announcement whose AS path is empty or holds AS 0 anywhere. A peer
 // 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 it on a
+// goroutine of its own, one peer at a time, and goes on taking messages
+// meanwhile. It holds back the peer's later messages until PeerDown returns,
+// then applies them in order.
+//
 // 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.
 type PrefixHandler struct {
@@ -67,7 +73,7 @@ type PrefixHandler struct {
 	// give a smaller one.
 	tableChangesPerWrite int
 
-	// Batching. mu guards batch, writing and writingTableChanges only: the
+	// Batching. mu guards batch, writing and writingTableChanges: the
 	// flush goroutine takes them under mu and writes them after releasing
 	// mu, so HandleMessage waits for a database write only while batch is at
 	// batchMaxSize.
@@ -87,6 +93,16 @@ type PrefixHandler struct {
 	// Stop does not wait for a write in progress to hand it over.
 	stopCh chan context.Context
 	done   chan struct{} // closed when the flush goroutine returns
+
+	// Peers whose session went down. peerDownMu lets one PeerDown run at a
+	// time, so the handler's own updates wait for the table's lock behind
+	// at most one. heldBack, under mu, holds each such peer's later
+	// messages, in order, until its PeerDown has returned and they are
+	// applied. A peer with no such messages yet has an empty entry.
+	// peerDowns counts the peerDown goroutines that have not returned.
+	peerDownMu sync.Mutex
+	heldBack   map[string][]*ristypes.RISMessage
+	peerDowns  sync.WaitGroup
 }
 
 // seenPrefix is what the batch keeps of an update: the prefix it names and its
@@ -150,6 +166,7 @@ func newPrefixHandler(
 		batchFullCh:          make(chan struct{}, 1),
 		stopCh:               make(chan context.Context, 1),
 		done:                 make(chan struct{}),
+		heldBack:             make(map[string][]*ristypes.RISMessage),
 	}
 	h.batchTaken.L = &h.mu
 
@@ -179,11 +196,86 @@ func (h *PrefixHandler) QueueCapacity() int {
 	return prefixHandlerQueueSize
 }
 
-// HandleMessage processes a message to track prefix information
+// HandleMessage processes a message to track prefix information. A message
+// from a peer whose PeerDown has not returned is held back until it has.
 func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	held, goingDown := h.heldBack[msg.Peer]
+	if goingDown {
+		h.heldBack[msg.Peer] = append(held, msg)
+
+		return
+	}
+
+	h.applyMessage(msg)
+}
+
+// Stop waits until ctx ends for every PeerDown the handler started, and for
+// the messages it held back meanwhile to be applied. Then it stops the flush
+// goroutine, which makes one last flush before it returns, and waits for it
+// until ctx ends. That flush writes at most tableChangesPerWrite of the
+// routing table's changes, and the prefixes of the updates the handler holds.
+// Once ctx has ended the flush goroutine skips that last flush, and a write
+// still running is lost when the process exits. A failed write of the table's
+// changes is not retried either. Stop logs how many updates and how many of
+// the table's changes are not written, and how many held-back messages are not
+// applied. The caller calls Stop once its last HandleMessage has returned.
+func (h *PrefixHandler) Stop(ctx context.Context) {
+	peerDownsDone := make(chan struct{})
+
+	go func() {
+		h.peerDowns.Wait()
+		close(peerDownsDone)
+	}()
+
+	select {
+	case <-peerDownsDone:
+	case <-ctx.Done():
+	}
+
+	h.stopCh <- ctx
+
+	select {
+	case <-h.done:
+	case <-ctx.Done():
+	}
+
+	h.mu.Lock()
+	unwritten := h.writing + len(h.batch)
+	tableChanges := h.writingTableChanges + h.table.ChangeCount()
+
+	heldMessages := 0
+	for _, held := range h.heldBack {
+		heldMessages += len(held)
+	}
+	h.mu.Unlock()
+
+	if unwritten == 0 && tableChanges == 0 && heldMessages == 0 {
+		return
+	}
+
+	if ctx.Err() != nil {
+		h.logger.Error("Stop deadline passed before the prefix handler wrote "+
+			"its updates", "updates", unwritten, "table_changes", tableChanges,
+			"held_messages", heldMessages)
+
+		return
+	}
+
+	h.logger.Error("The prefix handler stopped before it wrote everything",
+		"updates", unwritten, "table_changes", tableChanges,
+		"held_messages", heldMessages)
+}
+
+// applyMessage applies msg, with mu held. A peer's session going down starts
+// the peer's PeerDown. An update is applied to the routing table, and its
+// prefixes join the batch.
+func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
 	if msg.Type == ristypes.MessageTypeState {
 		if msg.State == ristypes.StateDown {
-			h.table.PeerDown(msg.Peer)
+			h.startPeerDown(msg.Peer)
 		}
 
 		return
@@ -198,9 +290,6 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// the path after the rest of it.
 	path := slices.Concat(msg.Path.Sequence, msg.Path.Set)
 
-	h.mu.Lock()
-	defer h.mu.Unlock()
-
 	// While the batch is at its maximum size, wait until the flush goroutine
 	// takes it. Meanwhile the handler takes nothing off its queue, which
 	// fills, and the streamer drops messages for it.
@@ -261,39 +350,50 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// Stop stops the flush goroutine, which makes one last flush before it returns,
-// and waits for it until ctx ends. That flush writes at most
-// tableChangesPerWrite of the routing table's changes, and the prefixes of
-// the updates the handler holds. Once ctx has ended the flush goroutine skips
-// that last flush, and a write still running is lost when the process exits. A
-// failed write of the table's changes is not retried either. Stop logs how many
-// updates and how many of the table's changes are not written.
-func (h *PrefixHandler) Stop(ctx context.Context) {
-	h.stopCh <- ctx
-
-	select {
-	case <-h.done:
-	case <-ctx.Done():
+// startPeerDown runs peerDown for peer on a goroutine of its own, with mu
+// held. The peer's messages are held back until it returns. Those held back
+// for an earlier PeerDown of the peer and not yet applied stay held.
+func (h *PrefixHandler) startPeerDown(peer string) {
+	_, goingDown := h.heldBack[peer]
+	if !goingDown {
+		h.heldBack[peer] = []*ristypes.RISMessage{}
 	}
 
+	h.peerDowns.Add(1)
+
+	go h.peerDown(peer)
+}
+
+// peerDown runs the table's PeerDown for peer once no other peer's runs. Then
+// it applies the messages held back from the peer, in order. When the peer's
+// session went down again among them, the PeerDown that starts applies the
+// rest.
+func (h *PrefixHandler) peerDown(peer string) {
+	defer h.peerDowns.Done()
+
+	h.peerDownMu.Lock()
+	h.table.PeerDown(peer)
+	h.peerDownMu.Unlock()
+
 	h.mu.Lock()
-	unwritten := h.writing + len(h.batch)
-	tableChanges := h.writingTableChanges + h.table.ChangeCount()
-	h.mu.Unlock()
+	defer h.mu.Unlock()
 
-	if unwritten == 0 && tableChanges == 0 {
-		return
+	// applyMessage lets mu go while it waits for the batch. HandleMessage
+	// then adds the peer's new messages after these.
+	for len(h.heldBack[peer]) > 0 {
+		msg := h.heldBack[peer][0]
+		h.heldBack[peer] = h.heldBack[peer][1:]
+		h.applyMessage(msg)
+
+		if msg.Type == ristypes.MessageTypeState &&
+			msg.State == ristypes.StateDown {
+			// applyMessage started the next PeerDown, which applies the
+			// rest.
+			return
+		}
 	}
 
-	if ctx.Err() != nil {
-		h.logger.Error("Stop deadline passed before the prefix handler wrote "+
-			"its updates", "updates", unwritten, "table_changes", tableChanges)
-
-		return
-	}
-
-	h.logger.Error("The prefix handler stopped before it wrote everything",
-		"updates", unwritten, "table_changes", tableChanges)
+	delete(h.heldBack, peer)
 }
 
 // takeBatch returns the pending updates and leaves an empty batch in their

internal/routewatch/prefixhandler_internal_test.go

@@ -237,6 +237,36 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
 	}
 }
 
+// TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a
+// deadline that has passed, while a peer's PeerDown is held up and one message
+// of the peer is held back. Stop must log that message as not applied.
+func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
+	t.Parallel()
+
+	var logs bytes.Buffer
+
+	h := newPrefixHandler(newMockStore(), newErrorLogger(&logs),
+		routetable.New(routetable.Params{}), longerThanAnyTest,
+		maxTableChangesPerWrite)
+
+	h.peerDownMu.Lock()
+	defer h.peerDownMu.Unlock()
+
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
+	h.HandleMessage(sessionDown())
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV6Prefix))
+
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	h.Stop(ctx)
+
+	logged := lastLogged(t, &logs)
+	if logged.HeldMessages != 1 {
+		t.Errorf("Stop logged %d held messages not applied, want 1",
+			logged.HeldMessages)
+	}
+}
+
 // longerThanAnyTest is a timer interval that no test lasts. A test gives it to
 // newPrefixHandler, newASHandler, newPeerHandler or newPeeringHandler when the
 // handler must write only when its batch is full, when the test says, or on
@@ -253,6 +283,7 @@ func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
 		table:                routetable.New(routetable.Params{}),
 		batchMaxSize:         prefixBatchMaxSize,
 		tableChangesPerWrite: maxTableChangesPerWrite,
+		heldBack:             make(map[string][]*ristypes.RISMessage),
 	}
 }
 

internal/routewatch/prefixhandler_table_internal_test.go

@@ -101,6 +101,58 @@ 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.
+func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
+	t.Parallel()
+
+	h := prefixHandlerWithoutTimer(newMockStore())
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
+
+	h.peerDownMu.Lock()
+
+	handled := returnsInTime(func() {
+		for _, msg := range []*ristypes.RISMessage{
+			sessionDown(),
+			peerAnnouncement(secondPeerIP, otherV4Prefix),
+			peerAnnouncement(testPeerIP, testV6Prefix),
+			sessionDown(),
+			peerAnnouncement(testPeerIP, setV4Prefix),
+		} {
+			h.HandleMessage(msg)
+		}
+	})
+	if !handled {
+		t.Fatalf("the handler did not take the messages after the session "+
+			"went down within %s", stepTimeout)
+	}
+
+	checkTableOrigins(t, h.table, map[string]map[int][]string{
+		testV4Prefix:  {testASNB: {testPeerIP}},
+		otherV4Prefix: {testASNB: {secondPeerIP}},
+		testV6Prefix:  {},
+		setV4Prefix:   {},
+	})
+
+	h.peerDownMu.Unlock()
+
+	if !returnsInTime(h.peerDowns.Wait) {
+		t.Fatalf("the PeerDowns did not return within %s", stepTimeout)
+	}
+
+	checkTableOrigins(t, h.table, map[string]map[int][]string{
+		testV4Prefix:  {},
+		otherV4Prefix: {testASNB: {secondPeerIP}},
+		testV6Prefix:  {},
+		setV4Prefix:   {testASNB: {testPeerIP}},
+	})
+}
+
 // TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable hands the
 // prefix handler announcements whose AS path gives no origin. One whose path
 // is empty or holds AS 0 removes the peer's route, as a withdrawal does; one
@@ -200,7 +252,8 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
 }
 
 // checkTableCases hands the messages of each case to a new prefix handler and
-// checks its routing table right after, with no flush.
+// checks its routing table once the PeerDowns it started have returned, with
+// no flush.
 func checkTableCases(t *testing.T, tests []tableCase) {
 	t.Helper()
 
@@ -213,17 +266,28 @@ func checkTableCases(t *testing.T, tests []tableCase) {
 				h.HandleMessage(msg)
 			}
 
-			for prefix, want := range tt.want {
-				got := tableOrigins(h.table, prefix)
-				if !reflect.DeepEqual(got, want) {
-					t.Errorf("peers of %s by origin AS = %v, want %v",
-						prefix, got, want)
-				}
-			}
+			h.peerDowns.Wait()
+			checkTableOrigins(t, h.table, tt.want)
 		})
 	}
 }
 
+// checkTableOrigins checks, for each prefix in want, the peers that carry it
+// in table with each origin AS.
+func checkTableOrigins(
+	t *testing.T, table *routetable.Table, want map[string]map[int][]string,
+) {
+	t.Helper()
+
+	for prefix, wantOrigins := range want {
+		got := tableOrigins(table, prefix)
+		if !reflect.DeepEqual(got, wantOrigins) {
+			t.Errorf("peers of %s by origin AS = %v, want %v",
+				prefix, got, wantOrigins)
+		}
+	}
+}
+
 // tableWritesStore is a real database that hands each write of the routing
 // table's changes to written once it is made.
 type tableWritesStore struct {
@@ -498,6 +562,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) {
 		h.HandleMessage(msg)
 	}
 
+	h.peerDowns.Wait()
 	h.flushBatch()
 
 	var logs bytes.Buffer
@@ -536,6 +601,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) {
 		restarted.HandleMessage(msg)
 	}
 
+	h.peerDowns.Wait()
 	restarted.Stop(t.Context())
 
 	read, err := db.ReadRouteTable(t.Context())