Take every down peer out of the table in one walk (closes #225)

The prefix handler ran one PeerDown per peer, one after another, and
stopped taking messages at 20,000 held-back messages. Sessions going
down together queued up their walks, so a returning full-table peer
could reach that limit.

The table's PeersDown now takes several peers out of every prefix in
one walk. One walk runs at a time. The next one takes every peer whose
session went down meanwhile, so a message is held back for two walks at
most. Each walk takes about 2.6 seconds for each full-table peer it
takes out. The limit is now 150,000 messages, about 150 MiB: enough
when four full-table peers go down together. The status page shows how
many messages each handler holds back.

Model: opus-5-5

README.md

@@ -652,9 +652,10 @@ What happens at each limit:
 
 - Held-back messages: the prefix handler holds back a peer's messages after the
   peer's session goes down. It applies them once the peer has left every prefix.
-  It holds at most 20,000 such messages (`maxHeldBackMessages`), about 20 MiB.
-  At that number it takes nothing more off its queue until it has applied one of
-  them.
+  It holds at most 150,000 such messages (`maxHeldBackMessages`), about 150 MiB.
+  The limit is enough when the sessions of four full-table peers go down
+  together. At the limit it takes nothing more off its queue until it has
+  applied one of them.
 
 With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
 with the goroutine count and Go memory figures.

TODO.md

@@ -35,6 +35,16 @@ tested.
 
 # Completed Steps
 
+- 2026-11-02: when several peers' sessions go down at once, the prefix handler
+  goes on taking messages. The table's `PeersDown` takes several peers out of
+  every prefix in one walk over the table. One walk runs at a time. The next one
+  takes every peer whose session went down meanwhile. So a peer's messages are
+  held back for two walks at most. Each walk takes about 2.6 seconds for each
+  full-table peer it takes out. The handler holds back at most 150,000 messages,
+  about 150 MiB. The limit is enough when the sessions of four full-table peers
+  go down together. The status page shows how many messages the handler holds
+  back. Each peer used to have a walk of its own, and the handler stopped taking
+  messages at 20,000 held back (closes #225)
 - 2026-11-02: like a failed WHOIS lookup, a lookup skipped because its RIR is
   paused makes the fetcher wait longer before the next lookup. The longest wait
   is one minute. The status page does not count a skip as a failure. A skip used

internal/database/database_internal_test.go

@@ -273,7 +273,7 @@ func TestVacuumReturnsFreePages(t *testing.T) {
 	writeTableChanges(t, db, table)
 
 	// The peer going down removes every prefix, and so every row.
-	table.PeerDown(peerA)
+	table.PeersDown([]string{peerA})
 	writeTableChanges(t, db, table)
 
 	var before int

internal/database/routes_internal_test.go

@@ -107,7 +107,7 @@ func TestReadRouteTableGivesBackWhatWasWritten(t *testing.T) {
 
 	// A peer moves to another origin, one goes down and a new one comes.
 	announceTo(source, tablePrefixV4, asB, peerA, 60)
-	source.PeerDown(peerC)
+	source.PeersDown([]string{peerC})
 	announceTo(source, tablePrefixV6, asB, peerE, 70)
 
 	writeTableChanges(t, db, source)
@@ -133,7 +133,7 @@ func TestReadRouteTableGivesATableThatTakesLaterUpdates(t *testing.T) {
 	// another one goes down, and a new peer comes.
 	for _, table := range []*routetable.Table{source, read} {
 		table.Withdraw(netip.MustParsePrefix(tablePrefixV4), peerA)
-		table.PeerDown(peerB)
+		table.PeersDown([]string{peerB})
 		announceTo(table, tablePrefixShortV4, asB, peerD, 50)
 	}
 

internal/routetable/reads_test.go

@@ -227,7 +227,7 @@ func TestCounts(t *testing.T) {
 	})
 
 	// peerC was the last peer of the IPv6 prefix. It keeps its number.
-	table.PeerDown(peerC)
+	table.PeersDown([]string{peerC})
 	checkCounts(t, table, routetable.Counts{
 		Pairs:                2,
 		Peers:                3,

internal/routetable/routetable.go

@@ -28,7 +28,7 @@ const (
 	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.
+	// one hold of the lock. Overview and PeersDown each walk the whole table.
 	// Callers waiting for the lock get it between batches.
 	walkBatch = 10000
 
@@ -205,20 +205,29 @@ 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
-// 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) {
+// PeersDown applies the sessions of peers going down: each of the peers leaves
+// every peer set, and every route of each is dropped. When any of them carries
+// a prefix, PeersDown visits every prefix in the table once, 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
+// one of the peers applied during the walk may be dropped too. So the caller
+// applies those peers' later updates only once PeersDown has returned.
+func (t *Table) PeersDown(peers []string) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
-	number, known := t.peerNumbers[peer]
-	if !known || t.peerPrefixes[number] == 0 {
+	// numbers holds the number of each of the peers that carries a prefix.
+	numbers := make([]uint, 0, len(peers))
+
+	for _, peer := range peers {
+		number, known := t.peerNumbers[peer]
+		if known && t.peerPrefixes[number] > 0 {
+			numbers = append(numbers, number)
+		}
+	}
+
+	if len(numbers) == 0 {
 		return
 	}
 
@@ -228,9 +237,15 @@ func (t *Table) PeerDown(peer string) {
 	for _, f := range []*family{&t.ipv4, &t.ipv6} {
 		for _, prefixes := range f.byLength {
 			for prefix, origins := range prefixes {
-				o := originWithPeer(origins, number)
-				if o != nil {
-					t.leave(prefix, o, number)
+				for _, number := range numbers {
+					o := originWithPeer(origins, number)
+					if o != nil {
+						t.leave(prefix, o, number)
+
+						// leave may have removed an origin of the
+						// prefix, or the prefix itself.
+						origins = prefixes[prefix]
+					}
 				}
 
 				// Go allows the map to change before the next step of

internal/routetable/routetable_internal_test.go

@@ -7,11 +7,11 @@ import (
 	"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) {
+// TestPeersDownLetsReadsIn reads the table's counts again and again while a
+// peer that carries more than two batches of prefixes goes down. PeersDown 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 TestPeersDownLetsReadsIn(t *testing.T) {
 	t.Parallel()
 
 	const (
@@ -32,8 +32,8 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
 		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
+	// Hold a read lock until PeersDown waits for the lock, so that the first
+	// read below starts once PeersDown holds it. TryRLock fails once a writer
 	// waits.
 	table.mu.RLock()
 
@@ -42,7 +42,7 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
 	go func() {
 		defer close(peerDownDone)
 
-		table.PeerDown(peer)
+		table.PeersDown([]string{peer})
 	}()
 
 	for table.mu.TryRLock() {
@@ -60,7 +60,7 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
 
 		select {
 		case <-peerDownDone:
-			t.Fatal("no read got in while PeerDown ran")
+			t.Fatal("no read got in while PeersDown ran")
 		default:
 		}
 	}

internal/routetable/routetable_test.go

@@ -69,10 +69,10 @@ func withdraw(peer string) step {
 	}
 }
 
-// peerDown is the step of peer's session going down.
-func peerDown(peer string) step {
+// peersDown is the step of the sessions of peers going down.
+func peersDown(peers ...string) step {
 	return func(table *routetable.Table, _ netip.Prefix) {
-		table.PeerDown(peer)
+		table.PeersDown(peers)
 	}
 }
 
@@ -272,21 +272,28 @@ func TestWithdrawalsAndPeerDowns(t *testing.T) {
 		{
 			name: "a peer going down leaves the prefix",
 			steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
-				peerDown(peerB)},
+				peersDown(peerB)},
 			want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
 				Routes: []routetable.Route{routeOf(peerA, asA, 10)}}},
 		},
 		{
 			name: "a peer going down after it withdrew changes nothing",
 			steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
-				withdraw(peerB), peerDown(peerB)},
+				withdraw(peerB), peersDown(peerB)},
 			want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
 				Routes: []routetable.Route{routeOf(peerA, asA, 10)}}},
 		},
+		{
+			name: "peers going down together each leave the prefix",
+			steps: []step{announce(peerA, asA, 10), announce(peerB, asB, 20),
+				announce(peerC, asB, 30), peersDown(peerA, peerB)},
+			want: []routetable.Origin{{ASN: asB, Peers: []string{peerC},
+				Routes: []routetable.Route{routeOf(peerC, asB, 30)}}},
+		},
 	})
 }
 
-func TestPeerDownLeavesEveryPrefix(t *testing.T) {
+func TestPeersDownLeavesEveryPrefix(t *testing.T) {
 	t.Parallel()
 
 	table := routetable.New(routetable.Params{})
@@ -303,7 +310,7 @@ func TestPeerDownLeavesEveryPrefix(t *testing.T) {
 	}
 
 	table.Announce(prefixes[0], asB, routeOf(peerB, asB, 20))
-	table.PeerDown(peerA)
+	table.PeersDown([]string{peerA})
 
 	want := []routetable.Origin{{ASN: asB, Peers: []string{peerB},
 		Routes: []routetable.Route{routeOf(peerB, asB, 20)}}}
@@ -333,6 +340,32 @@ func TestPeerDownLeavesEveryPrefix(t *testing.T) {
 	}
 }
 
+// TestPeersDownTakesOutPeersOfDifferentPrefixes takes peerA and peerB out in
+// one PeersDown. peerA carries only prefixV4, and peerB only prefixV6. So
+// peerA's leaving removes prefixV4 before PeersDown looks for peerB there. Both
+// prefixes must then be gone.
+func TestPeersDownTakesOutPeersOfDifferentPrefixes(t *testing.T) {
+	t.Parallel()
+
+	table := routetable.New(routetable.Params{})
+	prefixes := []netip.Prefix{
+		netip.MustParsePrefix(prefixV4),
+		netip.MustParsePrefix(prefixV6),
+	}
+
+	table.Announce(prefixes[0], asA, routeOf(peerA, asA, 10))
+	table.Announce(prefixes[1], asB, routeOf(peerB, asB, 20))
+	table.PeersDown([]string{peerA, peerB})
+
+	for _, prefix := range prefixes {
+		got := table.Origins(prefix)
+		if got != nil {
+			t.Errorf("Origins(%s) = %+v after its only peer went down, want nil",
+				prefix, got)
+		}
+	}
+}
+
 func TestCountsOriginsWithoutKeptRoute(t *testing.T) {
 	t.Parallel()
 
@@ -358,7 +391,7 @@ func TestCountsOriginsWithoutKeptRoute(t *testing.T) {
 		{announce(peerC, asA, 30), 0},
 		{withdraw(peerB), 0},
 		{withdraw(peerC), 1},
-		{peerDown(peerA), 0},
+		{peersDown(peerA), 0},
 	}
 
 	for i, s := range steps {
@@ -464,7 +497,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
 	checkChanges(t, table, []change{}, map[string]uint{})
 
 	// peerA going down changes v4, the one prefix it carries.
-	table.PeerDown(peerA)
+	table.PeersDown([]string{peerA})
 	checkChanges(t, table, []change{
 		{Prefix: prefixV4, OriginASN: asA, Peers: []uint{1, 2},
 			Routes: []routetable.Route{
@@ -473,7 +506,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
 	}, map[string]uint{})
 
 	// peerA, which carries nothing now, goes down again.
-	table.PeerDown(peerA)
+	table.PeersDown([]string{peerA})
 	checkChanges(t, table, []change{}, map[string]uint{})
 }
 
@@ -617,7 +650,7 @@ func TestConcurrentUse(t *testing.T) {
 		func(round int) { table.Announce(prefix, asA, routeOf(peerA, asA, round)) },
 		func(round int) { table.Announce(prefix, asB, routeOf(peerB, asB, round)) },
 		func(int) { table.Withdraw(prefix, peerA) },
-		func(int) { table.PeerDown(peerB) },
+		func(int) { table.PeersDown([]string{peerB}) },
 		func(int) { table.PutBackChanges(table.TakeChanges(math.MaxInt)) },
 		func(int) { table.ChangeCount() },
 		func(int) { table.Lookup(prefix.Addr()) },

internal/routewatch/prefixhandler.go

@@ -46,10 +46,19 @@ const (
 	maxTableChangesPerWrite = 20000
 
 	// maxHeldBackMessages is the most messages the handler holds back from
-	// peers whose PeerDown has not returned. A message takes about 1 KiB, so
-	// that many take about 20 MiB (README "Memory"). At this number
-	// HandleMessage waits until one of them is applied.
-	maxHeldBackMessages = 20000
+	// peers whose session went down. A peer's messages are held back while
+	// up to two PeersDown calls run: the one running when its session goes
+	// down, and the next, which takes the peer out. A call takes about 2.6
+	// seconds for each full-table peer it takes out. Take a full-table peer
+	// whose session goes down together with those of three other full-table
+	// peers. The two calls then take about 10.4 seconds, 2.6 for each of the
+	// four. The feed peaks at about 14,000 messages a second, so the four
+	// peers send at most about 146,000 messages in that time. Once the second
+	// call has returned, the handler applies the held-back messages faster
+	// than the feed sends more. A message takes about 1 KiB, so 150,000 take
+	// about 150 MiB (README "Memory"). At this number HandleMessage waits
+	// until one of them is applied.
+	maxHeldBackMessages = 150000
 )
 
 // PrefixHandler keeps the routing table in memory current from the feed. It
@@ -58,14 +67,15 @@ 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 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. 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.
+// The table's PeersDown applies sessions going down. It visits every prefix,
+// which takes seconds on a full table. So the handler runs PeersDown on a
+// separate goroutine and goes on taking messages meanwhile. One PeersDown runs
+// at a time. The next one takes every peer whose session went down meanwhile.
+// PeersDown lets the table's lock go about half of the time, so the handler
+// applies other peers' messages while PeersDown runs. The handler holds back a
+// down peer's later messages until the peer has left every prefix. 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.
@@ -103,20 +113,25 @@ type PrefixHandler struct {
 	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.
-	// heldCount, under mu, is how many messages heldBack holds.
+	// Peers whose session went down. heldBack, under mu, holds each such
+	// peer's later messages, in order, until the peer has left every prefix
+	// and they are applied. A peer with no such messages yet has an empty
+	// entry. heldCount, under mu, is how many messages heldBack holds.
 	// heldTaken is signalled each time a held-back message is taken out to
-	// be applied; its L is &mu. peerDowns counts the peerDown goroutines
-	// that have not returned.
-	peerDownMu sync.Mutex
-	heldBack   map[string][]*ristypes.RISMessage
-	heldCount  int
-	heldTaken  sync.Cond
-	peerDowns  sync.WaitGroup
+	// be applied; its L is &mu.
+	heldBack  map[string][]*ristypes.RISMessage
+	heldCount int
+	heldTaken sync.Cond
+	// downPeers, under mu, lists the peers whose session went down and that
+	// the next PeersDown takes. peerDownRunning, under mu, says whether
+	// peerDown is running. peerDowns counts the peerDown goroutines that have
+	// not returned.
+	downPeers       []string
+	peerDownRunning bool
+	peerDowns       sync.WaitGroup
+	// tablePeersDown is the table's PeersDown. Only peerDown calls it, so a
+	// test may replace it before it hands the handler a message.
+	tablePeersDown func(peers []string)
 }
 
 // seenPrefix is what the batch keeps of an update: the prefix it names and its
@@ -181,6 +196,7 @@ func newPrefixHandler(
 		stopCh:               make(chan context.Context, 1),
 		done:                 make(chan struct{}),
 		heldBack:             make(map[string][]*ristypes.RISMessage),
+		tablePeersDown:       table.PeersDown,
 	}
 	h.batchTaken.L = &h.mu
 	h.heldTaken.L = &h.mu
@@ -212,7 +228,8 @@ func (h *PrefixHandler) QueueCapacity() int {
 }
 
 // HandleMessage processes a message to track prefix information. A message
-// from a peer whose PeerDown has not returned is held back until it has.
+// from a peer whose session went down is held back until the peer has left
+// every prefix.
 func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -235,7 +252,16 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.applyMessage(msg)
 }
 
-// Stop waits for every PeerDown the handler started and for the messages held
+// HeldBackCount returns how many messages the handler holds back from peers
+// whose session went down. The status page shows it.
+func (h *PrefixHandler) HeldBackCount() int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	return h.heldCount
+}
+
+// Stop waits for every PeersDown the handler started and for the messages held
 // back meanwhile to be applied. It gives up this wait once ctx ends. Then Stop
 // stops the flush goroutine and waits for it until ctx ends. The flush
 // goroutine makes one last flush before it returns. That flush writes at most
@@ -290,8 +316,8 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
 }
 
 // applyMessage applies msg. The caller holds mu. 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.
+// has the peer leave every prefix. 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 {
@@ -370,34 +396,64 @@ func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// startPeerDown starts peerDown for peer on a new goroutine. The caller holds
-// mu. The peer's messages are held back until peerDown has applied them.
-// Messages held back for an earlier PeerDown of the peer, and not yet applied,
-// stay held back.
+// startPeerDown adds peer to downPeers. It starts peerDown on a new goroutine
+// unless peerDown is running. The caller holds mu. The peer's messages are
+// held back until peerDown has applied them. Messages held back for an earlier
+// session of the peer going down, and not yet applied, stay held back.
 func (h *PrefixHandler) startPeerDown(peer string) {
 	_, goingDown := h.heldBack[peer]
 	if !goingDown {
 		h.heldBack[peer] = []*ristypes.RISMessage{}
 	}
 
+	h.downPeers = append(h.downPeers, peer)
+
+	if h.peerDownRunning {
+		return
+	}
+
+	h.peerDownRunning = true
 	h.peerDowns.Add(1)
 
-	go h.peerDown(peer)
+	go h.peerDown()
 }
 
-// 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
-// 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) {
+// peerDown takes every peer in downPeers and runs the table's PeersDown for
+// them. Then it applies the messages held back from each of those peers. It
+// repeats this until downPeers is empty.
+func (h *PrefixHandler) peerDown() {
 	defer h.peerDowns.Done()
 
-	h.peerDownMu.Lock()
-	h.table.PeerDown(peer)
-	h.peerDownMu.Unlock()
+	for {
+		h.mu.Lock()
 
+		peers := h.downPeers
+		h.downPeers = nil
+
+		if len(peers) == 0 {
+			h.peerDownRunning = false
+			h.mu.Unlock()
+
+			return
+		}
+
+		h.mu.Unlock()
+
+		h.tablePeersDown(peers)
+
+		for _, peer := range peers {
+			h.applyHeldBack(peer)
+		}
+	}
+}
+
+// applyHeldBack applies the messages held back from 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 adds the
+// peer to downPeers. The rest stay held back until the next PeersDown has
+// returned.
+func (h *PrefixHandler) applyHeldBack(peer string) {
 	for {
 		h.mu.Lock()
 

internal/routewatch/prefixhandler_internal_test.go

@@ -238,7 +238,7 @@ 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
+// deadline that has passed, while a peer's PeersDown 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()
@@ -249,8 +249,8 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
 		routetable.New(routetable.Params{}), longerThanAnyTest,
 		maxTableChangesPerWrite)
 
-	h.peerDownMu.Lock()
-	defer h.peerDownMu.Unlock()
+	release := holdUpPeersDown(h)
+	defer close(release)
 
 	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
 	h.HandleMessage(sessionDown())
@@ -268,8 +268,8 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
 }
 
 // TestPrefixHandlerStopWaitsForTheMessagesHeldBack stops a prefix handler
-// while a peer's PeerDown is held up and an announcement of the peer is held
-// back. Stop must wait. Once the PeerDown is let go and Stop returns, the
+// while a peer's PeersDown is held up and an announcement of the peer is held
+// back. Stop must wait. Once the PeersDown is let go and Stop returns, the
 // announcement must be in the database.
 func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
 	t.Parallel()
@@ -278,7 +278,7 @@ func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
 	h := newPrefixHandler(db, logger.New(), routetable.New(routetable.Params{}),
 		longerThanAnyTest, maxTableChangesPerWrite)
 
-	h.peerDownMu.Lock()
+	release := holdUpPeersDown(h)
 	h.HandleMessage(sessionDown())
 	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
 
@@ -291,16 +291,16 @@ func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
 
 	select {
 	case <-stopped:
-		t.Error("Stop returned while a PeerDown was held up")
+		t.Error("Stop returned while a PeersDown was held up")
 	case <-time.After(stillWaitingAfter):
 	}
 
-	h.peerDownMu.Unlock()
+	close(release)
 
 	select {
 	case <-stopped:
 	case <-time.After(stepTimeout):
-		t.Fatalf("Stop did not return within %s of the PeerDown's release",
+		t.Fatalf("Stop did not return within %s of the PeersDown's release",
 			stepTimeout)
 	}
 
@@ -324,16 +324,32 @@ const longerThanAnyTest = time.Hour
 // applies the updates to an empty routing table. It is built without
 // NewPrefixHandler's flush goroutine, so the test flushes itself.
 func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
+	table := routetable.New(routetable.Params{})
+
 	return &PrefixHandler{
 		db:                   store,
 		logger:               logger.New(),
-		table:                routetable.New(routetable.Params{}),
+		table:                table,
 		batchMaxSize:         prefixBatchMaxSize,
 		tableChangesPerWrite: maxTableChangesPerWrite,
 		heldBack:             make(map[string][]*ristypes.RISMessage),
+		tablePeersDown:       table.PeersDown,
 	}
 }
 
+// holdUpPeersDown makes each PeersDown of h wait until the returned channel is
+// closed, as another peer's long PeersDown would hold it up. The test calls it
+// before it hands h a message.
+func holdUpPeersDown(h *PrefixHandler) chan struct{} {
+	release := make(chan struct{})
+	h.tablePeersDown = func(peers []string) {
+		<-release
+		h.table.PeersDown(peers)
+	}
+
+	return release
+}
+
 // newTestDatabase opens a database in a new temporary directory. It is closed
 // when the test ends.
 func newTestDatabase(t *testing.T) *database.Database {

internal/routewatch/prefixhandler_table_internal_test.go

@@ -103,23 +103,24 @@ 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
+// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeersDown of a
+// peer whose session went down, as another peer's long PeersDown 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:
 //   - 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.
+// HeldBackCount must count them. Once the PeersDown calls 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))
+	const heldMessages = 3
 
-	h.peerDownMu.Lock()
+	h := prefixHandlerWithoutTimer(newMockStore())
+	release := holdUpPeersDown(h)
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
 
 	handled := returnsInTime(func() {
 		for _, msg := range []*ristypes.RISMessage{
@@ -144,10 +145,15 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
 		setV4Prefix:   {},
 	})
 
-	h.peerDownMu.Unlock()
+	if h.HeldBackCount() != heldMessages {
+		t.Errorf("HeldBackCount() = %d, want %d", h.HeldBackCount(),
+			heldMessages)
+	}
+
+	close(release)
 
 	if !returnsInTime(h.peerDowns.Wait) {
-		t.Fatalf("the PeerDowns did not return within %s", stepTimeout)
+		t.Fatalf("the PeersDown calls did not return within %s", stepTimeout)
 	}
 
 	checkTableOrigins(t, h.table, map[string]map[int][]string{
@@ -158,21 +164,84 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
 	})
 }
 
-// TestPrefixHandlerAppliesMessagesWhilePeerDownRuns gives a peer 50,000
+// TestPrefixHandlerTakesPeersDownTogether holds up the PeersDown of a peer
+// whose session went down. Meanwhile the sessions of three more peers go down,
+// and each of them announces a prefix. Once the first PeersDown is let go, the
+// next one must take the three peers together. Each one's announcement must
+// then be in the table.
+func TestPrefixHandlerTakesPeersDownTogether(t *testing.T) {
+	t.Parallel()
+
+	others := []string{secondPeerIP, thirdPeerIP, fourthPeerIP}
+	h := prefixHandlerWithoutTimer(newMockStore())
+
+	// calls receives the peers of each PeersDown, with room for one call
+	// for each peer.
+	calls := make(chan []string, len(others)+1)
+	release := make(chan struct{})
+	h.tablePeersDown = func(peers []string) {
+		calls <- slices.Clone(peers)
+
+		<-release
+		h.table.PeersDown(peers)
+	}
+
+	for _, peer := range append([]string{testPeerIP}, others...) {
+		h.HandleMessage(peerAnnouncement(peer, testV4Prefix))
+	}
+
+	h.HandleMessage(sessionDown())
+
+	select {
+	case <-calls:
+	case <-time.After(stepTimeout):
+		t.Fatalf("PeersDown did not begin within %s", stepTimeout)
+	}
+
+	for _, peer := range others {
+		h.HandleMessage(sessionDownOf(peer))
+		h.HandleMessage(peerAnnouncement(peer, otherV4Prefix))
+	}
+
+	close(release)
+
+	if !returnsInTime(h.peerDowns.Wait) {
+		t.Fatalf("the PeersDown calls did not return within %s", stepTimeout)
+	}
+
+	close(calls)
+
+	later := make([][]string, 0, len(calls))
+	for peers := range calls {
+		later = append(later, peers)
+	}
+
+	if !reflect.DeepEqual(later, [][]string{others}) {
+		t.Errorf("the PeersDown calls after the first took %v, want %v "+
+			"together", later, others)
+	}
+
+	checkTableOrigins(t, h.table, map[string]map[int][]string{
+		testV4Prefix:  {},
+		otherV4Prefix: {testASNB: others},
+	})
+}
+
+// TestPrefixHandlerAppliesMessagesWhilePeersDownRuns 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
+// table's PeersDown then visits the prefixes in five batches. Once PeersDown
+// has begun, the handler is handed 1,000 announcements of another peer. The
+// handler must apply them all before PeersDown ends, while the down peer still
 // carries some of its prefixes.
-func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
+func TestPrefixHandlerAppliesMessagesWhilePeersDownRuns(t *testing.T) {
 	t.Parallel()
 
 	const (
-		// downPeerPrefixes is five times the 10,000 prefixes PeerDown visits
+		// downPeerPrefixes is five times the 10,000 prefixes PeersDown 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
+		// while PeersDown runs, if PeersDown 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.
@@ -184,7 +253,7 @@ func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
 
 	h.HandleMessage(sessionDown())
 
-	// Wait until PeerDown has begun: the down peer has left some of its
+	// Wait until PeersDown 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 {
@@ -207,15 +276,76 @@ func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
 	}
 
 	if counts.IPv4Routes == 0 {
-		t.Errorf("PeerDown ended before the handler applied the other peer's "+
+		t.Errorf("PeersDown 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)
+		t.Fatalf("the PeersDown did not return within %s", stepTimeout)
 	}
 }
 
+// TestPrefixHandlerTakesMessagesWhileAReturningPeerSendsItsTable gives a peer
+// 50,000 prefixes and hands the prefix handler that peer's session going down.
+// Once PeersDown has begun, the peer comes back and sends more announcements
+// than the handler's queue holds, as a peer that sends its whole table again
+// does. The handler must hold them back and go on taking messages. Another
+// peer's announcement, handed next, must be applied before PeersDown ends.
+func TestPrefixHandlerTakesMessagesWhileAReturningPeerSendsItsTable(t *testing.T) {
+	t.Parallel()
+
+	const (
+		// downPeerPrefixes is five times the 10,000 prefixes PeersDown visits
+		// in one hold of the table's lock.
+		downPeerPrefixes  = 50000
+		returningMessages = prefixHandlerQueueSize + 1
+	)
+
+	table := routetable.New(routetable.Params{})
+	announceManyPrefixes(table, downPeerPrefixes)
+
+	h := newPrefixHandler(newMockStore(), logger.New(), table,
+		longerThanAnyTest, maxTableChangesPerWrite)
+	defer stopInTime(t, h)
+
+	h.HandleMessage(sessionDown())
+
+	// Wait until PeersDown has begun: the down peer has left some of its
+	// prefixes, which are all IPv4.
+	for h.table.Counts().IPv4Routes == downPeerPrefixes {
+		runtime.Gosched()
+	}
+
+	returning := peerAnnouncement(testPeerIP, testV6Prefix)
+
+	if !returnsInTime(func() {
+		for range returningMessages {
+			h.HandleMessage(returning)
+		}
+
+		h.HandleMessage(peerAnnouncement(secondPeerIP, testV6Prefix))
+	}) {
+		t.Fatalf("the handler did not take the messages within %s", stepTimeout)
+	}
+
+	if h.table.Counts().IPv4Routes == 0 {
+		t.Error("PeersDown ended before the handler took the other peer's " +
+			"announcement")
+	}
+
+	checkTableOrigins(t, h.table, map[string]map[int][]string{
+		testV6Prefix: {testASNB: {secondPeerIP}},
+	})
+
+	if !returnsInTime(h.peerDowns.Wait) {
+		t.Fatalf("the PeersDown did not return within %s", stepTimeout)
+	}
+
+	checkTableOrigins(t, h.table, map[string]map[int][]string{
+		testV6Prefix: {testASNB: {testPeerIP, secondPeerIP}},
+	})
+}
+
 // 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
@@ -237,14 +367,14 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
 
 	defer stopInTime(t, h)
 
-	h.peerDownMu.Lock()
+	release := holdUpPeersDown(h)
 	h.HandleMessage(sessionDown())
 
 	for range heldMessages {
 		h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix))
 	}
 
-	h.peerDownMu.Unlock()
+	close(release)
 
 	select {
 	case <-logs.firstWrite:
@@ -268,10 +398,10 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
 	}
 }
 
-// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeerDown of a peer
+// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeersDown of a peer
 // whose session went down. It hands the handler maxHeldBackMessages messages
 // of that peer. HandleMessage must then wait, even for another peer's
-// message, until the PeerDown returns and a held-back message is applied.
+// message, until the PeersDown returns and a held-back message is applied.
 func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
 	t.Parallel()
 
@@ -280,7 +410,7 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
 		maxTableChangesPerWrite)
 	defer stopInTime(t, h)
 
-	h.peerDownMu.Lock()
+	release := holdUpPeersDown(h)
 	h.HandleMessage(sessionDown())
 
 	held := peerAnnouncement(testPeerIP, testV6Prefix)
@@ -302,12 +432,12 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
 	case <-time.After(stillWaitingAfter):
 	}
 
-	h.peerDownMu.Unlock()
+	close(release)
 
 	select {
 	case <-returned:
 	case <-time.After(stepTimeout):
-		t.Fatal("HandleMessage did not return after the PeerDown returned")
+		t.Fatal("HandleMessage did not return after the PeersDown returned")
 	}
 }
 
@@ -424,7 +554,7 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
 }
 
 // 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
+// checks the handler's routing table once the handler's PeersDown calls have
 // returned, with no flush.
 func checkTableCases(t *testing.T, tests []tableCase) {
 	t.Helper()

internal/server/handlers.go

@@ -486,6 +486,7 @@ type HandlerStatsInfo struct {
 	QueueHighWaterMark int     `json:"queue_high_water_mark"`
 	ProcessedCount     uint64  `json:"processed_count"`
 	DroppedCount       uint64  `json:"dropped_count"`
+	HeldBackCount      int     `json:"held_back_count"`
 	AvgProcessTimeMs   float64 `json:"avg_process_time_ms"`
 	MinProcessTimeMs   float64 `json:"min_process_time_ms"`
 	MaxProcessTimeMs   float64 `json:"max_process_time_ms"`
@@ -570,6 +571,7 @@ func handlerStatsInfo(handlerStats []streamer.HandlerStats) []HandlerStatsInfo {
 			QueueHighWaterMark: hs.QueueHighWaterMark,
 			ProcessedCount:     hs.ProcessedCount,
 			DroppedCount:       hs.DroppedCount,
+			HeldBackCount:      hs.HeldBackCount,
 			AvgProcessTimeMs: float64(hs.AvgProcessTime.Microseconds()) /
 				microsecondsPerMillisecond,
 			MinProcessTimeMs: float64(hs.MinProcessTime.Microseconds()) /

internal/server/handlers_internal_test.go

@@ -295,6 +295,43 @@ func TestStatsAnswerCarriesChurnPerMinute(t *testing.T) {
 	}
 }
 
+// heldBackMessages is how many messages heldBackHandler holds back.
+const heldBackMessages = 7
+
+// heldBackHandler is a stream handler that wants no message and holds back
+// heldBackMessages messages.
+type heldBackHandler struct{}
+
+func (heldBackHandler) WantsMessage(string) bool           { return false }
+func (heldBackHandler) HandleMessage(*ristypes.RISMessage) {}
+func (heldBackHandler) QueueCapacity() int                 { return 1 }
+func (heldBackHandler) HeldBackCount() int                 { return heldBackMessages }
+
+// TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack registers a handler
+// that holds back messages. Its figures in /api/v1/stats, which the status
+// page shows, must give how many.
+func TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack(t *testing.T) {
+	t.Parallel()
+
+	s := newServerWithTestRoutes(t)
+	s.streamer.RegisterHandler(heldBackHandler{})
+
+	var body struct {
+		Data StatsResponse `json:"data"`
+	}
+
+	err := json.Unmarshal(get(t, s, statsPath).Body.Bytes(), &body)
+	if err != nil {
+		t.Fatalf("decoding the answer: %v", err)
+	}
+
+	stats := body.Data.HandlerStats
+	if len(stats) != 1 || stats[0].HeldBackCount != heldBackMessages {
+		t.Errorf("handler figures %+v, want one handler holding back %d "+
+			"messages", stats, heldBackMessages)
+	}
+}
+
 // fixedIntervalASNFetcher is a WHOIS fetcher that does nothing and reports
 // interval as its wait between fetches.
 type fixedIntervalASNFetcher struct {

internal/streamer/streamer.go

@@ -351,6 +351,10 @@ type HandlerStats struct {
 	AvgProcessTime     time.Duration
 	MinProcessTime     time.Duration
 	MaxProcessTime     time.Duration
+	// HeldBackCount is how many messages the handler holds back to apply
+	// later: what its HeldBackCount method returns, or 0 for a handler
+	// without one.
+	HeldBackCount int
 }
 
 // GetHandlerStats returns a snapshot of performance statistics for all registered
@@ -392,6 +396,11 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
 
 		info.metrics.mu.Unlock()
 
+		holder, holdsBack := info.handler.(interface{ HeldBackCount() int })
+		if holdsBack {
+			hs.HeldBackCount = holder.HeldBackCount()
+		}
+
 		stats = append(stats, hs)
 	}
 

internal/templates/status.html

@@ -541,6 +541,10 @@
                         <span class="metric-label">Dropped</span>
                         <span class="metric-value ${handler.dropped_count > 0 ? 'disconnected' : ''}">${formatNumber(handler.dropped_count)}</span>
                     </div>
+                    <div class="metric">
+                        <span class="metric-label">Held Back</span>
+                        <span class="metric-value">${formatNumber(handler.held_back_count)}</span>
+                    </div>
                     <div class="metric">
                         <span class="metric-label">Avg Time</span>
                         <span class="metric-value">${formatProcessingTime(handler.avg_process_time_ms)}</span>