Keep taking messages while a peer's session goes down (closes #216)

The table's PeerDown visits every prefix under the write lock, which
takes over a second on a full table. The prefix handler called it
inline, so its queue filled and the streamer dropped updates.

PeerDown now lets the lock go after every 10,000 prefixes. The handler
runs it on a separate goroutine, one peer at a time. It holds back the
down peer's later messages until that PeerDown returns. Then it applies
them in order, one at a time, and takes new messages in between. It
holds back at most 20,000 messages. Stop waits for all of this before
its last flush.

Model: opus-5-5

README.md

@@ -429,10 +429,12 @@ the routing table, and everything else from the database.
     - Parses BGP UPDATE messages from JSON stream
     - Dispatches to registered handlers based on message type
     - Each handler has its own queue with backpressure handling
-    - Handlers batch writes for efficiency: the AS, peer and prefix handlers
+    - Handlers batch writes for efficiency. The AS, peer and prefix handlers
-      write 30,000, 10,000 and 25,000 updates at a time, or after 2, 2 and 1
+      write a batch when it reaches a size, or after a time:
-      seconds; the peering handler writes the peerings it has collected every 30
+        - the AS handler at 30,000 ASes, or after 2 seconds;
-      seconds
+        - the peer handler at 10,000 updates, or after 2 seconds;
+        - the prefix handler at 25,000 updates, or after 1 second.
+    - The peering handler writes the peerings it has collected every 30 seconds
 
 4. **Handler Details**
     - **ASHandler**: Tracks all ASNs seen in AS paths
@@ -442,7 +444,9 @@ the routing table, and everything else from the database.
         - An announcement gives the peer a route to the prefix.
         - A withdrawal removes the peer's route. So does an announcement whose
           AS path is empty or holds AS 0 anywhere.
-        - A peer whose BGP session goes down leaves every prefix.
+        - A peer whose BGP session goes down leaves every prefix. The handler
+          holds back the peer's later updates until the peer has left every
+          prefix, then applies them in order.
 
         Every second it writes the changed prefix and origin AS pairs to the
         `routes` table. A write takes at most 20,000 pairs, and the rest follow
@@ -578,8 +582,8 @@ Budget:
 - Go heap: a 2.5 GiB soft limit (`GOMEMLIMIT=2560MiB`, set in the image).
     - The routing table takes about 700 bytes per prefix and origin AS. A full
       table has about 1.4 million of these pairs, so it takes about 1 GB.
-    - The rest of the heap takes up to about 0.5 GiB. Most of it is the
+    - The rest of the heap takes up to about 0.5 GiB. Most of it is what the
-      handlers' batches and queues, described below.
+      handlers hold, described below.
     - The remaining 1 GiB is room for the garbage collector. With much less, it
       would run almost all the time.
 - SQLite: 656 MiB of page cache.
@@ -632,16 +636,25 @@ What happens at each limit:
   `STATE` message, which it holds, in feed order, until there is room: a few
   hundred bytes each, about ten a second.
 - Handler batches: while the AS, peer or prefix handler writes a batch to the
-  database, the updates that arrive collect in its next batch. That batch stops
+  database, the updates that arrive collect in its next batch.
-  at ten times the batch size:
+    - The AS handler's batch holds one entry per AS, with the earliest and the
-    - 300,000 updates for the AS handler (`asnBatchMaxSize`, about 10 MiB);
+      latest time of its updates. About 80,000 ASes appear in the feed, so the
-    - 100,000 for the peer handler (`peerBatchMaxSize`, about 10 MiB);
+      batch takes at most about 10 MiB, however fast the feed runs.
-    - 250,000 for the prefix handler (`prefixBatchMaxSize`, about 15 MiB).
+    - The peer handler's batch stops at 100,000 updates (`peerBatchMaxSize`),
+      about 10 MiB.
+    - The prefix handler's batch stops at 250,000 updates
+      (`prefixBatchMaxSize`), about 15 MiB.
 
-    At that size the handler takes nothing more off its queue until the write
+    At that maximum the peer or prefix handler takes nothing more off its queue
-    ends. The queue then fills, and the streamer drops messages as above. With
+    until the write ends. The queue then fills, and the streamer drops messages
-    the batches being written, the three handlers' batches take at most about
+    as above. With the batches being written, the three handlers' batches take
-    0.1 GiB of the Go heap.
+    at most about 0.1 GiB of the Go heap.
+
+- 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.
 
 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,21 @@ tested.
 
 # Completed Steps
 
+- 2026-11-02: a peer's session going down no longer stops the prefix handler.
+  The handler runs the table's `PeerDown` on a separate goroutine, one peer at a
+  time, and goes on taking messages. It holds back the down peer's later
+  messages until that peer's `PeerDown` returns. Then it applies them in order,
+  one at a time, and takes new messages in between. It holds back at most 20,000
+  messages. At that number, `HandleMessage` waits until one of them is applied.
+  `PeerDown` lets the table's lock go after every 10,000 prefixes, so the IP
+  lookup and the pages no longer wait for the whole visit. When `Stop` runs past
+  its deadline, it logs how many held-back messages were not applied (closes
+  #216)
+- 2026-11-02: the AS handler never waits for a database write. Its batch holds
+  one entry per AS, with the earliest and the latest time of its updates. So the
+  batch no longer grows with the feed's rate. It used to hold one entry for each
+  AS of each path and to wait once it held 300,000. In a burst of the feed its
+  queue then filled, and the streamer dropped messages for it (closes #217)
 - 2026-10-27: the tests of `internal/routewatch` no longer fail on a loaded
   host. Five of them gave `HandleMessage` two seconds to apply a full prefix
   batch, which under the race detector can take longer. A test now waits up to

internal/routetable/overview.go

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

internal/routetable/routetable.go

@@ -27,6 +27,11 @@ const (
 	// kept routes in order: each of its nodes holds up to 63 times.
 	keptTimesDegree = 32
 
+	// walkBatch is how many prefixes a walk over the whole table visits in
+	// one hold of the lock. Overview and PeerDown each walk the whole table.
+	// Callers waiting for the lock get it between batches.
+	walkBatch = 10000
+
 	ipv4Bits = 32
 	ipv6Bits = 128
 
@@ -202,7 +207,10 @@ func (t *Table) Withdraw(prefix netip.Prefix, peer string) {
 
 // PeerDown applies a peer's session going down: the peer leaves every peer
 // set, and every route of it is dropped. When the peer carries any prefix, it
-// visits every prefix in the table.
+// visits every prefix in the table. It lets the table's lock go after every
+// walkBatch prefixes, so that waiting reads and updates get in. An
+// announcement by the peer made meanwhile may be dropped too. So the caller
+// applies the peer's later updates only once PeerDown has returned.
 func (t *Table) PeerDown(peer string) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
@@ -212,6 +220,8 @@ func (t *Table) PeerDown(peer string) {
 		return
 	}
 
+	visited := 0
+
 	for _, f := range []*family{&t.ipv4, &t.ipv6} {
 		for _, prefixes := range f.byLength {
 			for prefix, origins := range prefixes {
@@ -219,6 +229,14 @@ func (t *Table) PeerDown(peer string) {
 				if o != nil {
 					t.leave(prefix, o, number)
 				}
+
+				// Go allows the map to change before the next step of
+				// the range over it.
+				visited++
+				if visited%walkBatch == 0 {
+					t.mu.Unlock()
+					t.mu.Lock()
+				}
 			}
 		}
 	}

internal/routetable/routetable_internal_test.go

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

internal/routewatch/app_integration_internal_test.go

@@ -644,9 +644,8 @@ func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) {
 	}
 
 	rw.dbHandler = NewASHandler(store, log)
-	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
 	startBlockedWrite(t, blocking, rw.dbHandler,
-		[]*ristypes.RISMessage{{Path: fill}})
+		[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}})
 
 	ctx, cancel := context.WithTimeout(t.Context(), deadline)
 	defer cancel()

internal/routewatch/ashandler.go

@@ -18,43 +18,34 @@ const (
 	// (asnBatchTimeout), so a queue smaller than asnBatchSize is fine.
 	asHandlerQueueSize = 20000
 
-	// asnBatchSize is the number of ASN operations to batch together
+	// asnBatchSize is the number of ASes in a full batch, which wakes the
+	// flush goroutine to write it.
 	asnBatchSize = 30000
 
-	// asnBatchMaxSize is the most ASN operations the batch holds while the
-	// flush goroutine writes the previous one: ten batches, about 10 MiB
-	// (README "Memory"). At this size HandleMessage waits until the flush
-	// goroutine takes the batch.
-	asnBatchMaxSize = 300000
-
 	// asnBatchTimeout is the maximum time to wait before flushing a batch
 	// DO NOT reduce this timeout - larger batches are more efficient
 	asnBatchTimeout = 2 * time.Second
 )
 
-// ASHandler processes Autonomous System Number (ASN) information extracted from
+// ASHandler records every AS in the AS paths of BGP UPDATE messages. Its batch
-// BGP UPDATE messages. It uses batched database operations to efficiently store
+// holds the earliest and the latest time of each AS's updates. The flush
-// ASN data, collecting operations into batches that are flushed either when the
+// goroutine writes the batch to the database when it holds asnBatchSize ASes,
-// batch reaches a size threshold or after a timeout period.
+// and every asnBatchTimeout.
 type ASHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// batchMaxSize is asnBatchMaxSize. Only HandleMessage reads it, so a test
+	// Batching. mu guards batch and writing only. The flush goroutine takes
-	// may lower it before it hands the handler a message.
+	// the batch under mu and writes it after releasing mu. So HandleMessage
-	batchMaxSize int
+	// never waits for a database write.
-
+	mu sync.Mutex
-	// Batching. mu guards batch and writing only: the flush goroutine takes
+	// batch holds the earliest and the latest time of the updates for each AS
-	// the batch under mu and writes it after releasing mu, so HandleMessage
+	// since the flush goroutine last took it. It holds one entry per AS,
-	// waits for a database write only while batch is at batchMaxSize.
+	// however many updates arrive.
-	mu    sync.Mutex
+	batch map[int]database.SeenTimes
-	batch []asnOp
+	// writing is how many ASes the flush goroutine took from batch and has
-	// writing is how many operations the flush goroutine took from batch
+	// not finished writing.
-	// and has not finished writing.
+	writing     int
-	writing int
-	// batchTaken is signalled when the flush goroutine takes batch; its L is
-	// &mu.
-	batchTaken  sync.Cond
 	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
 	// stopCh hands Stop's context to the flush goroutine. It holds one, so
 	// Stop does not wait for a write in progress to hand it over.
@@ -62,16 +53,10 @@ type ASHandler struct {
 	done   chan struct{} // closed when the flush goroutine returns
 }
 
-type asnOp struct {
-	number    int
-	timestamp time.Time
-}
-
 // NewASHandler creates and returns a new ASHandler instance. It initializes
 // the batching system and starts a background goroutine that periodically
-// flushes accumulated ASN operations to the database. The caller must call
+// writes the collected ASes to the database. The caller must call Stop when
-// Stop when finished to ensure all pending operations are flushed and the
+// finished, to write the pending ASes and end the background goroutine.
-// background goroutine is terminated.
 func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
 	return newASHandler(db, logger, asnBatchTimeout)
 }
@@ -83,15 +68,13 @@ func newASHandler(
 	db database.Store, logger *logger.Logger, batchTimeout time.Duration,
 ) *ASHandler {
 	h := &ASHandler{
-		db:           db,
+		db:          db,
-		logger:       logger,
+		logger:      logger,
-		batchMaxSize: asnBatchMaxSize,
+		batch:       make(map[int]database.SeenTimes),
-		batch:        make([]asnOp, 0, asnBatchSize),
+		batchFullCh: make(chan struct{}, 1),
-		batchFullCh:  make(chan struct{}, 1),
+		stopCh:      make(chan context.Context, 1),
-		stopCh:       make(chan context.Context, 1),
+		done:        make(chan struct{}),
-		done:         make(chan struct{}),
 	}
-	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
 	go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch)
@@ -115,48 +98,25 @@ func (h *ASHandler) QueueCapacity() int {
 	return asHandlerQueueSize
 }
 
-// HandleMessage processes a RIS Live BGP message by extracting all ASNs from
+// HandleMessage adds the time of msg to the batch entry of each AS in its AS
-// the AS path and queuing them for batch insertion into the database. The
+// path, the origin AS and the members of an AS set included. When the batch
-// origin ASN and all transit ASNs are recorded with their associated
+// holds asnBatchSize ASes, HandleMessage wakes the flush goroutine to write it.
-// timestamps. When the batch reaches the configured size threshold, the flush
-// goroutine is woken to write it.
 func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
-	// Use the pre-parsed timestamp
-	timestamp := msg.ParsedTimestamp
-
-	originASN := msg.Path.Origin()
-
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
-	// While the batch is at its maximum size, wait until the flush goroutine
+	// The origin AS is the last AS of the sequence, so the loop adds it too.
-	// takes it. Meanwhile the handler takes nothing off its queue, which
-	// fills, and the streamer drops messages for it.
-	for len(h.batch) >= h.batchMaxSize {
-		h.batchTaken.Wait()
-	}
-
-	// Queue origin ASN operation
-	if originASN > 0 {
-		h.batch = append(h.batch, asnOp{
-			number:    originASN,
-			timestamp: timestamp,
-		})
-	}
-
-	// Also track all ASNs in the path, the members of an AS set included
 	for _, asn := range slices.Concat(msg.Path.Sequence, msg.Path.Set) {
 		if asn > 0 {
-			h.batch = append(h.batch, asnOp{
+			seen := h.batch[asn]
-				number:    asn,
+			seen.Add(msg.ParsedTimestamp)
-				timestamp: timestamp,
+			h.batch[asn] = seen
-			})
 		}
 	}
 
 	// A full batch wakes the flush goroutine. If it is writing, the wake-up
-	// waits in batchFullCh and the batch keeps growing until the write ends
+	// waits in batchFullCh and the batch keeps taking updates until the write
-	// or the batch reaches batchMaxSize.
+	// ends.
 	if len(h.batch) >= asnBatchSize {
 		select {
 		case h.batchFullCh <- struct{}{}:
@@ -165,10 +125,11 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// Stop stops the flush goroutine, which writes the operations the handler holds
+// Stop stops the flush goroutine and waits for it until ctx ends. The flush
-// before it returns, and waits for it until ctx ends. Once ctx has ended the
+// goroutine writes the batch before it returns, unless ctx has ended. A write
-// flush goroutine skips that last write, and a write still running is lost
+// still running when the process exits is lost. Stop logs how many entries are
-// when the process exits, so Stop logs how many operations are not written.
+// not written, in the batch being written and in the next batch. An AS in both
+// batches counts twice.
 func (h *ASHandler) Stop(ctx context.Context) {
 	h.stopCh <- ctx
 
@@ -183,14 +144,14 @@ func (h *ASHandler) Stop(ctx context.Context) {
 
 	if unwritten > 0 {
 		h.logger.Error("Stop deadline passed before the AS handler wrote "+
-			"its updates", "updates", unwritten)
+			"its batches", "batch_entries", unwritten)
 	}
 }
 
-// takeBatch returns the pending operations and leaves an empty batch in their
+// takeBatch returns the pending ASes and leaves an empty batch in their place
-// place for HandleMessage to fill. They count in writing until flushBatch has
+// for HandleMessage to fill. They count in writing until flushBatch has written
-// written them.
+// them.
-func (h *ASHandler) takeBatch() []asnOp {
+func (h *ASHandler) takeBatch() map[int]database.SeenTimes {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
@@ -200,8 +161,7 @@ func (h *ASHandler) takeBatch() []asnOp {
 
 	batch := h.batch
 	h.writing = len(batch)
-	h.batch = make([]asnOp, 0, asnBatchSize)
+	h.batch = make(map[int]database.SeenTimes)
-	h.batchTaken.Broadcast()
 
 	return batch
 }
@@ -214,21 +174,11 @@ func (h *ASHandler) flushBatch() {
 		return
 	}
 
-	// The earliest and the latest time the batch saw each AS. The batch holds
-	// several updates for an AS, from many peers, so a new AS is stored first
-	// seen at the earliest of them and last seen at the latest.
-	asns := make(map[int]database.SeenTimes)
-	for _, op := range batch {
-		seen := asns[op.number]
-		seen.Add(op.timestamp)
-		asns[op.number] = seen
-	}
-
 	// Process all ASNs in a single batch transaction
-	err := h.db.GetOrCreateASNBatch(asns)
+	err := h.db.GetOrCreateASNBatch(batch)
 	if err != nil {
 		h.logger.Error("Failed to process ASN batch",
-			"error", err, "count", len(asns))
+			"error", err, "count", len(batch))
 	}
 
 	h.mu.Lock()

internal/routewatch/ashandler_internal_test.go

@@ -5,28 +5,43 @@ import (
 	"slices"
 	"testing"
 
+	"git.eeqj.de/sneak/routewatch/internal/database"
 	"git.eeqj.de/sneak/routewatch/internal/logger"
+	"git.eeqj.de/sneak/routewatch/internal/ristypes"
 )
 
+// pathOfASes returns an AS path of n different ASes: AS 1 to AS n. With n up to
+// asnBatchSize, none of them is testASNA to testASND.
+func pathOfASes(n int) ristypes.ASPath {
+	sequence := make([]int, n)
+	for i := range sequence {
+		sequence[i] = i + 1
+	}
+
+	return ristypes.ASPath{Sequence: sequence}
+}
+
 // TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS handles two updates
-// whose path carries an AS the database does not hold yet, the second 5
+// whose path carries an AS the database does not hold yet, in one batch into a
-// seconds after the first, in one batch into a real database. The AS is stored
+// real database. The second is 5 seconds after the first, and the handler gets
-// first seen at the time of the first update and last seen at the time of the
+// it first. The AS is stored first seen at the time of the first update and
-// second.
+// last seen at the time of the second.
 func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
 	t.Parallel()
 
 	db := newTestDatabase(t)
 
 	// Built without NewASHandler's flush goroutine, so the test flushes itself.
-	h := &ASHandler{db: db, logger: logger.New(), batchMaxSize: asnBatchMaxSize}
+	h := &ASHandler{
+		db: db, logger: logger.New(), batch: make(map[int]database.SeenTimes),
+	}
 
 	first := peerAnnouncement(testPeerIP, testV4Prefix)
 	second := peerAnnouncement(testPeerIP, testV4Prefix)
 	second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter)
 
-	h.HandleMessage(first)
 	h.HandleMessage(second)
+	h.HandleMessage(first)
 	h.flushBatch()
 
 	asn, _, err := db.GetAS(t.Context(), testASNA)
@@ -54,7 +69,9 @@ func TestASHandlerRecordsNothingFromAPathWithASZero(t *testing.T) {
 	store := newMockStore()
 
 	// Built without NewASHandler's flush goroutine, so the test flushes itself.
-	h := &ASHandler{db: store, logger: logger.New(), batchMaxSize: asnBatchMaxSize}
+	h := &ASHandler{
+		db: store, logger: logger.New(), batch: make(map[int]database.SeenTimes),
+	}
 
 	msg := peerAnnouncement(testPeerIP, testV4Prefix)
 	msg.Path = decodePath(t, testASNA, 0, testASNB)

internal/routewatch/batchwrite_internal_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"log/slog"
 	"slices"
 	"sync"
@@ -177,12 +178,14 @@ type stopLog struct {
 	Msg          string `json:"msg"`
 	Updates      int    `json:"updates"`
 	TableChanges int    `json:"table_changes"`
+	HeldMessages int    `json:"held_messages"`
 	Paths        int    `json:"paths"`
+	BatchEntries int    `json:"batch_entries"`
 }
 
 // newErrorLogger returns a logger that writes only errors, as JSON lines, to
 // logs.
-func newErrorLogger(logs *bytes.Buffer) *logger.Logger {
+func newErrorLogger(logs io.Writer) *logger.Logger {
 	return &logger.Logger{Logger: slog.New(slog.NewJSONHandler(logs,
 		&slog.HandlerOptions{Level: slog.LevelError}))}
 }
@@ -495,20 +498,20 @@ func TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress(t *testing.T)
 	}
 }
 
+// TestASHandlerSkipsItsLastWriteAfterTheStopDeadline also checks what Stop
+// logs: the asnBatchSize entries of the held write and the entry for testASNB,
+// handled since, are not written.
 func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	t.Parallel()
 
-	// Each message gives an update for its origin AS and one for each AS
+	const wantEntries = asnBatchSize + 1
-	// in its path: fill asnBatchSize+1, extra 2.
-	const wantUpdates = asnBatchSize + 1 + 2
 
 	var logs bytes.Buffer
 
 	store := newBlockingStore()
 	h := newASHandler(store, newErrorLogger(&logs), longerThanAnyTest)
-	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
-		[]*ristypes.RISMessage{{Path: fill}},
+		[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}},
 		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
 
 	store.mu.Lock()
@@ -518,26 +521,27 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 		t.Errorf("AS%d was written after the stop deadline", testASNB)
 	}
 
-	if logged := lastLogged(t, &logs); logged.Updates != wantUpdates {
+	if logged := lastLogged(t, &logs); logged.BatchEntries != wantEntries {
-		t.Errorf("Stop logged %d updates not written, want %d",
+		t.Errorf("Stop logged %d batch entries not written, want %d",
-			logged.Updates, wantUpdates)
+			logged.BatchEntries, wantEntries)
 	}
 }
 
 // TestASHandlerSkipsAFullBatchAfterTheStopDeadline hands the AS handler a full
 // batch while a write is held, so the batch's wake-up of the flush goroutine
-// waits. The stop deadline passes meanwhile. When the write ends, the flush
+// waits. The full batch holds the fill's ASes again, and testASNB. The stop
-// goroutine must stop instead of writing the batch.
+// deadline passes meanwhile. When the write ends, the flush goroutine must stop
+// instead of writing the batch.
 func TestASHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
 	t.Parallel()
 
 	store := newBlockingStore()
 	h := newASHandler(store, logger.New(), longerThanAnyTest)
-	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
+	fill := &ristypes.RISMessage{Path: pathOfASes(asnBatchSize)}
-	full := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)}
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
-		[]*ristypes.RISMessage{{Path: fill}},
+		[]*ristypes.RISMessage{fill},
-		&ristypes.RISMessage{Path: full})
+		fill,
+		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
 
 	store.mu.Lock()
 	defer store.mu.Unlock()
@@ -703,9 +707,8 @@ func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
 
 	store := newBlockingStore()
 	h := newASHandler(store, logger.New(), longerThanAnyTest)
-	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
 	handleWhileAWriteIsBlocked(t, store, h,
-		[]*ristypes.RISMessage{{Path: fill}},
+		[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}},
 		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
 
 	store.mu.Lock()
@@ -762,25 +765,41 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	}
 }
 
-func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
+// TestASHandlerNeverWaitsForAWrite starts a write that the store holds. Then it
+// hands the AS handler twenty messages whose paths each hold one AS
+// asnBatchSize times. HandleMessage must return each time, and the batch must
+// hold one entry, for that AS.
+func TestASHandlerNeverWaitsForAWrite(t *testing.T) {
 	t.Parallel()
 
+	const messages = 20
+
 	store := newBlockingStore()
 	h := newASHandler(store, logger.New(), longerThanAnyTest)
-	h.batchMaxSize = asnBatchSize
-	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
-	toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
-	handleAtMaxBatchSize(t, store, h,
-		[]*ristypes.RISMessage{{Path: fill}},
-		[]*ristypes.RISMessage{{Path: toMax}},
-		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
 
-	store.mu.Lock()
+	defer stopInTime(t, h)
-	defer store.mu.Unlock()
+	defer store.releaseWrites()
 
-	if !store.asns[testASNB] {
+	startBlockedWrite(t, store, h,
-		t.Errorf("Stop did not write AS%d, handled after the batch reached its "+
+		[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}})
-			"maximum size", testASNB)
+
+	oneAS := &ristypes.RISMessage{
+		Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)},
+	}
+
+	if !returnsInTime(func() {
+		for range messages {
+			h.HandleMessage(oneAS)
+		}
+	}) {
+		t.Fatal("HandleMessage did not return while a database write was blocked")
+	}
+
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	if len(h.batch) != 1 {
+		t.Errorf("the batch holds %d entries for one AS, want 1", len(h.batch))
 	}
 }
 
@@ -844,13 +863,11 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			batchSize: asnBatchSize,
 			handle: func(n int) chan struct{} {
 				h := &ASHandler{
-					batchMaxSize: asnBatchMaxSize,
+					batch:       make(map[int]database.SeenTimes),
-					batchFullCh:  make(chan struct{}, 1),
+					batchFullCh: make(chan struct{}, 1),
 				}
-				// The origin AS and each AS in the path are one update each.
+				// Each of the path's n ASes is one entry of the batch.
-				h.HandleMessage(&ristypes.RISMessage{
+				h.HandleMessage(&ristypes.RISMessage{Path: pathOfASes(n)})
-					Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)},
-				})
 
 				return h.batchFullCh
 			},

internal/routewatch/prefixhandler.go

@@ -44,6 +44,12 @@ const (
 	// most one pair. So the million pairs that a full-table peer's session
 	// going down changes are written in about a minute.
 	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
 )
 
 // PrefixHandler keeps the routing table in memory current from the feed. It
@@ -52,6 +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 PeerDown applies a session going down. It visits every prefix,
+// which takes over a second on a full table. So the handler runs PeerDown on
+// a separate goroutine, one peer at a time, and goes on taking messages
+// meanwhile. The handler holds back the down peer's later messages until
+// PeerDown returns. Then it applies them in order, one at a time, and takes
+// new messages in between. It holds back at most maxHeldBackMessages
+// messages.
+//
 // Each flush writes the table's changes to the database. It also stores each
 // prefix the updates name with the time it was first seen.
 type PrefixHandler struct {
@@ -67,7 +81,7 @@ type PrefixHandler struct {
 	// give a smaller one.
 	tableChangesPerWrite int
 
-	// Batching. mu guards batch, writing and writingTableChanges only: the
+	// Batching. mu guards batch, writing and writingTableChanges: the
 	// flush goroutine takes them under mu and writes them after releasing
 	// mu, so HandleMessage waits for a database write only while batch is at
 	// batchMaxSize.
@@ -87,6 +101,21 @@ type PrefixHandler struct {
 	// Stop does not wait for a write in progress to hand it over.
 	stopCh chan context.Context
 	done   chan struct{} // closed when the flush goroutine returns
+
+	// Peers whose session went down. peerDownMu lets one PeerDown run at a
+	// time, so the handler's own updates wait for the table's lock behind
+	// at most one. heldBack, under mu, holds each such peer's later
+	// messages, in order, until its PeerDown has returned and they are
+	// applied. A peer with no such messages yet has an empty entry.
+	// 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
 }
 
 // seenPrefix is what the batch keeps of an update: the prefix it names and its
@@ -150,8 +179,10 @@ func newPrefixHandler(
 		batchFullCh:          make(chan struct{}, 1),
 		stopCh:               make(chan context.Context, 1),
 		done:                 make(chan struct{}),
+		heldBack:             make(map[string][]*ristypes.RISMessage),
 	}
 	h.batchTaken.L = &h.mu
+	h.heldTaken.L = &h.mu
 
 	// Start the flush timer goroutine
 	go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch)
@@ -179,11 +210,91 @@ func (h *PrefixHandler) QueueCapacity() int {
 	return prefixHandlerQueueSize
 }
 
-// HandleMessage processes a message to track prefix information
+// HandleMessage processes a message to track prefix information. A message
+// from a peer whose PeerDown has not returned is held back until it has.
 func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	// While maxHeldBackMessages messages are held back, wait until one of them
+	// is taken out to be applied. Meanwhile the handler takes nothing off its
+	// queue, as at batchMaxSize.
+	for h.heldCount >= maxHeldBackMessages {
+		h.heldTaken.Wait()
+	}
+
+	held, goingDown := h.heldBack[msg.Peer]
+	if goingDown {
+		h.heldBack[msg.Peer] = append(held, msg)
+		h.heldCount++
+
+		return
+	}
+
+	h.applyMessage(msg)
+}
+
+// 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
+// tableChangesPerWrite of the routing table's changes, and the prefixes of the
+// updates the handler holds. Once ctx has ended the flush goroutine skips that
+// last flush, and a write still running is lost when the process exits. A
+// failed write of the table's changes is not retried either. Stop logs how
+// many updates and table changes are not written. It also logs how many
+// held-back messages are not applied. The caller calls Stop once its last
+// HandleMessage has returned.
+func (h *PrefixHandler) Stop(ctx context.Context) {
+	peerDownsDone := make(chan struct{})
+
+	go func() {
+		h.peerDowns.Wait()
+		close(peerDownsDone)
+	}()
+
+	select {
+	case <-peerDownsDone:
+	case <-ctx.Done():
+	}
+
+	h.stopCh <- ctx
+
+	select {
+	case <-h.done:
+	case <-ctx.Done():
+	}
+
+	h.mu.Lock()
+	unwritten := h.writing + len(h.batch)
+	tableChanges := h.writingTableChanges + h.table.ChangeCount()
+	heldMessages := h.heldCount
+	h.mu.Unlock()
+
+	if unwritten == 0 && tableChanges == 0 && heldMessages == 0 {
+		return
+	}
+
+	if ctx.Err() != nil {
+		h.logger.Error("Stop deadline passed before the prefix handler wrote "+
+			"its updates", "updates", unwritten, "table_changes", tableChanges,
+			"held_messages", heldMessages)
+
+		return
+	}
+
+	h.logger.Error("The prefix handler stopped before it wrote everything",
+		"updates", unwritten, "table_changes", tableChanges,
+		"held_messages", heldMessages)
+}
+
+// applyMessage applies msg. 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.
+func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
 	if msg.Type == ristypes.MessageTypeState {
 		if msg.State == ristypes.StateDown {
-			h.table.PeerDown(msg.Peer)
+			h.startPeerDown(msg.Peer)
 		}
 
 		return
@@ -198,9 +309,6 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// the path after the rest of it.
 	path := slices.Concat(msg.Path.Sequence, msg.Path.Set)
 
-	h.mu.Lock()
-	defer h.mu.Unlock()
-
 	// While the batch is at its maximum size, wait until the flush goroutine
 	// takes it. Meanwhile the handler takes nothing off its queue, which
 	// fills, and the streamer drops messages for it.
@@ -261,39 +369,58 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// Stop stops the flush goroutine, which makes one last flush before it returns,
+// startPeerDown starts peerDown for peer on a new goroutine. The caller holds
-// and waits for it until ctx ends. That flush writes at most
+// mu. The peer's messages are held back until peerDown has applied them.
-// tableChangesPerWrite of the routing table's changes, and the prefixes of
+// Messages held back for an earlier PeerDown of the peer, and not yet applied,
-// the updates the handler holds. Once ctx has ended the flush goroutine skips
+// stay held back.
-// that last flush, and a write still running is lost when the process exits. A
+func (h *PrefixHandler) startPeerDown(peer string) {
-// failed write of the table's changes is not retried either. Stop logs how many
+	_, goingDown := h.heldBack[peer]
-// updates and how many of the table's changes are not written.
+	if !goingDown {
-func (h *PrefixHandler) Stop(ctx context.Context) {
+		h.heldBack[peer] = []*ristypes.RISMessage{}
-	h.stopCh <- ctx
-
-	select {
-	case <-h.done:
-	case <-ctx.Done():
 	}
 
-	h.mu.Lock()
+	h.peerDowns.Add(1)
-	unwritten := h.writing + len(h.batch)
-	tableChanges := h.writingTableChanges + h.table.ChangeCount()
-	h.mu.Unlock()
 
-	if unwritten == 0 && tableChanges == 0 {
+	go h.peerDown(peer)
-		return
+}
+
+// peerDown runs the table's PeerDown for peer, once no other PeerDown is
+// running. Then it applies the messages held back from the peer, in order. It
+// takes mu for one message at a time, so HandleMessage takes new messages in
+// between. The peer's new messages join the end of the held-back ones. A
+// held-back message that says the peer's session went down again starts a new
+// peerDown, which applies the rest.
+func (h *PrefixHandler) peerDown(peer string) {
+	defer h.peerDowns.Done()
+
+	h.peerDownMu.Lock()
+	h.table.PeerDown(peer)
+	h.peerDownMu.Unlock()
+
+	for {
+		h.mu.Lock()
+
+		held := h.heldBack[peer]
+		if len(held) == 0 {
+			delete(h.heldBack, peer)
+			h.mu.Unlock()
+
+			return
+		}
+
+		msg := held[0]
+		held[0] = nil // so that the message is freed once applied
+		h.heldBack[peer] = held[1:]
+		h.heldCount--
+		h.heldTaken.Broadcast()
+		h.applyMessage(msg)
+		h.mu.Unlock()
+
+		if msg.Type == ristypes.MessageTypeState &&
+			msg.State == ristypes.StateDown {
+			return
+		}
 	}
-
-	if ctx.Err() != nil {
-		h.logger.Error("Stop deadline passed before the prefix handler wrote "+
-			"its updates", "updates", unwritten, "table_changes", tableChanges)
-
-		return
-	}
-
-	h.logger.Error("The prefix handler stopped before it wrote everything",
-		"updates", unwritten, "table_changes", tableChanges)
 }
 
 // takeBatch returns the pending updates and leaves an empty batch in their

internal/routewatch/prefixhandler_internal_test.go

@@ -237,6 +237,83 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
 	}
 }
 
+// TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a
+// deadline that has passed, while a peer's PeerDown is held up and one message
+// of the peer is held back. Stop must log that message as not applied.
+func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
+	t.Parallel()
+
+	var logs bytes.Buffer
+
+	h := newPrefixHandler(newMockStore(), newErrorLogger(&logs),
+		routetable.New(routetable.Params{}), longerThanAnyTest,
+		maxTableChangesPerWrite)
+
+	h.peerDownMu.Lock()
+	defer h.peerDownMu.Unlock()
+
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
+	h.HandleMessage(sessionDown())
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV6Prefix))
+
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	h.Stop(ctx)
+
+	logged := lastLogged(t, &logs)
+	if logged.HeldMessages != 1 {
+		t.Errorf("Stop logged %d held messages not applied, want 1",
+			logged.HeldMessages)
+	}
+}
+
+// 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
+// announcement must be in the database.
+func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
+	t.Parallel()
+
+	db := newTestDatabase(t)
+	h := newPrefixHandler(db, logger.New(), routetable.New(routetable.Params{}),
+		longerThanAnyTest, maxTableChangesPerWrite)
+
+	h.peerDownMu.Lock()
+	h.HandleMessage(sessionDown())
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
+
+	stopped := make(chan struct{})
+
+	go func() {
+		h.Stop(t.Context())
+		close(stopped)
+	}()
+
+	select {
+	case <-stopped:
+		t.Error("Stop returned while a PeerDown was held up")
+	case <-time.After(stillWaitingAfter):
+	}
+
+	h.peerDownMu.Unlock()
+
+	select {
+	case <-stopped:
+	case <-time.After(stepTimeout):
+		t.Fatalf("Stop did not return within %s of the PeerDown's release",
+			stepTimeout)
+	}
+
+	read, err := db.ReadRouteTable(t.Context())
+	if err != nil {
+		t.Fatalf("ReadRouteTable: %v", err)
+	}
+
+	checkTableOrigins(t, read, map[string]map[int][]string{
+		testV4Prefix: {testASNB: {testPeerIP}},
+	})
+}
+
 // longerThanAnyTest is a timer interval that no test lasts. A test gives it to
 // newPrefixHandler, newASHandler, newPeerHandler or newPeeringHandler when the
 // handler must write only when its batch is full, when the test says, or on
@@ -253,6 +330,7 @@ func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
 		table:                routetable.New(routetable.Params{}),
 		batchMaxSize:         prefixBatchMaxSize,
 		tableChangesPerWrite: maxTableChangesPerWrite,
+		heldBack:             make(map[string][]*ristypes.RISMessage),
 	}
 }
 

internal/routewatch/prefixhandler_table_internal_test.go

@@ -9,6 +9,7 @@ import (
 	"reflect"
 	"slices"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -101,6 +102,167 @@ func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) {
 	})
 }
 
+// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeerDown of a
+// peer whose session went down, as another peer's long PeerDown would. The
+// handler must go on taking messages. It applies another peer's announcement
+// at once. It holds back the down peer's own messages: an announcement, its
+// session going down again, and a second announcement. Once the PeerDowns
+// have returned, the down peer must carry only the prefix of the second
+// announcement.
+func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
+	t.Parallel()
+
+	h := prefixHandlerWithoutTimer(newMockStore())
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
+
+	h.peerDownMu.Lock()
+
+	handled := returnsInTime(func() {
+		for _, msg := range []*ristypes.RISMessage{
+			sessionDown(),
+			peerAnnouncement(secondPeerIP, otherV4Prefix),
+			peerAnnouncement(testPeerIP, testV6Prefix),
+			sessionDown(),
+			peerAnnouncement(testPeerIP, setV4Prefix),
+		} {
+			h.HandleMessage(msg)
+		}
+	})
+	if !handled {
+		t.Fatalf("the handler did not take the messages after the session "+
+			"went down within %s", stepTimeout)
+	}
+
+	checkTableOrigins(t, h.table, map[string]map[int][]string{
+		testV4Prefix:  {testASNB: {testPeerIP}},
+		otherV4Prefix: {testASNB: {secondPeerIP}},
+		testV6Prefix:  {},
+		setV4Prefix:   {},
+	})
+
+	h.peerDownMu.Unlock()
+
+	if !returnsInTime(h.peerDowns.Wait) {
+		t.Fatalf("the PeerDowns did not return within %s", stepTimeout)
+	}
+
+	checkTableOrigins(t, h.table, map[string]map[int][]string{
+		testV4Prefix:  {},
+		otherV4Prefix: {testASNB: {secondPeerIP}},
+		testV6Prefix:  {},
+		setV4Prefix:   {testASNB: {testPeerIP}},
+	})
+}
+
+// TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many
+// messages of a peer whose session went down. Each names a prefix that does
+// not parse, so the handler logs an error as it applies it. Each log write
+// takes a millisecond. While the handler applies the held-back messages, it
+// must take another peer's announcement.
+func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
+	t.Parallel()
+
+	const (
+		// heldMessages take the handler at least a second to apply.
+		heldMessages = 1000
+		badPrefix    = "not a prefix"
+	)
+
+	logs := &slowWriter{firstWrite: make(chan struct{})}
+	h := newPrefixHandler(newMockStore(), newErrorLogger(logs),
+		routetable.New(routetable.Params{}), longerThanAnyTest,
+		maxTableChangesPerWrite)
+
+	defer stopInTime(t, h)
+
+	h.peerDownMu.Lock()
+	h.HandleMessage(sessionDown())
+
+	for range heldMessages {
+		h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix))
+	}
+
+	h.peerDownMu.Unlock()
+
+	select {
+	case <-logs.firstWrite:
+	case <-time.After(stepTimeout):
+		t.Fatal("the handler did not start to apply the held-back messages")
+	}
+
+	if !returnsInTime(func() {
+		h.HandleMessage(peerAnnouncement(secondPeerIP, otherV4Prefix))
+	}) {
+		t.Fatalf("HandleMessage did not return within %s", stepTimeout)
+	}
+
+	h.mu.Lock()
+	left := len(h.heldBack[testPeerIP])
+	h.mu.Unlock()
+
+	if left == 0 {
+		t.Error("the handler took another peer's announcement only once it " +
+			"had applied every held-back message")
+	}
+}
+
+// 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 PeerDown returns and a held-back message is applied.
+func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
+	t.Parallel()
+
+	h := newPrefixHandler(newMockStore(), logger.New(),
+		routetable.New(routetable.Params{}), longerThanAnyTest,
+		maxTableChangesPerWrite)
+	defer stopInTime(t, h)
+
+	h.peerDownMu.Lock()
+	h.HandleMessage(sessionDown())
+
+	held := peerAnnouncement(testPeerIP, testV6Prefix)
+	for range maxHeldBackMessages {
+		h.HandleMessage(held)
+	}
+
+	returned := make(chan struct{})
+
+	go func() {
+		h.HandleMessage(peerAnnouncement(secondPeerIP, otherV4Prefix))
+		close(returned)
+	}()
+
+	select {
+	case <-returned:
+		t.Error("HandleMessage returned while maxHeldBackMessages messages " +
+			"were held back")
+	case <-time.After(stillWaitingAfter):
+	}
+
+	h.peerDownMu.Unlock()
+
+	select {
+	case <-returned:
+	case <-time.After(stepTimeout):
+		t.Fatal("HandleMessage did not return after the PeerDown returned")
+	}
+}
+
+// slowWriter takes a millisecond over each write and drops what it is given.
+// It closes firstWrite at the first write.
+type slowWriter struct {
+	firstWrite chan struct{}
+	once       sync.Once
+}
+
+func (w *slowWriter) Write(p []byte) (int, error) {
+	w.once.Do(func() { close(w.firstWrite) })
+	time.Sleep(time.Millisecond)
+
+	return len(p), nil
+}
+
 // TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable hands the
 // prefix handler announcements whose AS path gives no origin. One whose path
 // is empty or holds AS 0 removes the peer's route, as a withdrawal does; one
@@ -200,7 +362,8 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
 }
 
 // checkTableCases hands the messages of each case to a new prefix handler and
-// checks its routing table right after, with no flush.
+// checks its routing table once the PeerDowns it started have returned, with
+// no flush.
 func checkTableCases(t *testing.T, tests []tableCase) {
 	t.Helper()
 
@@ -213,17 +376,28 @@ func checkTableCases(t *testing.T, tests []tableCase) {
 				h.HandleMessage(msg)
 			}
 
-			for prefix, want := range tt.want {
+			h.peerDowns.Wait()
-				got := tableOrigins(h.table, prefix)
+			checkTableOrigins(t, h.table, tt.want)
-				if !reflect.DeepEqual(got, want) {
-					t.Errorf("peers of %s by origin AS = %v, want %v",
-						prefix, got, want)
-				}
-			}
 		})
 	}
 }
 
+// checkTableOrigins checks, for each prefix in want, the peers that carry it
+// in table with each origin AS.
+func checkTableOrigins(
+	t *testing.T, table *routetable.Table, want map[string]map[int][]string,
+) {
+	t.Helper()
+
+	for prefix, wantOrigins := range want {
+		got := tableOrigins(table, prefix)
+		if !reflect.DeepEqual(got, wantOrigins) {
+			t.Errorf("peers of %s by origin AS = %v, want %v",
+				prefix, got, wantOrigins)
+		}
+	}
+}
+
 // tableWritesStore is a real database that hands each write of the routing
 // table's changes to written once it is made.
 type tableWritesStore struct {
@@ -498,6 +672,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) {
 		h.HandleMessage(msg)
 	}
 
+	h.peerDowns.Wait()
 	h.flushBatch()
 
 	var logs bytes.Buffer
@@ -536,6 +711,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) {
 		restarted.HandleMessage(msg)
 	}
 
+	h.peerDowns.Wait()
 	restarted.Stop(t.Context())
 
 	read, err := db.ReadRouteTable(t.Context())