Keep one entry per AS in the AS handler's batch (closes #217)

HandleMessage added one entry to the batch for every AS of every path,
and waited for the write in progress once the batch held 300,000. In a
burst of the feed its queue then filled, and the streamer dropped
messages.

The batch is now a map from AS number to the earliest and the latest
time of its updates. It holds one entry per AS, so it no longer grows
with the feed's rate. HandleMessage never waits for a write.
asnBatchSize now counts ASes. Stop logs the entries of the batch being
written and of the next batch under "batch_entries". An AS in both
counts twice.

Model: opus-5-5

README.md

@@ -638,14 +638,15 @@ What happens at each limit:
     - The AS handler's batch holds one entry per AS, with the earliest and the
       latest time of its updates. About 80,000 ASes appear in the feed, so the
       batch takes at most about 10 MiB, however fast the feed runs.
-    - The peer and prefix handlers' batches stop at ten times their batch size:
+    - The peer handler's batch stops at 100,000 updates (`peerBatchMaxSize`),
-      100,000 updates for the peer handler (`peerBatchMaxSize`, about 10 MiB),
+      about 10 MiB.
-      and 250,000 for the prefix handler (`prefixBatchMaxSize`, about 15 MiB).
+    - The prefix handler's batch stops at 250,000 updates
-      At that size the handler takes nothing more off its queue until the write
+      (`prefixBatchMaxSize`), about 15 MiB.
-      ends. The queue then fills, and the streamer drops messages as above.
 
-    With the batches being written, the three handlers' batches take at most
+    At that maximum the peer or prefix handler takes nothing more off its queue
-    about 0.1 GiB of the Go heap.
+    until the write ends. The queue then fills, and the streamer drops messages
+    as above. With the batches being written, the three handlers' batches take
+    at most about 0.1 GiB of the Go heap.
 
 With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
 with the goroutine count and Go memory figures.

TODO.md

@@ -36,10 +36,10 @@ tested.
 # Completed Steps
 
 - 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 it
+  one entry per AS, with the earliest and the latest time of its updates. So the
-  no longer grows with the feed's rate. It used to hold one entry for each AS of
+  batch no longer grows with the feed's rate. It used to hold one entry for each
-  each path and to wait once it held 300,000. In a burst of the feed its queue
+  AS of each path and to wait once it held 300,000. In a burst of the feed its
-  then filled, and the streamer dropped messages for it (closes #217)
+  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/routewatch/ashandler.go

@@ -27,16 +27,16 @@ const (
 	asnBatchTimeout = 2 * time.Second
 )
 
-// ASHandler records every AS in the AS paths of BGP UPDATE messages. It
+// ASHandler records every AS in the AS paths of BGP UPDATE messages. Its batch
-// collects, for each AS, the earliest and the latest time of its updates in a
+// holds the earliest and the latest time of each AS's updates. The flush
-// batch. The flush goroutine writes the batch to the database when it holds
+// goroutine writes the batch to the database when it holds asnBatchSize ASes,
-// asnBatchSize ASes, and every asnBatchTimeout.
+// and every asnBatchTimeout.
 type ASHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// Batching. mu guards batch and writing only: the flush goroutine takes
+	// Batching. mu guards batch and writing only. The flush goroutine takes
-	// the batch under mu and writes it after releasing mu, so HandleMessage
+	// the batch under mu and writes it after releasing mu. So HandleMessage
 	// never waits for a database write.
 	mu sync.Mutex
 	// batch holds the earliest and the latest time of the updates for each AS
@@ -56,8 +56,7 @@ type ASHandler struct {
 // 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
-// finished to ensure all pending ASes are written and the background goroutine
+// finished, to write the pending ASes and end the background goroutine.
-// is terminated.
 func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
 	return newASHandler(db, logger, asnBatchTimeout)
 }
@@ -101,7 +100,7 @@ func (h *ASHandler) QueueCapacity() int {
 
 // HandleMessage adds the time of msg to the batch entry of each AS in its AS
 // path, the origin AS and the members of an AS set included. When the batch
-// holds asnBatchSize ASes, it wakes the flush goroutine to write it.
+// holds asnBatchSize ASes, HandleMessage wakes the flush goroutine to write it.
 func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -126,10 +125,11 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// Stop stops the flush goroutine, which writes the ASes the handler holds before
+// Stop stops the flush goroutine and waits for it until ctx ends. The flush
-// it returns, and waits for it until ctx ends. Once ctx has ended the flush
+// goroutine writes the batch before it returns, unless ctx has ended. A write
-// goroutine skips that last write, and a write still running is lost when the
+// still running when the process exits is lost. Stop logs how many entries are
-// process exits, so Stop logs how many ASes 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
 
@@ -144,7 +144,7 @@ func (h *ASHandler) Stop(ctx context.Context) {
 
 	if unwritten > 0 {
 		h.logger.Error("Stop deadline passed before the AS handler wrote "+
-			"its ASes", "asns", unwritten)
+			"its batches", "batch_entries", unwritten)
 	}
 }
 

internal/routewatch/batchwrite_internal_test.go

@@ -178,7 +178,7 @@ type stopLog struct {
 	Updates      int    `json:"updates"`
 	TableChanges int    `json:"table_changes"`
 	Paths        int    `json:"paths"`
-	ASNs         int    `json:"asns"`
+	BatchEntries int    `json:"batch_entries"`
 }
 
 // newErrorLogger returns a logger that writes only errors, as JSON lines, to
@@ -497,12 +497,12 @@ func TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress(t *testing.T)
 }
 
 // TestASHandlerSkipsItsLastWriteAfterTheStopDeadline also checks what Stop
-// logs: the asnBatchSize ASes of the held write and testASNB, handled since,
+// logs: the asnBatchSize entries of the held write and the entry for testASNB,
-// are not written.
+// handled since, are not written.
 func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 	t.Parallel()
 
-	const wantASNs = asnBatchSize + 1
+	const wantEntries = asnBatchSize + 1
 
 	var logs bytes.Buffer
 
@@ -519,9 +519,9 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 		t.Errorf("AS%d was written after the stop deadline", testASNB)
 	}
 
-	if logged := lastLogged(t, &logs); logged.ASNs != wantASNs {
+	if logged := lastLogged(t, &logs); logged.BatchEntries != wantEntries {
-		t.Errorf("Stop logged %d ASes not written, want %d",
+		t.Errorf("Stop logged %d batch entries not written, want %d",
-			logged.ASNs, wantASNs)
+			logged.BatchEntries, wantEntries)
 	}
 }