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

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 and PeerDown each walk the whole table.
+	// Callers waiting for the lock get it between batches.
+	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 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.
 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
+}