Rewrite TODO.md sentence by sentence (part of #157)

Every sentence of TODO.md is rewritten from scratch against the rules
of #157: about 30 words at
most, one idea each, subject and verb early, identifiers in backticks.
Status and Next Step describe the repo as the tracker shows it now.
Every Completed Steps entry keeps its date, its facts and its issue
reference, checked against the commit it names. The Workflow section
keeps its meaning. The entry for this rewrite is added.

Model: fable-5-1

TODO.md

@@ -22,7 +22,7 @@ formatting of the Go code and the markdown. Then it builds the image.
 The image sets memory ceilings for a 5 GiB container (README "Memory"). README
 "Running under upaas" says how to run it.
 
-A five-day run of `1c65656` on the live feed, from 2026-10-27 to 2026-11-01,
+A five-day run of `1c65656` on the live feed, from 27 October to 1 November,
 found three problems:
 
 - The prefix handler dropped updates while a peer's session went down
@@ -34,12 +34,12 @@ found three problems:
 
 All three are fixed on `next`. So is
 https://git.eeqj.de/sneak/routewatch/issues/222, which the third fix brought up.
-A run of `26d939a` on the live feed showed the prefix handler still dropping
-updates while the home page's cards were worked out.
-https://git.eeqj.de/sneak/routewatch/issues/226 fixed that on `next`.
-https://git.eeqj.de/sneak/routewatch/issues/225, also on `next`, takes the peers
-whose sessions go down together out of the table in one walk. A 24-hour run of
-`9bd0e91` on the live feed is checking whether any handler still drops updates.
+A 24-hour run of `26d939a` on the live feed is checking whether any handler
+still drops updates.
+
+The README rewrite, https://git.eeqj.de/sneak/routewatch/pulls/210, has failed
+three reviews. It is paused until sneak has reviewed it. It is part of
+https://git.eeqj.de/sneak/routewatch/issues/157.
 
 Two questions wait for sneak: which history the prefix page keeps
 (https://git.eeqj.de/sneak/routewatch/issues/161), and which fields the IP
@@ -50,42 +50,17 @@ https://git.eeqj.de/sneak/routewatch/issues/86.
 
 # Next Step
 
-Post the outcome of the 24-hour run of `9bd0e91` on
-https://git.eeqj.de/sneak/routewatch/issues/226: whether any handler still drops
+Post the outcome of the 24-hour run of `26d939a` on
+https://git.eeqj.de/sneak/routewatch/issues/216: whether any handler still drops
 updates, and the process's peak memory against 5 GiB. File an issue for each
 problem the run finds.
 
 # Completed Steps
 
-- 2026-11-03: the streamer drops an update for a handler only when that
-  handler's queue is full. It used to drop a random share of the updates once
-  the queue was half full. The prefix handler's queue now holds 100,000
-  messages, about 100 MiB. It used to hold 20,000. While the routing table's
-  `Overview` walks every prefix, the prefix handler applies updates more slowly.
-  In a run on the live feed, the feed paused during such a walk and then sent
-  its backlog. The streamer then dropped about 10,000 of the prefix handler's
-  updates (closes #229)
-- 2026-11-02: `README.md` is rewritten from scratch, sentence by sentence,
-  against the rules of https://git.eeqj.de/sneak/routewatch/issues/157. Every
-  statement was checked against the code on `next` (closes #157)
 - 2026-11-02: `TODO.md` is rewritten from scratch, sentence by sentence, against
   the rules of https://git.eeqj.de/sneak/routewatch/issues/157. Every Completed
   Steps entry keeps its date, its facts and its issue. Status and Next Step
   describe the repo as the tracker shows it (part of #157)
-- 2026-11-02: when the sessions of several peers go down, the prefix handler
-  takes the peers out of the routing table together. The table's `PeersDown`
-  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 since the previous one
-  began. Each peer used to have a walk of its own. A peer's messages stayed held
-  back while the walks queued ahead of its own ran. The handler holds back at
-  most 150,000 messages, about 150 MiB. It used to hold back at most 20,000. The
-  status page shows how many messages the handler holds back (closes #225)
-- 2026-11-02: working out the home page's cards no longer stops the prefix
-  handler. The walk over the routing table holds the table's lock for 10,000
-  prefixes at a time, then lets the lock go for as long again, as `PeerDown`
-  does. It used to take the lock back at once. The prefix handler then applied
-  about one update per 10,000 prefixes walked. During a walk, its queue could
-  fill, and the streamer then dropped updates for it (closes #226)
 - 2026-11-02: A skipped WHOIS lookup now makes the fetcher wait longer before
   the next one, as a failed lookup does. A lookup is skipped when its RIR is
   paused. An RIR is one of the five regional Internet registries. The longest
@@ -432,3 +407,8 @@ problem the run finds.
 - 2025-07-27: The first commit: RouteWatch, a BGP stream monitor
 
 # Future Steps
+
+- Finish https://git.eeqj.de/sneak/routewatch/pulls/210 as sneak's review
+  directs. It merges once a review passes. Of the two PRs for
+  https://git.eeqj.de/sneak/routewatch/issues/157, the one that merges last
+  closes the issue.

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.PeersDown([]string{peerA})
+	table.PeerDown(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.PeersDown([]string{peerC})
+	source.PeerDown(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.PeersDown([]string{peerB})
+		table.PeerDown(peerB)
 		announceTo(table, tablePrefixShortV4, asB, peerD, 50)
 	}
 

internal/routetable/overview.go

@@ -79,9 +79,8 @@ type OriginPeers struct {
 
 // Overview works out the table's overview. It lists at most maxListed bogon
 // announcements and at most maxListed prefixes with more than one origin AS:
-// the first in prefix order. It 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.
+// the first in prefix order. It walks over every prefix, but holds the read
+// lock for walkBatch prefixes at a time.
 func (t *Table) Overview(maxListed int) Overview {
 	var overview Overview
 
@@ -111,18 +110,15 @@ func (t *Table) Overview(maxListed int) Overview {
 }
 
 // walkPrefixes calls visit with every prefix of the table and its origins. It
-// holds the read lock for walkBatch prefixes at a time. 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, updates can then take the lock about half
-// of the time. The walk may or may not visit a prefix that is added or removed
-// while the lock is let go. Go allows this in a range over a map that changes.
-// visit must not keep origins once it returns.
+// holds the read lock, but lets it go after every walkBatch prefixes, so the
+// updates waiting for it get in. A prefix that is added or removed in such a
+// gap may or may not be visited, as Go allows for a map that changes during a
+// range over it. visit must not keep origins once it returns.
 func (t *Table) walkPrefixes(visit func(prefix netip.Prefix, origins []*origin)) {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
 
 	visited := 0
-	batchStart := time.Now()
 
 	for _, f := range []*family{&t.ipv4, &t.ipv6} {
 		for _, prefixes := range f.byLength {
@@ -131,13 +127,8 @@ func (t *Table) walkPrefixes(visit func(prefix netip.Prefix, origins []*origin))
 
 				visited++
 				if visited%walkBatch == 0 {
-					held := time.Since(batchStart)
-
 					t.mu.RUnlock()
-					time.Sleep(held)
 					t.mu.RLock()
-
-					batchStart = time.Now()
 				}
 			}
 		}

internal/routetable/overview_internal_test.go

@@ -131,61 +131,3 @@ func TestWalkPrefixesLetsUpdatesIn(t *testing.T) {
 			"before its second batch")
 	}
 }
-
-// TestOverviewLetsUpdatesIn applies another peer's 1,000 announcements while
-// Overview walks a table with more prefixes than five of the walk's batches.
-// After each batch, the walk lets the lock go for as long as it held the lock.
-// So all 1,000 announcements must be applied before Overview returns. A walk
-// that takes the lock back at once lets in about one update per batch.
-func TestOverviewLetsUpdatesIn(t *testing.T) {
-	t.Parallel()
-
-	const (
-		asn       = 1000
-		peer      = "192.0.2.1"
-		otherPeer = "192.0.2.2"
-		// prefixCount is more prefixes than five of the walk's batches.
-		prefixCount   = 5*walkBatch + 1
-		announcements = 1000
-	)
-
-	routeTime := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
-	table := New(Params{})
-	prefixes := make([]netip.Prefix, 0, prefixCount)
-	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})
-		prefixes = append(prefixes, prefix)
-		addr = addr.Next()
-	}
-
-	overviewDone := make(chan struct{})
-
-	go func() {
-		defer close(overviewDone)
-
-		table.Overview(0)
-	}()
-
-	// Wait until Overview holds the read lock. TryLock fails while a reader
-	// holds it.
-	for table.mu.TryLock() {
-		table.mu.Unlock()
-		runtime.Gosched()
-	}
-
-	for _, prefix := range prefixes[:announcements] {
-		table.Announce(prefix, asn, Route{Peer: otherPeer, Time: routeTime})
-	}
-
-	select {
-	case <-overviewDone:
-		t.Fatal("Overview returned before another peer's 1,000 announcements " +
-			"were applied")
-	default:
-	}
-
-	<-overviewDone
-}

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.PeersDown([]string{peerC})
+	table.PeerDown(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 PeersDown each walk the whole table.
+	// one hold of the lock. Overview and PeerDown each walk the whole table.
 	// Callers waiting for the lock get it between batches.
 	walkBatch = 10000
 
@@ -205,29 +205,20 @@ func (t *Table) Withdraw(prefix netip.Prefix, 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) {
+// 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) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
-	// 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 {
+	number, known := t.peerNumbers[peer]
+	if !known || t.peerPrefixes[number] == 0 {
 		return
 	}
 
@@ -237,15 +228,9 @@ func (t *Table) PeersDown(peers []string) {
 	for _, f := range []*family{&t.ipv4, &t.ipv6} {
 		for _, prefixes := range f.byLength {
 			for prefix, origins := range prefixes {
-				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]
-					}
+				o := originWithPeer(origins, number)
+				if o != nil {
+					t.leave(prefix, o, number)
 				}
 
 				// Go allows the map to change before the next step of

internal/routetable/routetable_internal_test.go

@@ -7,11 +7,11 @@ import (
 	"time"
 )
 
-// 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) {
+// 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 (
@@ -32,8 +32,8 @@ func TestPeersDownLetsReadsIn(t *testing.T) {
 		addr = addr.Next()
 	}
 
-	// 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
+	// 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()
 
@@ -42,7 +42,7 @@ func TestPeersDownLetsReadsIn(t *testing.T) {
 	go func() {
 		defer close(peerDownDone)
 
-		table.PeersDown([]string{peer})
+		table.PeerDown(peer)
 	}()
 
 	for table.mu.TryRLock() {
@@ -60,7 +60,7 @@ func TestPeersDownLetsReadsIn(t *testing.T) {
 
 		select {
 		case <-peerDownDone:
-			t.Fatal("no read got in while PeersDown ran")
+			t.Fatal("no read got in while PeerDown ran")
 		default:
 		}
 	}

internal/routetable/routetable_test.go

@@ -69,10 +69,10 @@ func withdraw(peer string) step {
 	}
 }
 
-// peersDown is the step of the sessions of peers going down.
-func peersDown(peers ...string) step {
+// peerDown is the step of peer's session going down.
+func peerDown(peer string) step {
 	return func(table *routetable.Table, _ netip.Prefix) {
-		table.PeersDown(peers)
+		table.PeerDown(peer)
 	}
 }
 
@@ -272,28 +272,21 @@ func TestWithdrawalsAndPeerDowns(t *testing.T) {
 		{
 			name: "a peer going down leaves the prefix",
 			steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
-				peersDown(peerB)},
+				peerDown(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), peersDown(peerB)},
+				withdraw(peerB), peerDown(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 TestPeersDownLeavesEveryPrefix(t *testing.T) {
+func TestPeerDownLeavesEveryPrefix(t *testing.T) {
 	t.Parallel()
 
 	table := routetable.New(routetable.Params{})
@@ -310,7 +303,7 @@ func TestPeersDownLeavesEveryPrefix(t *testing.T) {
 	}
 
 	table.Announce(prefixes[0], asB, routeOf(peerB, asB, 20))
-	table.PeersDown([]string{peerA})
+	table.PeerDown(peerA)
 
 	want := []routetable.Origin{{ASN: asB, Peers: []string{peerB},
 		Routes: []routetable.Route{routeOf(peerB, asB, 20)}}}
@@ -340,32 +333,6 @@ func TestPeersDownLeavesEveryPrefix(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()
 
@@ -391,7 +358,7 @@ func TestCountsOriginsWithoutKeptRoute(t *testing.T) {
 		{announce(peerC, asA, 30), 0},
 		{withdraw(peerB), 0},
 		{withdraw(peerC), 1},
-		{peersDown(peerA), 0},
+		{peerDown(peerA), 0},
 	}
 
 	for i, s := range steps {
@@ -497,7 +464,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
 	checkChanges(t, table, []change{}, map[string]uint{})
 
 	// peerA going down changes v4, the one prefix it carries.
-	table.PeersDown([]string{peerA})
+	table.PeerDown(peerA)
 	checkChanges(t, table, []change{
 		{Prefix: prefixV4, OriginASN: asA, Peers: []uint{1, 2},
 			Routes: []routetable.Route{
@@ -506,7 +473,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
 	}, map[string]uint{})
 
 	// peerA, which carries nothing now, goes down again.
-	table.PeersDown([]string{peerA})
+	table.PeerDown(peerA)
 	checkChanges(t, table, []change{}, map[string]uint{})
 }
 
@@ -650,7 +617,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.PeersDown([]string{peerB}) },
+		func(int) { table.PeerDown(peerB) },
 		func(int) { table.PutBackChanges(table.TakeChanges(math.MaxInt)) },
 		func(int) { table.ChangeCount() },
 		func(int) { table.Lookup(prefix.Addr()) },

internal/routewatch/prefixhandler.go

@@ -15,15 +15,12 @@ import (
 )
 
 const (
-	// prefixHandlerQueueSize is the most messages the handler's queue holds.
-	// While the routing table's Overview or PeersDown walks every prefix, it
-	// holds the table's lock about half of the time. The handler then applies
-	// updates more slowly. When the feed pauses for some seconds and then
-	// catches up, its backlog can arrive faster than that, and the queue
-	// holds the backlog. The streamer drops an update that finds the queue
-	// full. A message takes about 1 KiB, so a full queue takes about 100 MiB
-	// (README "Memory").
-	prefixHandlerQueueSize = 100000
+	// prefixHandlerQueueSize is the queue capacity for prefix tracking
+	// operations, about 4 seconds of feed at peak. The streamer drops updates
+	// rather than blocks when a queue is full (a STATE message is held until
+	// there is room), so this bounds memory. Batches still flush on a timer
+	// (prefixBatchTimeout).
+	prefixHandlerQueueSize = 20000
 
 	// prefixBatchSize is the number of prefix updates to batch together
 	prefixBatchSize = 25000
@@ -49,19 +46,10 @@ const (
 	maxTableChangesPerWrite = 20000
 
 	// maxHeldBackMessages is the most messages the handler holds back from
-	// peers whose session went down. A peer's messages are held back from
-	// the moment its session goes down until peerDown has applied them. If
-	// peerDown is running at that moment, the peer waits for the rest of
-	// peerDown's current PeersDown call. The peer also waits while peerDown
-	// applies the messages held back from that call's peers, those that
-	// arrive meanwhile included. The next PeersDown call takes the peer out.
-	// Once that call returns, peerDown applies the messages held back from
-	// that call's peers, one peer after another, in the order the call lists
-	// them. A held-back message that says the peer's session went down again
-	// makes the peer's later messages wait for the next call. 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
+	// 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
 )
 
 // PrefixHandler keeps the routing table in memory current from the feed. It
@@ -70,16 +58,14 @@ 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 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 since the
-// previous one began. 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 peerDown applies them,
-// as the doc of maxHeldBackMessages describes. It applies them in order, one at
-// a time, and takes new messages in between. It holds back at most
-// maxHeldBackMessages messages.
+// 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.
 //
 // 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.
@@ -117,25 +103,20 @@ type PrefixHandler struct {
 	stopCh chan context.Context
 	done   chan struct{} // closed when the flush goroutine returns
 
-	// 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.
+	// 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.
 	// heldTaken is signalled each time a held-back message is taken out to
-	// 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)
+	// 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
 }
 
 // seenPrefix is what the batch keeps of an update: the prefix it names and its
@@ -200,7 +181,6 @@ 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
@@ -232,7 +212,7 @@ func (h *PrefixHandler) QueueCapacity() int {
 }
 
 // HandleMessage processes a message to track prefix information. A message
-// from a peer whose session went down is held back until peerDown applies it.
+// 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()
@@ -255,16 +235,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.applyMessage(msg)
 }
 
-// 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
+// Stop waits for every PeerDown 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
@@ -319,8 +290,8 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
 }
 
 // applyMessage applies msg. The caller holds mu. A peer's session going down
-// has the peer leave every prefix. An update is applied to the routing table,
-// and its prefixes join the batch.
+// 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 {
@@ -399,64 +370,34 @@ func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// 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.
+// 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.
 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()
+	go h.peerDown(peer)
 }
 
-// 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() {
+// 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) {
 	defer h.peerDowns.Done()
 
-	for {
-		h.mu.Lock()
+	h.peerDownMu.Lock()
+	h.table.PeerDown(peer)
+	h.peerDownMu.Unlock()
 
-		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 PeersDown is held up and one message
+// 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()
@@ -249,8 +249,8 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
 		routetable.New(routetable.Params{}), longerThanAnyTest,
 		maxTableChangesPerWrite)
 
-	release := holdUpPeersDown(h)
-	defer close(release)
+	h.peerDownMu.Lock()
+	defer h.peerDownMu.Unlock()
 
 	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 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
+// 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
 // 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)
 
-	release := holdUpPeersDown(h)
+	h.peerDownMu.Lock()
 	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 PeersDown was held up")
+		t.Error("Stop returned while a PeerDown was held up")
 	case <-time.After(stillWaitingAfter):
 	}
 
-	close(release)
+	h.peerDownMu.Unlock()
 
 	select {
 	case <-stopped:
 	case <-time.After(stepTimeout):
-		t.Fatalf("Stop did not return within %s of the PeersDown's release",
+		t.Fatalf("Stop did not return within %s of the PeerDown's release",
 			stepTimeout)
 	}
 
@@ -324,32 +324,16 @@ 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:                table,
+		table:                routetable.New(routetable.Params{}),
 		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,25 +103,24 @@ func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) {
 	})
 }
 
-// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeersDown of a
-// peer whose session went down, as another peer's long PeersDown would. The
+// 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 these messages of the down peer, in this order:
 //   - an announcement;
 //   - its session going down again;
 //   - a second announcement.
 //
-// HeldBackCount must count them. Once the PeersDown calls have returned, the
-// down peer must carry only the prefix of the 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()
 
-	const heldMessages = 3
-
 	h := prefixHandlerWithoutTimer(newMockStore())
-	release := holdUpPeersDown(h)
 	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
 
+	h.peerDownMu.Lock()
+
 	handled := returnsInTime(func() {
 		for _, msg := range []*ristypes.RISMessage{
 			sessionDown(),
@@ -145,15 +144,10 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
 		setV4Prefix:   {},
 	})
 
-	if h.HeldBackCount() != heldMessages {
-		t.Errorf("HeldBackCount() = %d, want %d", h.HeldBackCount(),
-			heldMessages)
-	}
-
-	close(release)
+	h.peerDownMu.Unlock()
 
 	if !returnsInTime(h.peerDowns.Wait) {
-		t.Fatalf("the PeersDown calls did not return within %s", stepTimeout)
+		t.Fatalf("the PeerDowns did not return within %s", stepTimeout)
 	}
 
 	checkTableOrigins(t, h.table, map[string]map[int][]string{
@@ -164,85 +158,21 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
 	})
 }
 
-// TestPrefixHandlerTakesPeersDownTogether has four peers announce testV4Prefix.
-// It holds up the PeersDown of the first peer, whose session goes down.
-// Meanwhile the sessions of the three others go down, and each of them
-// announces otherV4Prefix. Once the first PeersDown is let go, the next one
-// must take the three others together. No peer may then carry testV4Prefix,
-// and the three others must carry otherV4Prefix.
-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
+// TestPrefixHandlerAppliesMessagesWhilePeerDownRuns gives a peer 50,000
 // prefixes and hands the prefix handler that peer's session going down. The
-// 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
+// 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 TestPrefixHandlerAppliesMessagesWhilePeersDownRuns(t *testing.T) {
+func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
 	t.Parallel()
 
 	const (
-		// downPeerPrefixes is five times the 10,000 prefixes PeersDown visits
+		// 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 PeersDown runs, if PeersDown takes the lock back at once after
+		// 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.
@@ -254,7 +184,7 @@ func TestPrefixHandlerAppliesMessagesWhilePeersDownRuns(t *testing.T) {
 
 	h.HandleMessage(sessionDown())
 
-	// Wait until PeersDown has begun: the down peer has left some of its
+	// 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 {
@@ -277,76 +207,15 @@ func TestPrefixHandlerAppliesMessagesWhilePeersDownRuns(t *testing.T) {
 	}
 
 	if counts.IPv4Routes == 0 {
-		t.Errorf("PeersDown ended before the handler applied the other peer's "+
+		t.Errorf("PeerDown ended before the handler applied the other peer's "+
 			"%d announcements", otherPeerMessages)
 	}
 
 	if !returnsInTime(h.peerDowns.Wait) {
-		t.Fatalf("the PeersDown did not return within %s", stepTimeout)
+		t.Fatalf("the PeerDown 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
@@ -368,14 +237,14 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
 
 	defer stopInTime(t, h)
 
-	release := holdUpPeersDown(h)
+	h.peerDownMu.Lock()
 	h.HandleMessage(sessionDown())
 
 	for range heldMessages {
 		h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix))
 	}
 
-	close(release)
+	h.peerDownMu.Unlock()
 
 	select {
 	case <-logs.firstWrite:
@@ -399,10 +268,10 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
 	}
 }
 
-// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeersDown of a peer
+// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeerDown 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 PeersDown returns and a held-back message is applied.
+// message, until the PeerDown returns and a held-back message is applied.
 func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
 	t.Parallel()
 
@@ -411,7 +280,7 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
 		maxTableChangesPerWrite)
 	defer stopInTime(t, h)
 
-	release := holdUpPeersDown(h)
+	h.peerDownMu.Lock()
 	h.HandleMessage(sessionDown())
 
 	held := peerAnnouncement(testPeerIP, testV6Prefix)
@@ -433,12 +302,12 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
 	case <-time.After(stillWaitingAfter):
 	}
 
-	close(release)
+	h.peerDownMu.Unlock()
 
 	select {
 	case <-returned:
 	case <-time.After(stepTimeout):
-		t.Fatal("HandleMessage did not return after the PeersDown returned")
+		t.Fatal("HandleMessage did not return after the PeerDown returned")
 	}
 }
 
@@ -555,7 +424,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 PeersDown calls have
+// 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()

internal/server/handlers.go

@@ -486,7 +486,6 @@ 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"`
@@ -571,7 +570,6 @@ 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,43 +295,6 @@ 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/server/middleware_internal_test.go

@@ -261,7 +261,7 @@ func checkAnswerCarriesMeta(
 ) {
 	t.Helper()
 
-	// The values README "JSON API" gives the two fields that never change.
+	// The values README "API v1" gives the two fields that never change.
 	const (
 		wantTimeZone   = "UTC"
 		wantAPIVersion = 1

internal/streamer/streamer.go

@@ -12,6 +12,7 @@ import (
 	"fmt"
 	"io"
 	"math"
+	"math/rand"
 	"net/http"
 	"sync"
 	"sync/atomic"
@@ -57,6 +58,10 @@ const (
 	// Well above an extended BGP message (RFC 8654) in hex plus its prefixes
 	maxLineSize           = bytesPerMB
 	maxConcurrentHandlers = 800 // Maximum number of concurrent message handlers
+
+	// Backpressure constants
+	backpressureThreshold = 0.5 // Start dropping at 50% queue utilization
+	backpressureSlope     = 2.0 // Slope for linear drop probability increase
 )
 
 // messageTypeNotification is the type of a BGP NOTIFICATION message, which ends
@@ -88,8 +93,9 @@ type MessageHandler interface {
 
 	// QueueCapacity returns the desired queue capacity for this handler.
 	// Handlers that process quickly can have larger queues to buffer bursts.
-	// While the queue is full, the streamer drops the handler's messages,
-	// except STATE messages, which are held until there is room.
+	// When the queue fills up, messages will be dropped according to the
+	// backpressure algorithm, except STATE messages, which are held until
+	// there is room.
 	QueueCapacity() int
 }
 
@@ -120,8 +126,8 @@ type handlerInfo struct {
 
 // Streamer manages a connection to the RIPE RIS Live streaming API for receiving
 // real-time BGP UPDATE messages. It handles automatic reconnection with exponential
-// backoff and dispatches messages to registered handlers via per-handler queues.
-// Its read loop never waits for room on a queue (see queueMessage).
+// backoff, dispatches messages to registered handlers via per-handler queues, and
+// implements backpressure to prevent queue overflow during high traffic periods.
 type Streamer struct {
 	logger *logger.Logger
 	client *http.Client
@@ -138,6 +144,7 @@ type Streamer struct {
 	workers      sync.WaitGroup // the handler workers; Stop waits for them
 	metrics      *metrics.Tracker
 	totalDropped atomic.Uint64   // Total dropped messages across all handlers
+	random       *rand.Rand      // Random number generator for backpressure drops
 	bgpPeers     map[string]bool // Track active BGP peers by peer IP
 	bgpPeersMu   sync.RWMutex    // Protects bgpPeers map
 }
@@ -161,6 +168,8 @@ func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer {
 		},
 		handlers: make([]*handlerInfo, 0),
 		metrics:  metrics,
+		//nolint:gosec // Non-cryptographic randomness is fine for backpressure
+		random:   rand.New(rand.NewSource(time.Now().UnixNano())),
 		bgpPeers: make(map[string]bool),
 	}
 }
@@ -342,10 +351,6 @@ 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
@@ -387,11 +392,6 @@ 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)
 	}
 
@@ -399,7 +399,7 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
 }
 
 // GetDroppedMessages returns the total number of messages dropped across all
-// handlers because their queue was full. This counter is monotonically
+// handlers due to queue overflow or backpressure. This counter is monotonically
 // increasing and is safe to call concurrently.
 func (s *Streamer) GetDroppedMessages() uint64 {
 	return s.totalDropped.Load()
@@ -882,13 +882,13 @@ func (s *Streamer) dispatchMessage(
 // msg instead. It never waits for room, so a full queue does not hold up the
 // read loop.
 //
-// A message that finds the queue full is dropped, except a STATE message. A
-// STATE message whose state is down is the only notice that all of a peer's
-// routes are gone. A STATE message that finds the queue full is held in
-// info.heldState instead. Held messages go into the queue, oldest first, when a
-// later message for the handler finds room, or when Stop is called. While any
-// message is held, the handler's other messages are dropped, so none of them
-// overtakes a held STATE message.
+// A message is dropped while the queue is filling up (shouldDropForBackpressure)
+// or full, except a STATE message: one whose state is down is the only notice
+// that all of a peer's routes are gone. A STATE message skips the random drop,
+// and one that finds the queue full is held in info.heldState. Held messages
+// go into the queue, oldest first, before anything else for the handler, as
+// soon as there is room, or when Stop is called; while any is still held, the
+// handler's other messages are dropped, so none of them overtakes a held down.
 //
 // The held list stays small: a STATE message is a few hundred bytes and the
 // feed sends about ten a second, so a queue full for a minute holds about 600.
@@ -907,7 +907,7 @@ func (s *Streamer) queueMessage(
 		return true
 	}
 
-	if len(info.heldState) > 0 {
+	if len(info.heldState) > 0 || s.shouldDropForBackpressure(info) {
 		return false
 	}
 
@@ -923,3 +923,26 @@ func trySend(queue chan *ristypes.RISMessage, msg *ristypes.RISMessage) bool {
 		return false
 	}
 }
+
+// shouldDropForBackpressure determines if a message should be dropped based on
+// queue utilization
+func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool {
+	// Calculate queue utilization
+	queueLen := len(info.queue)
+	queueCap := cap(info.queue)
+	utilization := float64(queueLen) / float64(queueCap)
+
+	// No drops below threshold
+	if utilization < backpressureThreshold {
+		return false
+	}
+
+	// Calculate drop probability (0.0 at threshold, 1.0 at 100% full)
+	dropProbability := (utilization - backpressureThreshold) * backpressureSlope
+	if dropProbability > 1.0 {
+		dropProbability = 1.0
+	}
+
+	// Random drop based on probability
+	return s.random.Float64() < dropProbability
+}

internal/streamer/streamer_internal_test.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"io"
 	"log/slog"
 	"net/http"
@@ -259,65 +258,6 @@ func TestHeldStateMessagesKeepTheirPlaceInTheQueue(t *testing.T) {
 	}
 }
 
-// burstQueueCapacity is the size of a burstHandler's queue.
-const burstQueueCapacity = 1000
-
-// burstHandler wants UPDATE messages and has a queue of burstQueueCapacity.
-// Tests read what it is handed from its queue, so HandleMessage does nothing.
-type burstHandler struct{}
-
-func (burstHandler) WantsMessage(messageType string) bool {
-	return messageType == ristypes.MessageTypeUpdate
-}
-
-func (burstHandler) QueueCapacity() int { return burstQueueCapacity }
-
-func (burstHandler) HandleMessage(*ristypes.RISMessage) {}
-
-// TestUpdatesDroppedOnlyWhenTheQueueIsFull has the server send one UPDATE more
-// than the handler's queue holds, each with its own timestamp. Nothing takes
-// messages off the queue meanwhile. The streamer must queue the updates in
-// feed order until the queue is full, and drop only the last one.
-func TestUpdatesDroppedOnlyWhenTheQueueIsFull(t *testing.T) {
-	t.Parallel()
-
-	const (
-		firstTimestamp = 1700000000
-		lineFormat     = `{"type":"ris_message","data":{"type":"UPDATE",` +
-			`"peer":"192.0.2.1","peer_asn":"64496","timestamp":%d}}` + "\n"
-	)
-
-	var lines strings.Builder
-
-	for i := range burstQueueCapacity + 1 {
-		fmt.Fprintf(&lines, lineFormat, firstTimestamp+i)
-	}
-
-	s := New(logger.New(), metrics.New())
-	s.RegisterHandler(burstHandler{})
-
-	streamLines(t, s, lines.String())
-
-	dropped := s.GetHandlerStats()[0].DroppedCount
-	if dropped != 1 {
-		t.Errorf("%d updates dropped, want only the one that found the queue "+
-			"full", dropped)
-	}
-
-	queue := s.handlers[0].queue
-	if len(queue) != burstQueueCapacity {
-		t.Fatalf("%d updates queued, want %d", len(queue), burstQueueCapacity)
-	}
-
-	for i := range burstQueueCapacity {
-		msg := <-queue
-		if msg.Timestamp != float64(firstTimestamp+i) {
-			t.Fatalf("update %d on the queue has timestamp %.0f, want %d",
-				i, msg.Timestamp, firstTimestamp+i)
-		}
-	}
-}
-
 // takeAll takes the messages off queue, as a handler's worker would, and sends
 // them on the returned channel once queue is closed.
 func takeAll(queue chan *ristypes.RISMessage) chan []*ristypes.RISMessage {

internal/templates/status.html

@@ -541,10 +541,6 @@
                         <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>