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

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

PeerDown now lets the lock go after every 10,000 prefixes. The handler
runs it on a goroutine of its own, one peer at a time, and holds back
the down peer's later messages until it returns, then applies them in
order. Stop waits for this before its last flush, and logs the
held-back messages it could not apply.

Model: opus-5-5

README.md

@@ -429,12 +429,10 @@ 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 a batch when it reaches a size, or after a time:
+      write 30,000, 10,000 and 25,000 updates at a time, or after 2, 2 and 1
-        - the AS handler at 30,000 ASes, or after 2 seconds;
+      seconds; the peering handler writes the peerings it has collected every 30
-        - the peer handler at 10,000 updates, or after 2 seconds;
+      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
@@ -445,8 +443,8 @@ the routing table, and everything else from the database.
         - 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. The handler
-          holds back the peer's later updates until the peer has left every
+          holds back the peer's later updates until it has left them all, then
-          prefix, then applies them in order.
+          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
@@ -582,8 +580,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 what the
+    - The rest of the heap takes up to about 0.5 GiB. Most of it is the
-      handlers hold, described below.
+      handlers' batches and queues, 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.
@@ -636,25 +634,16 @@ 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.
+  database, the updates that arrive collect in its next batch. That batch stops
-    - The AS handler's batch holds one entry per AS, with the earliest and the
+  at ten times the batch size:
-      latest time of its updates. About 80,000 ASes appear in the feed, so the
+    - 300,000 updates for the AS handler (`asnBatchMaxSize`, about 10 MiB);
-      batch takes at most about 10 MiB, however fast the feed runs.
+    - 100,000 for the peer handler (`peerBatchMaxSize`, about 10 MiB);
-    - The peer handler's batch stops at 100,000 updates (`peerBatchMaxSize`),
+    - 250,000 for the prefix handler (`prefixBatchMaxSize`, about 15 MiB).
-      about 10 MiB.
-    - The prefix handler's batch stops at 250,000 updates
-      (`prefixBatchMaxSize`), about 15 MiB.
 
-    At that maximum the peer or prefix handler takes nothing more off its queue
+    At that size the handler takes nothing more off its queue until the write
-    until the write ends. The queue then fills, and the streamer drops messages
+    ends. The queue then fills, and the streamer drops messages as above. With
-    as above. With the batches being written, the three handlers' batches take
+    the batches being written, the three handlers' batches take at most about
-    at most about 0.1 GiB of the Go heap.
+    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

@@ -36,20 +36,12 @@ 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
+  The handler runs the table's `PeerDown` on a goroutine of its own, one peer at
-  time, and goes on taking messages. It holds back the down peer's later
+  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,
+  messages until its `PeerDown` returns, then applies them in order. `PeerDown`
-  one at a time, and takes new messages in between. It holds back at most 20,000
+  lets the table's lock go after every 10,000 prefixes, so the IP lookup and the
-  messages. At that number, `HandleMessage` waits until one of them is applied.
+  pages no longer wait for the whole visit. A stop past its deadline logs the
-  `PeerDown` lets the table's lock go after every 10,000 prefixes, so the IP
+  held-back messages it did not apply (closes #216)
-  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/routetable.go

@@ -28,8 +28,8 @@ const (
 	keptTimesDegree = 32
 
 	// walkBatch is how many prefixes a walk over the whole table visits in
-	// one hold of the lock. Overview and PeerDown each walk the whole table.
+	// one hold of the lock: Overview's walk and PeerDown's. The others
-	// Callers waiting for the lock get it between batches.
+	// waiting for the lock get it in between.
 	walkBatch = 10000
 
 	ipv4Bits = 32
@@ -207,10 +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. It lets the table's lock go after every
+// visits every prefix in the table. It lets the lock go after every walkBatch
-// walkBatch prefixes, so that waiting reads and updates get in. An
+// prefixes, so the reads and updates waiting for it get in. An announcement of
-// announcement by the peer made meanwhile may be dropped too. So the caller
+// the peer made meanwhile may be dropped too, so the caller applies the peer's
-// applies the peer's later updates only once PeerDown has returned.
+// later updates only once PeerDown has returned.
 func (t *Table) PeerDown(peer string) {
 	t.mu.Lock()
 	defer t.mu.Unlock()

internal/routewatch/app_integration_internal_test.go

@@ -644,8 +644,9 @@ 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: pathOfASes(asnBatchSize)}})
+		[]*ristypes.RISMessage{{Path: fill}})
 
 	ctx, cancel := context.WithTimeout(t.Context(), deadline)
 	defer cancel()

internal/routewatch/ashandler.go

@@ -18,34 +18,43 @@ const (
 	// (asnBatchTimeout), so a queue smaller than asnBatchSize is fine.
 	asHandlerQueueSize = 20000
 
-	// asnBatchSize is the number of ASes in a full batch, which wakes the
+	// asnBatchSize is the number of ASN operations to batch together
-	// 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 records every AS in the AS paths of BGP UPDATE messages. Its batch
+// ASHandler processes Autonomous System Number (ASN) information extracted from
-// holds the earliest and the latest time of each AS's updates. The flush
+// BGP UPDATE messages. It uses batched database operations to efficiently store
-// goroutine writes the batch to the database when it holds asnBatchSize ASes,
+// ASN data, collecting operations into batches that are flushed either when the
-// and every asnBatchTimeout.
+// batch reaches a size threshold or after a timeout period.
 type ASHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// Batching. mu guards batch and writing only. The flush goroutine takes
+	// batchMaxSize is asnBatchMaxSize. Only HandleMessage reads it, so a test
-	// the batch under mu and writes it after releasing mu. So HandleMessage
+	// may lower it before it hands the handler a message.
-	// never waits for a database write.
+	batchMaxSize int
+
+	// Batching. mu guards batch and writing only: the flush goroutine takes
+	// the batch under mu and writes it after releasing mu, so HandleMessage
+	// waits for a database write only while batch is at batchMaxSize.
 	mu    sync.Mutex
-	// batch holds the earliest and the latest time of the updates for each AS
+	batch []asnOp
-	// since the flush goroutine last took it. It holds one entry per AS,
+	// writing is how many operations the flush goroutine took from batch
-	// however many updates arrive.
+	// and has not finished writing.
-	batch map[int]database.SeenTimes
-	// writing is how many ASes the flush goroutine took from batch and has
-	// not finished writing.
 	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.
@@ -53,10 +62,16 @@ 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
-// writes the collected ASes to the database. The caller must call Stop when
+// flushes accumulated ASN operations to the database. The caller must call
-// finished, to write the pending ASes and end the background goroutine.
+// Stop when finished to ensure all pending operations are flushed and the
+// background goroutine is terminated.
 func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
 	return newASHandler(db, logger, asnBatchTimeout)
 }
@@ -70,11 +85,13 @@ func newASHandler(
 	h := &ASHandler{
 		db:           db,
 		logger:       logger,
-		batch:       make(map[int]database.SeenTimes),
+		batchMaxSize: asnBatchMaxSize,
+		batch:        make([]asnOp, 0, asnBatchSize),
 		batchFullCh:  make(chan struct{}, 1),
 		stopCh:       make(chan context.Context, 1),
 		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)
@@ -98,25 +115,48 @@ func (h *ASHandler) QueueCapacity() int {
 	return asHandlerQueueSize
 }
 
-// HandleMessage adds the time of msg to the batch entry of each AS in its AS
+// HandleMessage processes a RIS Live BGP message by extracting all ASNs from
-// path, the origin AS and the members of an AS set included. When the batch
+// the AS path and queuing them for batch insertion into the database. The
-// holds asnBatchSize ASes, HandleMessage wakes the flush goroutine to write it.
+// origin ASN and all transit ASNs are recorded with their associated
+// 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()
 
-	// The origin AS is the last AS of the sequence, so the loop adds it too.
+	// 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.
+	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 {
-			seen := h.batch[asn]
+			h.batch = append(h.batch, asnOp{
-			seen.Add(msg.ParsedTimestamp)
+				number:    asn,
-			h.batch[asn] = seen
+				timestamp: timestamp,
+			})
 		}
 	}
 
 	// A full batch wakes the flush goroutine. If it is writing, the wake-up
-	// waits in batchFullCh and the batch keeps taking updates until the write
+	// waits in batchFullCh and the batch keeps growing until the write ends
-	// ends.
+	// or the batch reaches batchMaxSize.
 	if len(h.batch) >= asnBatchSize {
 		select {
 		case h.batchFullCh <- struct{}{}:
@@ -125,11 +165,10 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// Stop stops the flush goroutine and waits for it until ctx ends. The flush
+// Stop stops the flush goroutine, which writes the operations the handler holds
-// goroutine writes the batch before it returns, unless ctx has ended. A write
+// before it returns, and waits for it until ctx ends. Once ctx has ended the
-// still running when the process exits is lost. Stop logs how many entries are
+// flush goroutine skips that last write, and a write still running is lost
-// not written, in the batch being written and in the next batch. An AS in both
+// when the process exits, so Stop logs how many operations are not written.
-// batches counts twice.
 func (h *ASHandler) Stop(ctx context.Context) {
 	h.stopCh <- ctx
 
@@ -144,14 +183,14 @@ func (h *ASHandler) Stop(ctx context.Context) {
 
 	if unwritten > 0 {
 		h.logger.Error("Stop deadline passed before the AS handler wrote "+
-			"its batches", "batch_entries", unwritten)
+			"its updates", "updates", unwritten)
 	}
 }
 
-// takeBatch returns the pending ASes and leaves an empty batch in their place
+// takeBatch returns the pending operations and leaves an empty batch in their
-// for HandleMessage to fill. They count in writing until flushBatch has written
+// place for HandleMessage to fill. They count in writing until flushBatch has
-// them.
+// written them.
-func (h *ASHandler) takeBatch() map[int]database.SeenTimes {
+func (h *ASHandler) takeBatch() []asnOp {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
@@ -161,7 +200,8 @@ func (h *ASHandler) takeBatch() map[int]database.SeenTimes {
 
 	batch := h.batch
 	h.writing = len(batch)
-	h.batch = make(map[int]database.SeenTimes)
+	h.batch = make([]asnOp, 0, asnBatchSize)
+	h.batchTaken.Broadcast()
 
 	return batch
 }
@@ -174,11 +214,21 @@ 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(batch)
+	err := h.db.GetOrCreateASNBatch(asns)
 	if err != nil {
 		h.logger.Error("Failed to process ASN batch",
-			"error", err, "count", len(batch))
+			"error", err, "count", len(asns))
 	}
 
 	h.mu.Lock()

internal/routewatch/ashandler_internal_test.go

@@ -5,43 +5,28 @@ 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, in one batch into a
+// whose path carries an AS the database does not hold yet, the second 5
-// real database. The second is 5 seconds after the first, and the handler gets
+// seconds after the first, in one batch into a real database. The AS is stored
-// it first. The AS is stored first seen at the time of the first update and
+// first seen at the time of the first update and last seen at the time of the
-// last seen at the time of the second.
+// second.
 func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
 	t.Parallel()
 
 	db := newTestDatabase(t)
 
 	// Built without NewASHandler's flush goroutine, so the test flushes itself.
-	h := &ASHandler{
+	h := &ASHandler{db: db, logger: logger.New(), batchMaxSize: asnBatchMaxSize}
-		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(second)
 	h.HandleMessage(first)
+	h.HandleMessage(second)
 	h.flushBatch()
 
 	asn, _, err := db.GetAS(t.Context(), testASNA)
@@ -69,9 +54,7 @@ func TestASHandlerRecordsNothingFromAPathWithASZero(t *testing.T) {
 	store := newMockStore()
 
 	// Built without NewASHandler's flush goroutine, so the test flushes itself.
-	h := &ASHandler{
+	h := &ASHandler{db: store, logger: logger.New(), batchMaxSize: asnBatchMaxSize}
-		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,7 +5,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"log/slog"
 	"slices"
 	"sync"
@@ -180,12 +179,11 @@ type stopLog struct {
 	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 io.Writer) *logger.Logger {
+func newErrorLogger(logs *bytes.Buffer) *logger.Logger {
 	return &logger.Logger{Logger: slog.New(slog.NewJSONHandler(logs,
 		&slog.HandlerOptions{Level: slog.LevelError}))}
 }
@@ -498,20 +496,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()
 
-	const wantEntries = asnBatchSize + 1
+	// Each message gives an update for its origin AS and one for each AS
+	// 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: pathOfASes(asnBatchSize)}},
+		[]*ristypes.RISMessage{{Path: fill}},
 		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
 
 	store.mu.Lock()
@@ -521,27 +519,26 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 		t.Errorf("AS%d was written after the stop deadline", testASNB)
 	}
 
-	if logged := lastLogged(t, &logs); logged.BatchEntries != wantEntries {
+	if logged := lastLogged(t, &logs); logged.Updates != wantUpdates {
-		t.Errorf("Stop logged %d batch entries not written, want %d",
+		t.Errorf("Stop logged %d updates not written, want %d",
-			logged.BatchEntries, wantEntries)
+			logged.Updates, wantUpdates)
 	}
 }
 
 // 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 full batch holds the fill's ASes again, and testASNB. The stop
+// waits. The stop deadline passes meanwhile. When the write ends, the flush
-// deadline passes meanwhile. When the write ends, the flush goroutine must stop
+// goroutine must stop instead of writing the batch.
-// instead of writing the batch.
 func TestASHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
 	t.Parallel()
 
 	store := newBlockingStore()
 	h := newASHandler(store, logger.New(), longerThanAnyTest)
-	fill := &ristypes.RISMessage{Path: pathOfASes(asnBatchSize)}
+	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
+	full := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)}
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
-		[]*ristypes.RISMessage{fill},
+		[]*ristypes.RISMessage{{Path: fill}},
-		fill,
+		&ristypes.RISMessage{Path: full})
-		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
 
 	store.mu.Lock()
 	defer store.mu.Unlock()
@@ -707,8 +704,9 @@ 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: pathOfASes(asnBatchSize)}},
+		[]*ristypes.RISMessage{{Path: fill}},
 		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
 
 	store.mu.Lock()
@@ -765,41 +763,25 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	}
 }
 
-// TestASHandlerNeverWaitsForAWrite starts a write that the store holds. Then it
+func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
-// 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}}})
 
-	defer stopInTime(t, h)
+	store.mu.Lock()
-	defer store.releaseWrites()
+	defer store.mu.Unlock()
 
-	startBlockedWrite(t, store, h,
+	if !store.asns[testASNB] {
-		[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}})
+		t.Errorf("Stop did not write AS%d, handled after the batch reached its "+
-
+			"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))
 	}
 }
 
@@ -863,11 +845,13 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			batchSize: asnBatchSize,
 			handle: func(n int) chan struct{} {
 				h := &ASHandler{
-					batch:       make(map[int]database.SeenTimes),
+					batchMaxSize: asnBatchMaxSize,
 					batchFullCh:  make(chan struct{}, 1),
 				}
-				// Each of the path's n ASes is one entry of the batch.
+				// The origin AS and each AS in the path are one update each.
-				h.HandleMessage(&ristypes.RISMessage{Path: pathOfASes(n)})
+				h.HandleMessage(&ristypes.RISMessage{
+					Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)},
+				})
 
 				return h.batchFullCh
 			},

internal/routewatch/prefixhandler.go

@@ -44,12 +44,6 @@ 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
@@ -59,12 +53,10 @@ const (
 // 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
+// which takes over a second on a full table. So the handler runs it on a
-// a separate goroutine, one peer at a time, and goes on taking messages
+// goroutine of its own, one peer at a time, and goes on taking messages
-// meanwhile. The handler holds back the down peer's later messages until
+// meanwhile. It holds back the peer's later messages until PeerDown returns,
-// PeerDown returns. Then it applies them in order, one at a time, and takes
+// then applies them in order.
-// 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.
@@ -107,14 +99,9 @@ type PrefixHandler struct {
 	// 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.
+	// peerDowns counts the peerDown goroutines that have not returned.
-	// 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
 }
 
@@ -182,7 +169,6 @@ func newPrefixHandler(
 		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)
@@ -216,17 +202,9 @@ 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
 	}
@@ -234,17 +212,16 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.applyMessage(msg)
 }
 
-// Stop waits for every PeerDown the handler started and for the messages held
+// Stop waits until ctx ends for every PeerDown the handler started, and for
-// back meanwhile to be applied. It gives up this wait once ctx ends. Then Stop
+// the messages it held back meanwhile to be applied. Then it stops the flush
-// stops the flush goroutine and waits for it until ctx ends. The flush
+// goroutine, which makes one last flush before it returns, and waits for it
-// goroutine makes one last flush before it returns. That flush writes at most
+// until ctx ends. That flush writes at most tableChangesPerWrite of the
-// tableChangesPerWrite of the routing table's changes, and the prefixes of the
+// routing table's changes, and the prefixes of the updates the handler holds.
-// updates the handler holds. Once ctx has ended the flush goroutine skips that
+// Once ctx has ended the flush goroutine skips that last flush, and a write
-// last flush, and a write still running is lost when the process exits. A
+// still running is lost when the process exits. A failed write of the table's
-// failed write of the table's changes is not retried either. Stop logs how
+// changes is not retried either. Stop logs how many updates and how many of
-// many updates and table changes are not written. It also logs how many
+// the table's changes are not written, and how many held-back messages are not
-// held-back messages are not applied. The caller calls Stop once its last
+// applied. The caller calls Stop once its last HandleMessage has returned.
-// HandleMessage has returned.
 func (h *PrefixHandler) Stop(ctx context.Context) {
 	peerDownsDone := make(chan struct{})
 
@@ -268,7 +245,11 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
 	h.mu.Lock()
 	unwritten := h.writing + len(h.batch)
 	tableChanges := h.writingTableChanges + h.table.ChangeCount()
-	heldMessages := h.heldCount
+
+	heldMessages := 0
+	for _, held := range h.heldBack {
+		heldMessages += len(held)
+	}
 	h.mu.Unlock()
 
 	if unwritten == 0 && tableChanges == 0 && heldMessages == 0 {
@@ -288,9 +269,9 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
 		"held_messages", heldMessages)
 }
 
-// applyMessage applies msg. The caller holds mu. A peer's session going down
+// applyMessage applies msg, with mu held. A peer's session going down starts
-// starts the peer's PeerDown. An update is applied to the routing table, and
+// the peer's PeerDown. An update is applied to the routing table, and its
-// its prefixes join the batch.
+// prefixes join the batch.
 func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
 	if msg.Type == ristypes.MessageTypeState {
 		if msg.State == ristypes.StateDown {
@@ -369,10 +350,9 @@ func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// startPeerDown starts peerDown for peer on a new goroutine. The caller holds
+// startPeerDown runs peerDown for peer on a goroutine of its own, with mu
-// mu. The peer's messages are held back until peerDown has applied them.
+// held. The peer's messages are held back until it returns. Those held back
-// Messages held back for an earlier PeerDown of the peer, and not yet applied,
+// for an earlier PeerDown of the peer and not yet applied stay held.
-// stay held back.
 func (h *PrefixHandler) startPeerDown(peer string) {
 	_, goingDown := h.heldBack[peer]
 	if !goingDown {
@@ -384,12 +364,10 @@ func (h *PrefixHandler) startPeerDown(peer string) {
 	go h.peerDown(peer)
 }
 
-// peerDown runs the table's PeerDown for peer, once no other PeerDown is
+// peerDown runs the table's PeerDown for peer once no other peer's runs. Then
-// running. Then it applies the messages held back from the peer, in order. It
+// it applies the messages held back from the peer, in order. When the peer's
-// takes mu for one message at a time, so HandleMessage takes new messages in
+// session went down again among them, the PeerDown that starts applies the
-// between. The peer's new messages join the end of the held-back ones. A
+// rest.
-// 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()
 
@@ -397,30 +375,25 @@ func (h *PrefixHandler) peerDown(peer string) {
 	h.table.PeerDown(peer)
 	h.peerDownMu.Unlock()
 
-	for {
 	h.mu.Lock()
+	defer h.mu.Unlock()
 
-		held := h.heldBack[peer]
+	// applyMessage lets mu go while it waits for the batch. HandleMessage
-		if len(held) == 0 {
+	// then adds the peer's new messages after these.
-			delete(h.heldBack, peer)
+	for len(h.heldBack[peer]) > 0 {
-			h.mu.Unlock()
+		msg := h.heldBack[peer][0]
-
+		h.heldBack[peer] = h.heldBack[peer][1:]
-			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 {
+			// applyMessage started the next PeerDown, which applies the
+			// rest.
 			return
 		}
 	}
+
+	delete(h.heldBack, peer)
 }
 
 // takeBatch returns the pending updates and leaves an empty batch in their

internal/routewatch/prefixhandler_internal_test.go

@@ -267,53 +267,6 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
 	}
 }
 
-// TestPrefixHandlerStopWaitsForTheMessagesHeldBack stops a prefix handler
-// while a peer's PeerDown is held up and an announcement of the peer is held
-// back. Stop must wait. Once the PeerDown is let go and Stop returns, the
-// 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

internal/routewatch/prefixhandler_table_internal_test.go

@@ -9,7 +9,6 @@ import (
 	"reflect"
 	"slices"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
@@ -154,115 +153,6 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
 	})
 }
 
-// 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