Give every JSON answer @meta, errors included (closes #201)

A panic in a JSON route's handler answered 500 with an empty body,
because the router's recoverer sits outside the middleware that adds
`@meta`. `JSONValidationMiddleware` now wraps the handler in the same
recoverer, and answers the panic with the JSON error body.

`/status.json` and the health check answered chi's empty 405 to any
method they do not take. They now answer with the same JSON 405 as the
routes under `/api/v1` and `/ip`.

The home page's IP lookup card showed only the answer's `data`. It now
shows the whole answer, `@meta` included.

A table test checks `@meta` on every JSON route, in its successes and
its errors.

Model: opus-5-5

README.md

@@ -9,38 +9,19 @@ table in real time. It works like this:
 
 ## Rationale
 
-RouteWatch exists to serve our other services as an internal API for information
+RIPE RIS Live streams the BGP updates that RIPE's route collectors receive, as
-about IP addresses.
+they happen, from the moment a client connects. It reports changes, not the
+routing table they add up to, so it cannot say which AS an IP address is routed
+to now or which routes a prefix has. RouteWatch applies the feed's updates to a
+routing table that it keeps in memory, with a copy in its own SQLite database.
+It answers IP-to-AS lookups and prefix pages from that table.
 
-RouteWatch follows BGP routing table updates from RIPE RIS Live. RIS Live
+The feed sends no table when a client connects. So the routing table holds only
-streams the updates that RIPE's route collectors receive, as they happen, from
+the routes announced since RouteWatch first started. An IP-to-AS lookup finds
-the moment a client connects. It reports changes, not the routing table they add
+nothing for an address whose routes have not been announced since then. The
-up to. RouteWatch applies those changes to a routing table that it keeps in
+routing table also misses updates RouteWatch drops when it falls behind, and
-memory, with a copy in its own SQLite database. From that table, RouteWatch sees
+updates sent while it is stopped. So a lookup or a prefix page can show a route
-directly:
+the feed has withdrawn.
-
-- which AS each IP address is routed to
-- which announced prefix holds each IP address, and how big that prefix is
-
-RouteWatch adds details of each AS from WHOIS, the public directory of the
-internet registries. The details include the organisation that holds the AS.
-RouteWatch keeps them in its database and looks them up again once they are 30
-days old. An IP lookup answers from those stored details and never waits for
-WHOIS.
-
-So our own services can ask RouteWatch about an IP address, instead of each
-working it out for itself.
-
-RIS Live sends no routing table when a client connects. So the routing table
-holds only the routes announced since RouteWatch first started. An IP lookup
-finds no route for an address until a prefix that holds it is announced. The
-routing table also misses these updates:
-
-- those sent while RouteWatch is stopped
-- those that RouteWatch drops when it falls behind
-
-A lookup or a prefix page can therefore show a route that has since been
-withdrawn.
 
 ## Features
 
@@ -429,12 +410,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
@@ -444,9 +423,7 @@ 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. The handler
+        - A peer whose BGP session goes down leaves every prefix.
-          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
@@ -582,8 +559,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 +613,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

@@ -35,41 +35,6 @@ 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. `PeerDown` holds the table's lock for
-  10,000 prefixes at a time, then lets the lock go for as long again. The
-  handler holds back the down peer's later messages until that peer's `PeerDown`
-  returns. Then the handler 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. When `Stop` runs past its
-  deadline, it logs how many held-back messages were not applied (closes #216)
-- 2026-11-02: RouteWatch pauses an RIR's or NIC.br's WHOIS server for an hour
-  after the server refuses a query. A refused or reset connection is a refusal.
-  So is an answer with an `%ERROR:201:` or `%ERROR:202:` line. Lookups at the
-  other servers go on. An AS whose RIR is paused is looked up after the pause.
-  IANA is never paused, because a pause of IANA would stop every lookup. A
-  refusal by IANA is an ordinary failure. NIC.br is also paused for an hour
-  after it says RouteWatch has passed NIC.br's query limit. Before, the fetcher
-  asked a server again within seconds of a refusal (closes #218)
-- 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
-  15 seconds for each step it expects. The tests of the handlers' limits give
-  the handlers small ones, so the package no longer runs past its 90-second
-  limit (closes #213, #215)
-- 2026-10-27: the server works out the home page's cards every minute until it
-  stops. It used to work them out only once, at startup: the end of the context
-  that `Start` is given stopped the refreshes (closes #211)
-- 2026-10-27: README "Rationale" says why RouteWatch exists: to answer our other
-  services' questions about IP addresses, from its own routing table and the
-  WHOIS details it keeps. It still says what RIS Live does not give (closes
-  #204)
 - 2026-10-27: every JSON answer carries `@meta`. A panic on a JSON route answers
   500 with the JSON error body. `/status.json` and the health check answer 405
   with the JSON error body to any method they do not take. The 500 and the 405

internal/routetable/overview.go

@@ -10,6 +10,10 @@ 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,11 +27,6 @@ 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
 
@@ -206,13 +201,8 @@ func (t *Table) Withdraw(prefix netip.Prefix, peer string) {
 }
 
 // PeerDown applies a peer's session going down: the peer leaves every peer
-// set, and every route of the peer is dropped. When the peer carries any
+// set, and every route of it is dropped. When the peer carries any prefix, it
-// prefix, PeerDown visits every prefix in the table, in batches of walkBatch.
+// visits every prefix in the table.
-// After each batch, it lets the table's lock go for as long as it held the
-// lock for the batch. During a walk over many batches, reads and updates can
-// then take the lock about half of the time. An announcement by the peer
-// applied during the walk may be dropped too. So the caller applies the peer's
-// later updates only once PeerDown has returned.
 func (t *Table) PeerDown(peer string) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
@@ -222,9 +212,6 @@ func (t *Table) PeerDown(peer string) {
 		return
 	}
 
-	visited := 0
-	batchStart := time.Now()
-
 	for _, f := range []*family{&t.ipv4, &t.ipv6} {
 		for _, prefixes := range f.byLength {
 			for prefix, origins := range prefixes {
@@ -232,19 +219,6 @@ 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 {
-					held := time.Since(batchStart)
-
-					t.mu.Unlock()
-					time.Sleep(held)
-					t.mu.Lock()
-
-					batchStart = time.Now()
-				}
 			}
 		}
 	}

internal/routetable/routetable_internal_test.go

@@ -1,69 +0,0 @@
-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

@@ -509,6 +509,8 @@ func announcementFeed(prefixes []string) string {
 func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
 	t.Parallel()
 
+	const maxWait = 10 * time.Second
+
 	prefixes := []string{"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"}
 	feed := announcementFeed(prefixes)
 
@@ -554,9 +556,9 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
 	for range cap(linesSeen) {
 		select {
 		case <-linesSeen:
-		case <-time.After(stepTimeout):
+		case <-time.After(maxWait):
 			t.Fatalf("the streamer did not read %d lines within %s",
-				cap(linesSeen), stepTimeout)
+				cap(linesSeen), maxWait)
 		}
 	}
 
@@ -569,8 +571,8 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
 
 	select {
 	case <-shutDown:
-	case <-time.After(stepTimeout):
+	case <-time.After(maxWait):
-		t.Fatalf("Shutdown did not return within %s", stepTimeout)
+		t.Fatalf("Shutdown did not return within %s", maxWait)
 	}
 
 	store.mu.Lock()
@@ -607,16 +609,21 @@ func freePort(t *testing.T) int {
 	return addr.Port
 }
 
-// TestShutdownDoesNotWaitForABlockedWrite starts the HTTP server and has the
+// TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked starts the HTTP server
-// AS handler start a write that the store holds until the test ends, past
+// and has the AS handler start a write that the store holds until the test
-// Shutdown's deadline. Shutdown must return without waiting for the write, with
+// ends, past Shutdown's deadline. Shutdown must return within its deadline
-// the HTTP server stopped, and must neither checkpoint nor close the database:
+// plus a small margin, with the HTTP server stopped, and must neither
-// the real one would wait for the write.
+// checkpoint nor close the database: the real one would wait for the write.
 // Before the fix Shutdown waited for the write.
-func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) {
+func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
 	t.Parallel()
 
-	const deadline = 100 * time.Millisecond
+	const (
+		deadline = 100 * time.Millisecond
+		// margin is how long Shutdown may take after its deadline, to stop
+		// the HTTP server.
+		margin = 2 * time.Second
+	)
 
 	blocking := newBlockingStore()
 	defer blocking.releaseWrites()
@@ -644,8 +651,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()
@@ -659,9 +667,9 @@ func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) {
 
 	select {
 	case <-shutDown:
-	case <-time.After(deadline + stepTimeout):
+	case <-time.After(deadline + margin):
 		t.Fatalf("Shutdown did not return within %s of its deadline while "+
-			"a write was blocked", stepTimeout)
+			"a write was blocked", margin)
 	}
 
 	var dialer net.Dialer
@@ -736,6 +744,8 @@ func (s *closeOrderStore) record(call string) {
 func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
 	t.Parallel()
 
+	const deadline = 10 * time.Second
+
 	store := &closeOrderStore{Store: newMockStore()}
 	log := logger.New()
 	cfg := &config.Config{}
@@ -753,7 +763,7 @@ func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
 		Path: ristypes.ASPath{Sequence: []int{testASNA}},
 	})
 
-	ctx, cancel := context.WithTimeout(t.Context(), stepTimeout)
+	ctx, cancel := context.WithTimeout(t.Context(), deadline)
 	defer cancel()
 
 	rw.Shutdown(ctx)

internal/routewatch/ashandler.go

@@ -18,34 +18,39 @@ 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
+	// 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.
+	// waits for a database write only while batch is at asnBatchMaxSize.
-	mu sync.Mutex
+	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 int
-	// writing is how many ASes the flush goroutine took from batch and has
+	// batchTaken is signalled when the flush goroutine takes batch; its L is
-	// not finished writing.
+	// &mu.
-	writing     int
+	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 +58,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 +81,12 @@ func newASHandler(
 	h := &ASHandler{
 		db:          db,
 		logger:      logger,
-		batch:       make(map[int]database.SeenTimes),
+		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 +110,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) >= asnBatchMaxSize {
+		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 asnBatchMaxSize.
 	if len(h.batch) >= asnBatchSize {
 		select {
 		case h.batchFullCh <- struct{}{}:
@@ -125,11 +160,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 +178,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 +195,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 +209,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()}
-		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()}
-		db: store, logger: logger.New(), batch: make(map[int]database.SeenTimes),
-	}
 
 	msg := peerAnnouncement(testPeerIP, testV4Prefix)
 	msg.Path = decodePath(t, testASNA, 0, testASNB)

internal/routewatch/asnfetcher.go

@@ -287,13 +287,6 @@ func (f *ASNFetcher) fetchNextStale(ctx context.Context) {
 // so the AS stays stale. An AS IANA refers to no RIR is stored with no
 // WHOIS data, so it is not looked up again until it is stale; that counts
 // as neither a success nor a failure.
-//
-// The WHOIS client pauses an RIR's or NIC.br's server for whois.PauseTime
-// after the server refuses a query. Lookups at the other servers go on. IANA
-// is never paused, so a refusal by IANA is an ordinary failure. A lookup
-// whose RIR is paused fails with whois.ErrPaused. Such a lookup leaves the
-// AS stale, and counts as neither a success nor a failure. The lookups that
-// QueueImmediate asks for also run through fetchAndUpdate.
 func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool {
 	ctx, cancel := context.WithTimeout(ctx, time.Minute)
 	defer cancel()
@@ -314,13 +307,6 @@ func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool {
 		return false
 	}
 
-	if errors.Is(err, whois.ErrPaused) {
-		f.logger.Info("Skipped WHOIS lookup, the RIR's server is paused",
-			"asn", asn, "error", err)
-
-		return false
-	}
-
 	if err != nil {
 		f.logger.Error("WHOIS lookup failed", "asn", asn, "error", err)
 		f.recordFailure()

internal/routewatch/asnfetcher_internal_test.go

@@ -13,7 +13,6 @@ import (
 
 	"git.eeqj.de/sneak/routewatch/internal/database"
 	"git.eeqj.de/sneak/routewatch/internal/logger"
-	"git.eeqj.de/sneak/routewatch/internal/whois"
 )
 
 // The WHOIS servers a lookup asks: IANA first, then the RIR IANA refers it
@@ -132,109 +131,6 @@ func TestFetchAndUpdateKeepsTheRowWhenTheLookupFails(t *testing.T) {
 	}
 }
 
-// TestFetchAndUpdateWaitsOutAPause looks up an AS at RIPE while RIPE refuses
-// the query. The test sets the WHOIS client's clock. The next lookup comes a
-// second before whois.PauseTime has passed. It sends RIPE nothing and leaves
-// the AS never looked up. It counts as neither a success nor a failure. A
-// lookup at whois.PauseTime asks RIPE again.
-func TestFetchAndUpdateWaitsOutAPause(t *testing.T) {
-	t.Parallel()
-
-	f, db := newFetcherWithWHOISAnswers(t, map[string]string{
-		ianaWHOISServer: ianaRefersToRIPE,
-		ripeWHOISServer: ripeAccessDenied,
-	})
-
-	ripeAsks := 0
-	dial := f.whoisClient.DialContext
-	f.whoisClient.DialContext = func(
-		ctx context.Context, network, address string,
-	) (net.Conn, error) {
-		if address == ripeWHOISServer {
-			ripeAsks++
-		}
-
-		return dial(ctx, network, address)
-	}
-
-	start := time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
-	now := start
-	f.whoisClient.Now = func() time.Time { return now }
-
-	f.fetchAndUpdate(t.Context(), testASNA)
-
-	now = start.Add(whois.PauseTime - time.Second)
-
-	f.fetchAndUpdate(t.Context(), testASNA)
-
-	if ripeAsks != 1 {
-		t.Errorf("RIPE was asked %d times during its pause, want 1", ripeAsks)
-	}
-
-	stats, err := db.GetWHOISStats(t.Context(), whoisStaleThreshold)
-	if err != nil {
-		t.Fatalf("GetWHOISStats: %v", err)
-	}
-
-	if stats.NeverFetched != 1 {
-		t.Errorf("never looked up = %d, want 1", stats.NeverFetched)
-	}
-
-	got := f.GetStats()
-	if got.ConsecutiveFails != 1 || got.ErrorsLastHour != 1 ||
-		got.SuccessesLastHour != 0 {
-		t.Errorf("%d failures in a row, %d errors, %d successes; want 1, 1, 0",
-			got.ConsecutiveFails, got.ErrorsLastHour, got.SuccessesLastHour)
-	}
-
-	now = start.Add(whois.PauseTime)
-
-	f.fetchAndUpdate(t.Context(), testASNA)
-
-	if ripeAsks != 2 {
-		t.Errorf("RIPE was asked %d times after its pause, want 2", ripeAsks)
-	}
-}
-
-// TestFetchAndUpdateKeepsAskingIANA looks up an AS while IANA's server
-// refuses the connection. The test sets the WHOIS client's clock. The next
-// lookup comes a second before whois.PauseTime has passed. IANA is never
-// paused, so that lookup asks IANA again. Each refusal counts as a failure.
-func TestFetchAndUpdateKeepsAskingIANA(t *testing.T) {
-	t.Parallel()
-
-	// With no answers, every server refuses the connection.
-	f, _ := newFetcherWithWHOISAnswers(t, map[string]string{})
-
-	ianaAsks := 0
-	dial := f.whoisClient.DialContext
-	f.whoisClient.DialContext = func(
-		ctx context.Context, network, address string,
-	) (net.Conn, error) {
-		if address == ianaWHOISServer {
-			ianaAsks++
-		}
-
-		return dial(ctx, network, address)
-	}
-
-	start := time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
-	now := start
-	f.whoisClient.Now = func() time.Time { return now }
-
-	f.fetchAndUpdate(t.Context(), testASNA)
-
-	now = start.Add(whois.PauseTime - time.Second)
-
-	f.fetchAndUpdate(t.Context(), testASNA)
-
-	got := f.GetStats()
-	if ianaAsks != 2 || got.ConsecutiveFails != 2 {
-		t.Errorf("IANA was asked %d times, with %d failures in a row; "+
-			"want 2 and 2", ianaAsks, got.ConsecutiveFails)
-	}
-}
-
 // TestRecordStepsTheInterval has the fetcher fail from its base interval of
 // 15 seconds, then succeed. Each failure makes the interval 1.5 times as
 // long, up to one minute, and each success halves it, down to one second.
@@ -328,7 +224,7 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
 
 	select {
 	case <-lookupStarted:
-	case <-time.After(stepTimeout):
+	case <-time.After(handleMessageTimeout):
 		t.Fatal("the fetcher did not start the lookup")
 	}
 
@@ -344,22 +240,22 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
 
 	select {
 	case <-f.done:
-	case <-time.After(stepTimeout):
+	case <-time.After(handleMessageTimeout):
 		t.Fatal("the fetcher goroutine did not return after the lookup ended")
 	}
 }
 
 // TestRecordKeepsOnlyTheStatsWindow records a success and a failure every
-// second, by a clock the test sets, for three times statsWindow, and never
+// second, by a clock the test sets, far longer than statsWindow, and never
 // asks for the stats. The fetcher must keep only the times within the last
 // statsWindow.
 func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) {
 	t.Parallel()
 
 	const (
+		records = 100_000
 		step    = time.Second
 		want    = int(statsWindow / step)
-		records = 3 * want
 	)
 
 	clock := time.Now()

internal/routewatch/batchwrite_internal_test.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"log/slog"
 	"slices"
 	"sync"
@@ -20,15 +19,9 @@ import (
 )
 
 const (
-	// stepTimeout is how long a test waits for a step it expects, such as
+	// handleMessageTimeout is how long a test waits for HandleMessage to
-	// HandleMessage returning, a write starting or Shutdown returning, before
+	// return, or for a write to start, before it fails.
-	// it fails. Only a step that never comes should reach it. On a loaded
+	handleMessageTimeout = 2 * time.Second
-	// host, under the race detector, HandleMessage takes seconds to apply a
-	// full prefix batch. The helpers below bound every wait with it, Stop
-	// included. A test that hangs in one of them waits out at most two of
-	// these bounds, its step's and Stop's, and then fails with its own
-	// message. That leaves time before go test's 90-second timeout.
-	stepTimeout = 15 * time.Second
 
 	// stillWaitingAfter is how long a test watches a HandleMessage that must
 	// wait. One that does not wait returns well within it.
@@ -178,14 +171,12 @@ 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 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}))}
 }
@@ -207,7 +198,7 @@ func lastLogged(t *testing.T, logs *bytes.Buffer) stopLog {
 	return logged
 }
 
-// returnsInTime reports whether f returns within stepTimeout.
+// returnsInTime reports whether f returns within handleMessageTimeout.
 func returnsInTime(f func()) bool {
 	done := make(chan struct{})
 
@@ -219,22 +210,11 @@ func returnsInTime(f func()) bool {
 	select {
 	case <-done:
 		return true
-	case <-time.After(stepTimeout):
+	case <-time.After(handleMessageTimeout):
 		return false
 	}
 }
 
-// stopInTime stops h, with no deadline, and fails the test if Stop does not
-// return within stepTimeout. It only marks the test failed, so it can be
-// deferred.
-func stopInTime(t *testing.T, h interface{ Stop(ctx context.Context) }) {
-	t.Helper()
-
-	if !returnsInTime(func() { h.Stop(t.Context()) }) {
-		t.Errorf("Stop did not return within %s", stepTimeout)
-	}
-}
-
 // startBlockedWrite hands h the messages in fill and waits until h starts to
 // write them to store. The prefix, AS and peer handlers start when fill fills
 // their batch, and the peering handler on its timer. store holds the write
@@ -260,7 +240,7 @@ func startBlockedWrite(
 
 	select {
 	case <-store.writeStarted:
-	case <-time.After(stepTimeout):
+	case <-time.After(handleMessageTimeout):
 		t.Fatal("the handler did not start to write the messages")
 	}
 }
@@ -281,7 +261,7 @@ func handleWhileAWriteIsBlocked(
 ) {
 	t.Helper()
 
-	defer stopInTime(t, h)
+	defer h.Stop(t.Context())
 	defer store.releaseWrites()
 
 	startBlockedWrite(t, store, h, fill)
@@ -295,9 +275,7 @@ func handleWhileAWriteIsBlocked(
 // while store holds the write, hands h the messages in toMax, which bring its
 // next batch to its maximum size, and the message extra. HandleMessage for
 // extra must wait until the test lets the write finish, and return after. Then
-// it stops h, which writes what it holds. The tests lower h's maximum batch
+// it stops h, which writes what it holds.
-// size to one batch, so toMax is one batch: the real maximum of ten batches is
-// slow to reach under the race detector.
 func handleAtMaxBatchSize(
 	t *testing.T,
 	store *blockingStore,
@@ -310,18 +288,13 @@ func handleAtMaxBatchSize(
 ) {
 	t.Helper()
 
-	defer stopInTime(t, h)
+	defer h.Stop(t.Context())
 	defer store.releaseWrites()
 
 	startBlockedWrite(t, store, h, fill)
 
-	if !returnsInTime(func() {
+	for _, msg := range toMax {
-		for _, msg := range toMax {
+		h.HandleMessage(msg)
-			h.HandleMessage(msg)
-		}
-	}) {
-		t.Fatal("HandleMessage did not return while the batch grew to its " +
-			"maximum size")
 	}
 
 	returned := make(chan struct{})
@@ -342,7 +315,7 @@ func handleAtMaxBatchSize(
 
 	select {
 	case <-returned:
-	case <-time.After(stepTimeout):
+	case <-time.After(handleMessageTimeout):
 		t.Fatal("HandleMessage did not return after the database write finished")
 	}
 }
@@ -372,12 +345,8 @@ func stopWhileAWriteIsBlocked(
 
 	startBlockedWrite(t, store, h, fill)
 
-	if !returnsInTime(func() {
+	for _, msg := range extra {
-		for _, msg := range extra {
+		h.HandleMessage(msg)
-			h.HandleMessage(msg)
-		}
-	}) {
-		t.Fatal("HandleMessage did not return while a database write was blocked")
 	}
 
 	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
@@ -392,7 +361,7 @@ func stopWhileAWriteIsBlocked(
 
 	select {
 	case <-done:
-	case <-time.After(stepTimeout):
+	case <-time.After(handleMessageTimeout):
 		t.Fatal("the flush goroutine did not return after the write finished")
 	}
 }
@@ -411,8 +380,7 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
 
 	store := newBlockingStore()
 	h := newPrefixHandler(store, newErrorLogger(&logs),
-		routetable.New(routetable.Params{}), longerThanAnyTest,
+		routetable.New(routetable.Params{}), longerThanAnyTest)
-		maxTableChangesPerWrite)
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
 		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
@@ -498,20 +466,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 +489,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()
@@ -630,7 +597,7 @@ func TestPeeringHandlerStopCountsTheWriteInProgress(t *testing.T) {
 
 	select {
 	case <-h.done:
-	case <-time.After(stepTimeout):
+	case <-time.After(handleMessageTimeout):
 		t.Fatal("the processing goroutine did not return after the write " +
 			"finished")
 	}
@@ -651,7 +618,7 @@ func TestPeeringHandlerStartsNoWriteAfterTheStopDeadline(t *testing.T) {
 	const processInterval = 10 * time.Millisecond
 
 	store := newBlockingStore()
-	h := newPeeringHandler(store, logger.New(), processInterval, maxTrackedPaths)
+	h := newPeeringHandler(store, logger.New(), processInterval)
 	stopWhileAWriteIsBlocked(t, store, h, h.done,
 		[]*ristypes.RISMessage{
 			{Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}},
@@ -707,8 +674,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()
@@ -746,11 +714,9 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	const extraPrefix = "198.51.100.0/24"
 
 	store := newBlockingStore()
-	h := newPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}),
+	h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))
-		longerThanAnyTest, maxTableChangesPerWrite)
-	h.batchMaxSize = prefixBatchSize
 	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
-	toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
+	toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchMaxSize)
 	handleAtMaxBatchSize(t, store, h,
 		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
 		[]*ristypes.RISMessage{prefixAnnouncement(toMax)},
@@ -765,48 +731,31 @@ 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)
+	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
+	toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchMaxSize)}
+	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))
 	}
 }
 
 // TestPeerHandlerWaitsAtTheMaximumBatchSize must not have its flush timer fire.
 // A timer flush during the fill would take only part of it. The batch would
 // then reach its maximum size before the test's last message, and the test
-// would fail.
+// would hang.
 func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	t.Parallel()
 
@@ -814,11 +763,10 @@ func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 
 	store := newBlockingStore()
 	h := newPeerHandler(store, logger.New(), longerThanAnyTest)
-	h.batchMaxSize = peerBatchSize
 	message := []*ristypes.RISMessage{{Peer: testPeerIP}}
 	handleAtMaxBatchSize(t, store, h,
 		slices.Repeat(message, peerBatchSize),
-		slices.Repeat(message, peerBatchSize),
+		slices.Repeat(message, peerBatchMaxSize),
 		&ristypes.RISMessage{Peer: otherPeerIP})
 
 	store.mu.Lock()
@@ -848,9 +796,8 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			batchSize: prefixBatchSize,
 			handle: func(n int) chan struct{} {
 				h := &PrefixHandler{
-					batchMaxSize: prefixBatchMaxSize,
+					batchFullCh: make(chan struct{}, 1),
-					batchFullCh:  make(chan struct{}, 1),
+					table:       routetable.New(routetable.Params{}),
-					table:        routetable.New(routetable.Params{}),
 				}
 				h.HandleMessage(prefixAnnouncement(
 					slices.Repeat([]string{testV4Prefix}, n)))
@@ -862,12 +809,11 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			name:      "AS",
 			batchSize: asnBatchSize,
 			handle: func(n int) chan struct{} {
-				h := &ASHandler{
+				h := &ASHandler{batchFullCh: make(chan struct{}, 1)}
-					batch:       make(map[int]database.SeenTimes),
+				// The origin AS and each AS in the path are one update each.
-					batchFullCh: make(chan struct{}, 1),
+				h.HandleMessage(&ristypes.RISMessage{
-				}
+					Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)},
-				// Each of the path's n ASes is one entry of the batch.
+				})
-				h.HandleMessage(&ristypes.RISMessage{Path: pathOfASes(n)})
 
 				return h.batchFullCh
 			},
@@ -876,10 +822,7 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			name:      "peer",
 			batchSize: peerBatchSize,
 			handle: func(n int) chan struct{} {
-				h := &PeerHandler{
+				h := &PeerHandler{batchFullCh: make(chan struct{}, 1)}
-					batchMaxSize: peerBatchMaxSize,
-					batchFullCh:  make(chan struct{}, 1),
-				}
 				for range n {
 					h.HandleMessage(&ristypes.RISMessage{Peer: testPeerIP})
 				}

internal/routewatch/dbmaintainer_internal_test.go

@@ -97,7 +97,7 @@ func TestDBMaintainerStopReturnsByItsDeadline(t *testing.T) {
 
 	select {
 	case <-store.started:
-	case <-time.After(stepTimeout):
+	case <-time.After(handleMessageTimeout):
 		t.Fatal("the maintainer did not start the checkpoint")
 	}
 
@@ -113,7 +113,7 @@ func TestDBMaintainerStopReturnsByItsDeadline(t *testing.T) {
 
 	select {
 	case <-m.done:
-	case <-time.After(stepTimeout):
+	case <-time.After(handleMessageTimeout):
 		t.Fatal("the maintenance goroutine did not return after the " +
 			"checkpoint ended")
 	}

internal/routewatch/peerhandler.go

@@ -43,14 +43,10 @@ type PeerHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// batchMaxSize is peerBatchMaxSize. Only HandleMessage reads it, so a test
-	// may lower it before it hands the handler a message.
-	batchMaxSize int
-
 	// Batching. mu guards peerBatch 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 peerBatch is at
-	// batchMaxSize.
+	// peerBatchMaxSize.
 	mu        sync.Mutex
 	peerBatch []peerUpdate
 	// writing is how many updates the flush goroutine took from peerBatch
@@ -88,13 +84,12 @@ func newPeerHandler(
 	db database.Store, logger *logger.Logger, batchTimeout time.Duration,
 ) *PeerHandler {
 	h := &PeerHandler{
-		db:           db,
+		db:          db,
-		logger:       logger,
+		logger:      logger,
-		batchMaxSize: peerBatchMaxSize,
+		peerBatch:   make([]peerUpdate, 0, peerBatchSize),
-		peerBatch:    make([]peerUpdate, 0, peerBatchSize),
+		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
 
@@ -140,7 +135,7 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// 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.peerBatch) >= h.batchMaxSize {
+	for len(h.peerBatch) >= peerBatchMaxSize {
 		h.batchTaken.Wait()
 	}
 
@@ -154,7 +149,7 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 
 	// 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
-	// or the batch reaches batchMaxSize.
+	// or the batch reaches peerBatchMaxSize.
 	if len(h.peerBatch) >= peerBatchSize {
 		select {
 		case h.batchFullCh <- struct{}{}:

internal/routewatch/peerhandler_internal_test.go

@@ -35,7 +35,7 @@ func TestPeerHandlerWritesTheEarliestAndLatestTimesOfANewPeer(t *testing.T) {
 
 	// Built without NewPeerHandler's flush goroutine, so the test flushes
 	// itself.
-	h := &PeerHandler{db: store, logger: logger.New(), batchMaxSize: peerBatchMaxSize}
+	h := &PeerHandler{db: store, logger: logger.New()}
 
 	first := peerAnnouncement(testPeerIP, testV4Prefix)
 	second := peerAnnouncement(testPeerIP, testV4Prefix)

internal/routewatch/peeringhandler.go

@@ -42,10 +42,6 @@ type PeeringHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// maxPaths is the most paths asPaths holds: maxTrackedPaths, except in
-	// tests that give a smaller one.
-	maxPaths int
-
 	// In-memory AS path tracking: each path, JSON-encoded, with the earliest
 	// and the latest time an update carried it.
 	mu           sync.Mutex
@@ -66,24 +62,21 @@ type PeeringHandler struct {
 // periodically processes accumulated AS paths into peering records. The
 // handler begins processing immediately upon creation.
 func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler {
-	return newPeeringHandler(db, logger, peeringProcessInterval, maxTrackedPaths)
+	return newPeeringHandler(db, logger, peeringProcessInterval)
 }
 
 // newPeeringHandler is NewPeeringHandler with the processing timer's interval,
-// processInterval, in place of peeringProcessInterval, and the most paths it
+// processInterval, in place of peeringProcessInterval. A test gives a short
-// holds, maxPaths, in place of maxTrackedPaths. A test gives a short interval to
+// one to have the timer start a write.
-// have the timer start a write, and a small maxPaths to fill the map quickly.
 func newPeeringHandler(
 	db database.Store, logger *logger.Logger, processInterval time.Duration,
-	maxPaths int,
 ) *PeeringHandler {
 	h := &PeeringHandler{
-		db:       db,
+		db:      db,
-		logger:   logger,
+		logger:  logger,
-		asPaths:  make(map[string]database.SeenTimes),
+		asPaths: make(map[string]database.SeenTimes),
-		maxPaths: maxPaths,
+		stopCh:  make(chan context.Context, 1),
-		stopCh:   make(chan context.Context, 1),
+		done:    make(chan struct{}),
-		done:     make(chan struct{}),
 	}
 
 	// Start the periodic processing goroutine
@@ -138,7 +131,7 @@ func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) {
 	defer h.mu.Unlock()
 
 	seen, exists := h.asPaths[key]
-	if !exists && len(h.asPaths) >= h.maxPaths {
+	if !exists && len(h.asPaths) >= maxTrackedPaths {
 		// Map is full; drop this new path and count it.
 		h.droppedPaths++
 

internal/routewatch/peeringhandler_internal_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"maps"
 	"slices"
+	"strconv"
 	"sync"
 	"testing"
 	"time"
@@ -19,10 +20,6 @@ const (
 	testASNB = 64501
 	testASNC = 64502
 	testASND = 64503
-
-	// testMaxTrackedPaths is a limit of the path map that a test fills
-	// quickly, in place of maxTrackedPaths.
-	testMaxTrackedPaths = 10
 )
 
 // recordingStore wraps mockStore to count every RecordPeering call, so a
@@ -55,10 +52,9 @@ func (r *recordingStore) callCount() int {
 // processing goroutine, so tests drive processing explicitly.
 func newTestHandler(db database.Store) *PeeringHandler {
 	return &PeeringHandler{
-		db:       db,
+		db:      db,
-		logger:   logger.New(),
+		logger:  logger.New(),
-		asPaths:  make(map[string]database.SeenTimes),
+		asPaths: make(map[string]database.SeenTimes),
-		maxPaths: maxTrackedPaths,
 	}
 }
 
@@ -248,28 +244,25 @@ func TestPeeringHandlerStopWritesTheRemainingPaths(t *testing.T) {
 
 // TestPeeringHandlerCapDropsAndCounts verifies that a full map drops new paths
 // and counts them, while a path already tracked is refreshed rather than
-// dropped. The handler's limit is small, and its timer does not fire.
+// dropped.
 func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 	t.Parallel()
 
 	store := &recordingStore{mockStore: newMockStore()}
-	h := newPeeringHandler(store, logger.New(), longerThanAnyTest,
+	h := newTestHandler(store)
-		testMaxTrackedPaths)
-
-	defer stopInTime(t, h)
 
 	now := time.Now().UTC()
 
-	// Fill the map to exactly its limit, including one real path key so the
+	// Fill the map to exactly maxTrackedPaths, including one real path key so
-	// "already tracked" branch can be exercised. The other paths are of one
+	// the "already tracked" branch can be exercised. The filler keys are never
-	// AS, so they carry no peering.
+	// processed in this test, so their contents do not matter.
 	existing := pathKey(t, testASNA, testASNB)
 
 	h.mu.Lock()
 	h.asPaths[existing] = database.SeenTimes{First: now, Last: now}
 
-	for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ {
+	for i := 0; len(h.asPaths) < maxTrackedPaths; i++ {
-		h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now}
+		h.asPaths[strconv.Itoa(i)] = database.SeenTimes{First: now, Last: now}
 	}
 	h.mu.Unlock()
 
@@ -277,9 +270,8 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 	handle(h, now.Add(time.Second), testASNA, testASNC)
 
 	tracked, dropped := snapshot(h)
-	if tracked != testMaxTrackedPaths {
+	if tracked != maxTrackedPaths {
-		t.Fatalf("want map size %d after drop, got %d", testMaxTrackedPaths,
+		t.Fatalf("want map size %d after drop, got %d", maxTrackedPaths, tracked)
-			tracked)
 	}
 
 	if dropped != 1 {
@@ -291,9 +283,8 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 	handle(h, refreshed, testASNA, testASNB)
 
 	tracked, dropped = snapshot(h)
-	if tracked != testMaxTrackedPaths {
+	if tracked != maxTrackedPaths {
-		t.Fatalf("want map size %d after refresh, got %d", testMaxTrackedPaths,
+		t.Fatalf("want map size %d after refresh, got %d", maxTrackedPaths, tracked)
-			tracked)
 	}
 
 	if dropped != 1 {
@@ -312,21 +303,17 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
 
 // TestPeeringHandlerRunResetsTheDroppedCount fills the map, has the handler
 // drop one path, and runs the processing. The count of dropped paths is the
-// run's own, so it is 0 again after the run. The handler's limit is small, and
+// run's own, so it is 0 again after the run.
-// its timer does not fire.
 func TestPeeringHandlerRunResetsTheDroppedCount(t *testing.T) {
 	t.Parallel()
 
-	h := newPeeringHandler(newMockStore(), logger.New(), longerThanAnyTest,
+	h := newTestHandler(newMockStore())
-		testMaxTrackedPaths)
-
-	defer stopInTime(t, h)
 
 	now := time.Now().UTC()
 
 	// Paths of one AS carry no peering, so the run records nothing.
 	h.mu.Lock()
-	for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ {
+	for asn := 1; len(h.asPaths) < maxTrackedPaths; asn++ {
 		h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now}
 	}
 	h.mu.Unlock()

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
@@ -58,15 +52,6 @@ const (
 // does an announcement whose AS path is empty or holds AS 0 anywhere. A peer
 // whose BGP session goes down leaves every prefix.
 //
-// The table's PeerDown applies a session going down. It visits every prefix,
-// which takes seconds on a full table. So the handler runs PeerDown on
-// a separate goroutine, one peer at a time, and goes on taking messages
-// meanwhile. PeerDown lets the table's lock go about half of the time, so the
-// handler applies other peers' messages while PeerDown runs. The handler holds
-// back the down peer's later messages until PeerDown returns. Then the handler
-// applies them in order, one at a time, and takes new messages in between. It
-// holds back at most maxHeldBackMessages messages.
-//
 // 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 {
@@ -75,17 +60,10 @@ type PrefixHandler struct {
 	metrics *metrics.Tracker
 	table   *routetable.Table
 
-	// batchMaxSize is prefixBatchMaxSize. Only HandleMessage reads it, so a
+	// Batching. mu guards batch, writing and writingTableChanges only: the
-	// test may lower it before it hands the handler a message.
-	batchMaxSize int
-	// tableChangesPerWrite is maxTableChangesPerWrite, except in tests that
-	// give a smaller one.
-	tableChangesPerWrite int
-
-	// 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.
+	// prefixBatchMaxSize.
 	mu    sync.Mutex
 	batch []seenPrefix
 	// writing is how many updates the flush goroutine took and has not
@@ -102,21 +80,6 @@ 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
@@ -156,34 +119,26 @@ func (u prefixUpdate) announcesRoute() bool {
 func NewPrefixHandler(
 	db database.Store, logger *logger.Logger, table *routetable.Table,
 ) *PrefixHandler {
-	return newPrefixHandler(db, logger, table, prefixBatchTimeout,
+	return newPrefixHandler(db, logger, table, prefixBatchTimeout)
-		maxTableChangesPerWrite)
 }
 
 // newPrefixHandler is NewPrefixHandler with the flush timer's interval,
-// batchTimeout, in place of prefixBatchTimeout, and tableChangesPerWrite in
+// batchTimeout, in place of prefixBatchTimeout. A test that the timer must not
-// place of maxTableChangesPerWrite. A test that the timer must not disturb
+// disturb gives an interval longer than the test runs.
-// gives an interval longer than the test runs. A test of writes split over
-// several flushes gives a small tableChangesPerWrite, so that it exceeds it
-// with few changes.
 func newPrefixHandler(
 	db database.Store, logger *logger.Logger, table *routetable.Table,
-	batchTimeout time.Duration, tableChangesPerWrite int,
+	batchTimeout time.Duration,
 ) *PrefixHandler {
 	h := &PrefixHandler{
-		db:                   db,
+		db:          db,
-		logger:               logger,
+		logger:      logger,
-		table:                table,
+		table:       table,
-		batch:                make([]seenPrefix, 0, prefixBatchSize),
+		batch:       make([]seenPrefix, 0, prefixBatchSize),
-		batchMaxSize:         prefixBatchMaxSize,
+		batchFullCh: make(chan struct{}, 1),
-		tableChangesPerWrite: tableChangesPerWrite,
+		stopCh:      make(chan context.Context, 1),
-		batchFullCh:          make(chan struct{}, 1),
+		done:        make(chan struct{}),
-		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)
@@ -211,91 +166,11 @@ func (h *PrefixHandler) QueueCapacity() int {
 	return prefixHandlerQueueSize
 }
 
-// HandleMessage processes a message to track prefix information. A message
+// HandleMessage processes a message to track prefix information
-// 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.startPeerDown(msg.Peer)
+			h.table.PeerDown(msg.Peer)
 		}
 
 		return
@@ -310,10 +185,13 @@ func (h *PrefixHandler) applyMessage(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.
-	for len(h.batch) >= h.batchMaxSize {
+	for len(h.batch) >= prefixBatchMaxSize {
 		h.batchTaken.Wait()
 	}
 
@@ -361,7 +239,7 @@ func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
 
 	// 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
-	// or the batch reaches batchMaxSize.
+	// or the batch reaches prefixBatchMaxSize.
 	if len(h.batch) >= prefixBatchSize {
 		select {
 		case h.batchFullCh <- struct{}{}:
@@ -370,58 +248,39 @@ func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// startPeerDown starts peerDown for peer on a new goroutine. The caller holds
+// Stop stops the flush goroutine, which makes one last flush before it returns,
-// mu. The peer's messages are held back until peerDown has applied them.
+// and waits for it until ctx ends. That flush writes at most
-// Messages held back for an earlier PeerDown of the peer, and not yet applied,
+// maxTableChangesPerWrite of the routing table's changes, and the prefixes of
-// stay held back.
+// the updates the handler holds. Once ctx has ended the flush goroutine skips
-func (h *PrefixHandler) startPeerDown(peer string) {
+// that last flush, and a write still running is lost when the process exits. A
-	_, goingDown := h.heldBack[peer]
+// failed write of the table's changes is not retried either. Stop logs how many
-	if !goingDown {
+// updates and how many of the table's changes are not written.
-		h.heldBack[peer] = []*ristypes.RISMessage{}
+func (h *PrefixHandler) Stop(ctx context.Context) {
+	h.stopCh <- ctx
+
+	select {
+	case <-h.done:
+	case <-ctx.Done():
 	}
 
-	h.peerDowns.Add(1)
+	h.mu.Lock()
+	unwritten := h.writing + len(h.batch)
+	tableChanges := h.writingTableChanges + h.table.ChangeCount()
+	h.mu.Unlock()
 
-	go h.peerDown(peer)
+	if unwritten == 0 && tableChanges == 0 {
-}
+		return
-
-// peerDown runs the table's PeerDown for peer, once no other PeerDown is
-// running. Then peerDown applies the messages held back from the peer, in
-// order. It takes mu for one message at a time, so HandleMessage takes new
-// messages in between. The peer's new messages join the end of the held-back
-// ones. A held-back message that says the peer's session went down again
-// starts a new peerDown, which applies the rest.
-func (h *PrefixHandler) peerDown(peer string) {
-	defer h.peerDowns.Done()
-
-	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
@@ -519,13 +378,14 @@ func (h *PrefixHandler) applyToTable(update prefixUpdate) {
 	})
 }
 
-// writeTableChanges writes at most tableChangesPerWrite changes of the routing
+// writeTableChanges writes at most maxTableChangesPerWrite changes of the
-// table in memory, and every peer the table numbered since the last write.
+// routing table in memory, and every peer the table numbered since the last
+// write.
 // When the write fails, they go back to the table, so that a later write writes
 // them again, each pair with its content then.
 func (h *PrefixHandler) writeTableChanges() {
 	h.mu.Lock()
-	changes, newPeers := h.table.TakeChanges(h.tableChangesPerWrite)
+	changes, newPeers := h.table.TakeChanges(maxTableChangesPerWrite)
 	h.writingTableChanges = len(changes)
 	h.mu.Unlock()
 

internal/routewatch/prefixhandler_internal_test.go

@@ -237,87 +237,9 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
 	}
 }
 
-// TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a
+// longerThanAnyTest is a flush timer interval that no test lasts. A test whose
-// deadline that has passed, while a peer's PeerDown is held up and one message
+// prefix, AS or peer handler must flush only when its batch is full and on Stop
-// of the peer is held back. Stop must log that message as not applied.
+// gives it to newPrefixHandler, newASHandler or newPeerHandler.
-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
-// Stop.
 const longerThanAnyTest = time.Hour
 
 // prefixHandlerWithoutTimer returns a prefix handler that writes to store and
@@ -325,12 +247,9 @@ const longerThanAnyTest = time.Hour
 // NewPrefixHandler's flush goroutine, so the test flushes itself.
 func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
 	return &PrefixHandler{
-		db:                   store,
+		db:     store,
-		logger:               logger.New(),
+		logger: logger.New(),
-		table:                routetable.New(routetable.Params{}),
+		table:  routetable.New(routetable.Params{}),
-		batchMaxSize:         prefixBatchMaxSize,
-		tableChangesPerWrite: maxTableChangesPerWrite,
-		heldBack:             make(map[string][]*ristypes.RISMessage),
 	}
 }
 

internal/routewatch/prefixhandler_table_internal_test.go

@@ -7,10 +7,8 @@ import (
 	"log/slog"
 	"net/netip"
 	"reflect"
-	"runtime"
 	"slices"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
@@ -32,13 +30,12 @@ const (
 
 	// firstManyPrefix is the first prefix announceManyPrefixes announces.
 	firstManyPrefix = "10.0.0.0/32"
-
-	// testMaxTableChangesPerWrite is a limit of one write of the routing
-	// table's changes that a test exceeds quickly, in place of
-	// maxTableChangesPerWrite.
-	testMaxTableChangesPerWrite = 10
 )
 
+// tableWriteTimeout is how long a test waits for the flush timer to write the
+// routing table's changes: a few ticks of it.
+const tableWriteTimeout = 5 * prefixBatchTimeout
+
 // tableCase is one test of the routing table in memory: the messages handed to
 // the prefix handler, and for each prefix the peers that carry it with each
 // origin AS after them.
@@ -103,228 +100,6 @@ 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 these messages of the down peer, in this order:
-//   - an announcement;
-//   - its session going down again;
-//   - a second announcement.
-//
-// Once the PeerDowns have returned, the down peer must carry only the prefix
-// of the second announcement.
-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}},
-	})
-}
-
-// TestPrefixHandlerAppliesMessagesWhilePeerDownRuns gives a peer 50,000
-// prefixes and hands the prefix handler that peer's session going down. The
-// table's PeerDown then visits the prefixes in five batches. Once PeerDown has
-// begun, the handler is handed 1,000 announcements of another peer. The
-// handler must apply them all before PeerDown ends, while the down peer still
-// carries some of its prefixes.
-func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
-	t.Parallel()
-
-	const (
-		// downPeerPrefixes is five times the 10,000 prefixes PeerDown visits
-		// in one hold of the table's lock.
-		downPeerPrefixes = 50000
-		// otherPeerMessages is far more messages than the handler applies
-		// while PeerDown runs, if PeerDown takes the lock back at once after
-		// each batch. The handler then applies only a few for each batch.
-		otherPeerMessages = 1000
-		// firstOtherAddr is the address of the other peer's first prefix.
-		firstOtherAddr = "2001:db8:2::"
-	)
-
-	h := prefixHandlerWithoutTimer(newMockStore())
-	announceManyPrefixes(h.table, downPeerPrefixes)
-
-	h.HandleMessage(sessionDown())
-
-	// Wait until PeerDown has begun: the down peer has left some of its
-	// prefixes. The down peer's prefixes are IPv4 and the other peer's are
-	// IPv6, so the counts tell them apart.
-	for h.table.Counts().IPv4Routes == downPeerPrefixes {
-		runtime.Gosched()
-	}
-
-	// Each announcement names the /128 of the next address.
-	addr := netip.MustParseAddr(firstOtherAddr)
-	for range otherPeerMessages {
-		h.HandleMessage(peerAnnouncement(secondPeerIP,
-			netip.PrefixFrom(addr, addr.BitLen()).String()))
-
-		addr = addr.Next()
-	}
-
-	counts := h.table.Counts()
-	if counts.IPv6Routes != otherPeerMessages {
-		t.Errorf("the other peer carries %d prefixes, want %d",
-			counts.IPv6Routes, otherPeerMessages)
-	}
-
-	if counts.IPv4Routes == 0 {
-		t.Errorf("PeerDown ended before the handler applied the other peer's "+
-			"%d announcements", otherPeerMessages)
-	}
-
-	if !returnsInTime(h.peerDowns.Wait) {
-		t.Fatalf("the PeerDown did not return within %s", stepTimeout)
-	}
-}
-
-// TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many
-// messages of a peer whose session went down. Each names a prefix that does
-// not parse, so applying it logs an error. Each log write takes a
-// 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
@@ -423,9 +198,8 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
 	}
 }
 
-// checkTableCases hands the messages of each case to a new prefix handler. It
+// checkTableCases hands the messages of each case to a new prefix handler and
-// checks the handler's routing table once the handler's PeerDowns have
+// checks its routing table right after, with no flush.
-// returned, with no flush.
 func checkTableCases(t *testing.T, tests []tableCase) {
 	t.Helper()
 
@@ -438,28 +212,17 @@ func checkTableCases(t *testing.T, tests []tableCase) {
 				h.HandleMessage(msg)
 			}
 
-			h.peerDowns.Wait()
+			for prefix, want := range tt.want {
-			checkTableOrigins(t, h.table, tt.want)
+				got := tableOrigins(h.table, prefix)
+				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 {
@@ -513,9 +276,9 @@ func TestPrefixHandlerWritesTheTableChangesEachSecond(t *testing.T) {
 				t.Fatalf("wrote %+v, want the pair of %s and AS%d with "+
 					"removed %t", changes, prefix, testASNB, step.wantRemoved)
 			}
-		case <-time.After(stepTimeout):
+		case <-time.After(tableWriteTimeout):
 			t.Fatalf("the routing table's change was not written within %s",
-				stepTimeout)
+				tableWriteTimeout)
 		}
 	}
 }
@@ -612,32 +375,25 @@ func (s *tableWriteSizesStore) WriteRouteChanges(
 // takes, then flushes three times. The first write must hold as many pairs as
 // one write takes, and the peer. The second must hold the other pairs, and the
 // third flush must write nothing. The database must then hold the table in
-// memory. The handler's limit is small, and its timer does not fire.
+// memory.
 func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
 	t.Parallel()
 
 	const (
-		pairs   = testMaxTableChangesPerWrite * 3 / 2
+		pairs   = maxTableChangesPerWrite * 3 / 2
 		flushes = 3
 	)
 
 	db := newTestDatabase(t)
 	store := &tableWriteSizesStore{Database: db}
-	table := routetable.New(routetable.Params{})
+	h := prefixHandlerWithoutTimer(store)
-	announceManyPrefixes(table, pairs)
+	announceManyPrefixes(h.table, pairs)
-
-	h := newPrefixHandler(store, logger.New(), table, longerThanAnyTest,
-		testMaxTableChangesPerWrite)
-
-	defer stopInTime(t, h)
 
 	for range flushes {
 		h.flushBatch()
 	}
 
-	wantChanges := []int{
+	wantChanges := []int{maxTableChangesPerWrite, pairs - maxTableChangesPerWrite}
-		testMaxTableChangesPerWrite, pairs - testMaxTableChangesPerWrite,
-	}
 	wantNewPeers := []int{1, 0}
 
 	if !slices.Equal(store.changes, wantChanges) ||
@@ -659,11 +415,11 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
 // takes. Its last flush writes one write's worth, so Stop must log the pairs
 // that no write took, and must not say that the stop deadline passed. The
 // flush timer must not fire, or its writes could leave no pairs for Stop to
-// log. The handler's limit of one write is small.
+// log.
 func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
 	t.Parallel()
 
-	const pairs = testMaxTableChangesPerWrite + 1
+	const pairs = maxTableChangesPerWrite + 1
 
 	var logs bytes.Buffer
 
@@ -671,8 +427,7 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
 	table := routetable.New(routetable.Params{})
 	announceManyPrefixes(table, pairs)
 
-	h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest,
+	h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest)
-		testMaxTableChangesPerWrite)
 	h.Stop(t.Context())
 
 	written := 0
@@ -734,7 +489,6 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) {
 		h.HandleMessage(msg)
 	}
 
-	h.peerDowns.Wait()
 	h.flushBatch()
 
 	var logs bytes.Buffer
@@ -773,7 +527,6 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) {
 		restarted.HandleMessage(msg)
 	}
 
-	h.peerDowns.Wait()
 	restarted.Stop(t.Context())
 
 	read, err := db.ReadRouteTable(t.Context())

internal/server/home_internal_test.go

@@ -6,7 +6,6 @@ import (
 	"strconv"
 	"strings"
 	"testing"
-	"testing/synctest"
 	"time"
 
 	"git.eeqj.de/sneak/routewatch/internal/database"
@@ -342,66 +341,6 @@ func TestStartedServerWorksOutTheCards(t *testing.T) {
 	}
 }
 
-// TestStartedServerWorksOutTheCardsUntilStop starts a server with a context
-// that ends as soon as Start returns. It runs in a synctest bubble, whose clock
-// is fake: time.Sleep there moves the clock on and returns at once. The server
-// must work the cards out again a minute later, and no more after Stop. Each
-// time it works them out, it keeps a new copy. Before the fix the end of
-// Start's context stopped the goroutine that works them out.
-func TestStartedServerWorksOutTheCardsUntilStop(t *testing.T) {
-	t.Parallel()
-
-	// The server is made outside the bubble. The streamer's metrics start a
-	// goroutine that never returns, and a bubble waits for each of its
-	// goroutines to return.
-	s := newServerWithTable(t, newCardsTable())
-
-	synctest.Test(t, func(t *testing.T) {
-		ctx, cancel := context.WithCancel(t.Context())
-
-		err := s.Start(ctx)
-		if err != nil {
-			t.Fatalf("Start: %v", err)
-		}
-
-		cancel()
-
-		// The bubble's clock moves only while each of its goroutines waits
-		// on the bubble, and the goroutine that serves HTTP waits on the
-		// network. The test sends no request, so it closes the HTTP server.
-		_ = s.srv.Close()
-
-		synctest.Wait()
-
-		started := s.cards.Load()
-		if started == nil {
-			t.Fatal("the server did not work out the cards when it started")
-		}
-
-		time.Sleep(homeCardsRefreshInterval)
-		synctest.Wait()
-
-		if s.cards.Load() == started {
-			t.Error("the server did not work out the cards again a minute " +
-				"after its start context ended")
-		}
-
-		err = s.Stop(t.Context())
-		if err != nil {
-			t.Fatalf("Stop: %v", err)
-		}
-
-		stopped := s.cards.Load()
-
-		time.Sleep(homeCardsRefreshInterval)
-		synctest.Wait()
-
-		if s.cards.Load() != stopped {
-			t.Error("the server worked out the cards again after Stop")
-		}
-	})
-}
-
 // waitUntil fails the test unless done reports true within maxWait.
 func waitUntil(t *testing.T, maxWait time.Duration, done func() bool) {
 	t.Helper()

internal/server/server.go

@@ -82,12 +82,6 @@ type Server struct {
 	// cards is the copy of the home page's cards that the pages serve, nil
 	// until the server first works them out
 	cards atomic.Pointer[homeCards]
-
-	// stopCards ends the goroutine that works out the cards, and
-	// cardsStopped is closed once it has returned. Both are nil until Start
-	// starts the goroutine.
-	stopCards    context.CancelFunc
-	cardsStopped chan struct{}
 }
 
 // New creates a new HTTP server that reads routes from table, the routing
@@ -117,8 +111,7 @@ func New(
 // Start listens on the configured port and serves HTTP in the background.
 // It returns an error if it cannot listen, for example because the port is
 // already in use. ctx bounds opening the port, not serving. Once listening,
-// the server works out the home page's cards every minute until Stop: the end
+// the server works out the home page's cards every minute until ctx ends.
-// of ctx does not stop it.
 func (s *Server) Start(ctx context.Context) error {
 	const (
 		network           = "tcp"
@@ -155,22 +148,16 @@ func (s *Server) Start(ctx context.Context) error {
 		}
 	}()
 
-	ctx, s.stopCards = context.WithCancel(context.WithoutCancel(ctx))
-	s.cardsStopped = make(chan struct{})
-
 	go func() {
-		defer close(s.cardsStopped)
-
 		s.keepHomeCardsFresh(ctx, homeCardsRefreshInterval)
 	}()
 
 	return nil
 }
 
-// Stop stops the goroutine that works out the home page's cards, and waits
+// Stop stops the HTTP server. It waits, until ctx ends, for the requests in
-// until it has returned or ctx ends. It then stops the HTTP server. It waits,
+// progress to finish, and then closes the connections of those still running
-// until ctx ends, for the requests in progress to finish, and then closes the
+// and returns ctx's error.
-// connections of those still running and returns ctx's error.
 func (s *Server) Stop(ctx context.Context) error {
 	if s.srv == nil {
 		return nil
@@ -178,16 +165,6 @@ func (s *Server) Stop(ctx context.Context) error {
 
 	s.logger.Info("Stopping HTTP server")
 
-	// Start returns before it starts the goroutine when it cannot listen.
-	if s.stopCards != nil {
-		s.stopCards()
-
-		select {
-		case <-s.cardsStopped:
-		case <-ctx.Done():
-		}
-	}
-
 	err := s.srv.Shutdown(ctx)
 	if err != nil {
 		return errors.Join(err, s.srv.Close())

internal/whois/whois.go

@@ -12,7 +12,6 @@ import (
 	"regexp"
 	"strings"
 	"sync"
-	"syscall"
 	"time"
 	"unicode/utf8"
 
@@ -71,12 +70,12 @@ const nicbrHeader = "% Copyright (c) Nic.br"
 // queries to stop the data being taken in bulk.
 const nicbrLimitReached = "Query rate limit exceeded"
 
-// PauseTime is how long the client sends an RIR's or NIC.br's server no
+// NICbrSkipTime is how long the client stops asking NIC.br after NIC.br
-// query after that server refuses a query. queryRegistry says what a
+// answers that the client is over its query limit. NIC.br publishes
-// refusal is. RIPE refuses an address for good when the address keeps
+// neither the limit nor how long it lasts. An hour outlasts a limit
-// passing RIPE's limit (see noPersonalFlag). An hour outlasts a limit
+// counted per minute or per hour. The ASes looked up meanwhile get
-// counted per minute or per hour.
+// NIC.br's fields when they are next refreshed.
-const PauseTime = time.Hour
+const NICbrSkipTime = time.Hour
 
 // RIR identifiers.
 const (
@@ -126,10 +125,6 @@ var (
 	errAnswerTooLarge = errors.New("WHOIS answer reached the size cap")
 )
 
-// ErrPaused is returned when a lookup needs a server that refused a query
-// less than PauseTime ago. The client sends that server nothing meanwhile.
-var ErrPaused = errors.New("WHOIS server paused after refusing a query")
-
 // ASNInfo contains parsed WHOIS information for an ASN.
 type ASNInfo struct {
 	ASN         int
@@ -161,12 +156,12 @@ type Client struct {
 
 	logger *slog.Logger
 
-	// mu guards pausedUntil.
+	// mu guards nicbrSkipUntil.
 	mu sync.Mutex
 
-	// pausedUntil maps a server, given as host:port, to the time the client
+	// nicbrSkipUntil is when the client next asks NIC.br, after NIC.br
-	// may ask it again.
+	// answered that the client is over its query limit.
-	pausedUntil map[string]time.Time
+	nicbrSkipUntil time.Time
 }
 
 // NewClient creates a new WHOIS client that logs to logger.
@@ -177,29 +172,23 @@ func NewClient(logger *slog.Logger) *Client {
 		DialContext: dialer.DialContext,
 		Now:         time.Now,
 		logger:      logger,
-		pausedUntil: make(map[string]time.Time),
 	}
 }
 
 // LookupASN asks IANA which RIR holds the given ASN, asks that RIR, and
 // returns what the RIR's answer says. When LACNIC's answer is NIC.br's, it
 // returns what NIC.br's own server says instead, unless that query fails or
-// NIC.br is paused, as askNICbr says.
+// is skipped, as askNICbr says.
 //
-// It returns ErrNoReferral when IANA refers the AS to no RIR. Its error
+// It returns ErrNoReferral when IANA refers the AS to no RIR, and another
-// wraps ErrPaused when the RIR is paused. It returns another error in each
+// error when the query to IANA or to the RIR fails, when the RIR refuses
-// of these cases:
+// it, when IANA's answer has neither a referral nor the AS's block, when
-//
+// IANA refers the AS to a server none of the five RIRs run, or when the
-//   - the query to IANA or to the RIR fails
+// answer carries none of ASNInfo's fields.
-//   - the RIR refuses the query
-//   - IANA's answer has neither a referral nor the AS's block
-//   - IANA refers the AS to a server none of the five RIRs run
-//   - the answer carries none of ASNInfo's fields
 func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
 	// Query IANA first to find the authoritative RIR
 	query := fmt.Sprintf("AS%d", asn)
 
-	// IANA is never paused, because a pause of IANA would stop every lookup.
 	ianaResp, err := c.query(ctx, whoisServerIANA, query)
 	if err != nil {
 		return nil, fmt.Errorf("IANA query failed: %w", err)
@@ -222,7 +211,7 @@ func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
 	}
 
 	// Query the authoritative RIR
-	rirResp, err := c.queryRegistry(ctx, whoisServer, rirQuery(rir, query))
+	rirResp, err := c.queryRIR(ctx, whoisServer, rirQuery(rir, query))
 	if err != nil {
 		return nil, fmt.Errorf("RIR query failed: %w", err)
 	}
@@ -252,19 +241,22 @@ func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
 // askNICbr returns lacnicInfo, what LACNIC's answer says, and logs a
 // warning when the query fails, when NIC.br sends the reduced answer for a
 // client over its limit, or when NIC.br's answer carries none of ASNInfo's
-// fields. The reduced answer pauses NIC.br, as a refusal does. While NIC.br
+// fields. After the reduced answer, it returns lacnicInfo without asking
-// is paused, askNICbr returns lacnicInfo without a warning. The ASes looked
+// NIC.br until NICbrSkipTime has passed.
-// up meanwhile get NIC.br's fields when they are next refreshed.
 func (c *Client) askNICbr(
 	ctx context.Context, asQuery string, lacnicInfo *ASNInfo,
 ) *ASNInfo {
 	asn := lacnicInfo.ASN
 
-	answer, err := c.queryRegistry(ctx, whoisServerNICbr, asQuery)
+	c.mu.Lock()
-	if errors.Is(err, ErrPaused) {
+	skipping := c.Now().Before(c.nicbrSkipUntil)
+	c.mu.Unlock()
+
+	if skipping {
 		return lacnicInfo
 	}
 
+	answer, err := c.query(ctx, whoisServerNICbr, asQuery)
 	if err != nil {
 		c.logger.Warn("NIC.br query failed, keeping LACNIC's fields",
 			"asn", asn, "error", err)
@@ -273,11 +265,13 @@ func (c *Client) askNICbr(
 	}
 
 	if strings.Contains(answer, nicbrLimitReached) {
-		c.pause(whoisServerNICbr)
+		c.mu.Lock()
+		c.nicbrSkipUntil = c.Now().Add(NICbrSkipTime)
+		c.mu.Unlock()
 
 		c.logger.Warn("NIC.br answered that the client is over its query "+
-			"limit, keeping LACNIC's fields and pausing NIC.br",
+			"limit, keeping LACNIC's fields and skipping NIC.br",
-			"asn", asn, "pause_for", PauseTime)
+			"asn", asn, "skip_for", NICbrSkipTime)
 
 		return lacnicInfo
 	}
@@ -315,65 +309,25 @@ func rirQuery(rir, asQuery string) string {
 	}
 }
 
-// queryRegistry sends query to the server of an RIR or of NIC.br, and
+// queryRIR sends query to an RIR's server and returns the answer. An answer
-// returns the answer. The client pauses the server for PauseTime when the
+// holding an error line with which the server refuses the query is an
-// server refuses the query. Each of these is a refusal:
+// error, and the error holds that line.
-//
+func (c *Client) queryRIR(
-//   - a refused connection
-//   - a reset connection
-//   - an %ERROR:201: or %ERROR:202: line in the answer
-//
-// A refusal is returned as an error. While the server is paused,
-// queryRegistry sends it nothing and returns an error that wraps ErrPaused.
-func (c *Client) queryRegistry(
 	ctx context.Context, server, query string,
 ) (string, error) {
-	c.mu.Lock()
-	pausedUntil := c.pausedUntil[server]
-	c.mu.Unlock()
-
-	if c.Now().Before(pausedUntil) {
-		return "", fmt.Errorf("%w: %s until %s", ErrPaused, server,
-			pausedUntil.UTC().Format(time.RFC3339))
-	}
-
 	answer, err := c.query(ctx, server, query)
-	if err == nil {
-		err = checkRefusal(answer)
-	}
-
-	if errors.Is(err, errRefused) || errors.Is(err, syscall.ECONNREFUSED) ||
-		errors.Is(err, syscall.ECONNRESET) {
-		c.pause(server)
-	}
-
 	if err != nil {
 		return "", err
 	}
 
-	return answer, nil
-}
-
-// checkRefusal returns an error holding the first line of answer that
-// starts with accessDeniedError or accessLimitError. It returns nil when
-// answer has no such line.
-func checkRefusal(answer string) error {
 	for line := range strings.SplitSeq(answer, "\n") {
 		if strings.HasPrefix(line, accessDeniedError) ||
 			strings.HasPrefix(line, accessLimitError) {
-			return fmt.Errorf("%w: %s", errRefused, line)
+			return "", fmt.Errorf("%w: %s", errRefused, line)
 		}
 	}
 
-	return nil
+	return answer, nil
-}
-
-// pause stops queryRegistry from querying server until PauseTime has passed.
-func (c *Client) pause(server string) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.pausedUntil[server] = c.Now().Add(PauseTime)
 }
 
 // query performs a raw WHOIS query to the specified server and returns the

internal/whois/whois_test.go

@@ -718,10 +718,9 @@ func TestLookupASNKeepsLACNICsAnswerWhenNICbrFails(t *testing.T) {
 
 // TestLookupASNSkipsNICbrAfterItsLimit looks up an AS that NIC.br
 // registers three times with one client, while NIC.br answers that the
-// client is over its query limit. The test sets the client's clock. The
+// client is over its query limit. The first lookup asks NIC.br. The
-// first lookup asks NIC.br. The next comes a second before PauseTime has
+// second, a second before NICbrSkipTime has passed, does not. The third,
-// passed, and does not ask NIC.br. The last comes at PauseTime and asks
+// once NICbrSkipTime has passed, asks again. Each keeps LACNIC's fields.
-// NIC.br again. Each keeps LACNIC's fields.
 func TestLookupASNSkipsNICbrAfterItsLimit(t *testing.T) {
 	t.Parallel()
 
@@ -741,8 +740,8 @@ func TestLookupASNSkipsNICbrAfterItsLimit(t *testing.T) {
 		wantAsks int
 	}{
 		{after: 0, wantAsks: 1},
-		{after: whois.PauseTime - time.Second, wantAsks: 1},
+		{after: whois.NICbrSkipTime - time.Second, wantAsks: 1},
-		{after: whois.PauseTime, wantAsks: 2},
+		{after: whois.NICbrSkipTime, wantAsks: 2},
 	} {
 		now = start.Add(step.after)
 
@@ -786,142 +785,6 @@ func TestLookupASNReportsARefusal(t *testing.T) {
 	}
 }
 
-// resetServer starts a TCP server on the loopback address and returns the
-// server's address. The server reads one query on each connection, then
-// resets the connection.
-func resetServer(t *testing.T) string {
-	t.Helper()
-
-	var listenConfig net.ListenConfig
-
-	listener, err := listenConfig.Listen(t.Context(), "tcp", "127.0.0.1:0")
-	if err != nil {
-		t.Fatalf("listen: %v", err)
-	}
-
-	t.Cleanup(func() { _ = listener.Close() })
-
-	go func() {
-		for {
-			conn, err := listener.Accept()
-			if err != nil {
-				return
-			}
-
-			_, _ = bufio.NewReader(conn).ReadString('\n')
-
-			// Closing a connection whose linger time is zero resets it.
-			if tcpConn, ok := conn.(*net.TCPConn); ok {
-				_ = tcpConn.SetLinger(0)
-			}
-
-			_ = conn.Close()
-		}
-	}()
-
-	return listener.Addr().String()
-}
-
-// newRIPEClient returns a client whose fake servers send answers. The
-// client adds one to *ripeAsks each time it connects to RIPE's server. When
-// reset is true, RIPE's server resets the connection instead of answering.
-func newRIPEClient(
-	t *testing.T, answers map[string]string, reset bool, ripeAsks *int,
-) *whois.Client {
-	t.Helper()
-
-	dial := fakeServers(t, answers)
-	resetAddress := resetServer(t)
-
-	client := whois.NewClient(slog.New(slog.DiscardHandler))
-	client.DialContext = func(
-		ctx context.Context, network, address string,
-	) (net.Conn, error) {
-		if address == ripeServer {
-			*ripeAsks++
-		}
-
-		if address == ripeServer && reset {
-			var dialer net.Dialer
-
-			return dialer.DialContext(ctx, network, resetAddress)
-		}
-
-		return dial(ctx, network, address)
-	}
-
-	return client
-}
-
-// TestLookupASNPausesAServerThatRefuses looks up an AS at RIPE while RIPE
-// refuses the query, in each of the ways a server refuses. The test sets
-// the client's clock. A lookup a second before PauseTime has passed fails
-// with ErrPaused and asks RIPE nothing. A lookup of an AS at APNIC succeeds
-// during the pause. A lookup at PauseTime asks RIPE again.
-func TestLookupASNPausesAServerThatRefuses(t *testing.T) {
-	t.Parallel()
-
-	for name, test := range map[string]struct {
-		// ripeAnswer is RIPE's answer. With none, RIPE's server refuses
-		// the connection, or resets it when reset is true.
-		ripeAnswer string
-		reset      bool
-	}{
-		"access denied":      {ripeAnswer: ripeRefusal},
-		"limit reached":      {ripeAnswer: ripeLimitReached + "\n"},
-		"refused connection": {},
-		"reset connection":   {reset: true},
-	} {
-		t.Run(name, func(t *testing.T) {
-			t.Parallel()
-
-			answers := map[string]string{
-				ianaQuery:      ianaReferral,
-				apnicIANAQuery: "refer:        whois.apnic.net\n",
-				apnicQuery:     apnicRecord,
-			}
-			if test.ripeAnswer != "" {
-				answers[ripeQuery] = test.ripeAnswer
-			}
-
-			ripeAsks := 0
-			client := newRIPEClient(t, answers, test.reset, &ripeAsks)
-
-			start := time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
-			now := start
-			client.Now = func() time.Time { return now }
-
-			_, err := client.LookupASN(t.Context(), testASN)
-			if err == nil || errors.Is(err, whois.ErrPaused) {
-				t.Fatalf("LookupASN while RIPE refuses: error %v, want a "+
-					"refusal", err)
-			}
-
-			now = start.Add(whois.PauseTime - time.Second)
-
-			_, err = client.LookupASN(t.Context(), testASN)
-			if !errors.Is(err, whois.ErrPaused) || ripeAsks != 1 {
-				t.Errorf("LookupASN during the pause: error %v after %d "+
-					"queries to RIPE, want ErrPaused after 1", err, ripeAsks)
-			}
-
-			_, err = client.LookupASN(t.Context(), apnicASN)
-			if err != nil {
-				t.Errorf("LookupASN of an AS at APNIC during RIPE's "+
-					"pause: %v", err)
-			}
-
-			now = start.Add(whois.PauseTime)
-
-			_, _ = client.LookupASN(t.Context(), testASN)
-			if ripeAsks != 2 {
-				t.Errorf("RIPE was asked %d times after the pause, want 2",
-					ripeAsks)
-			}
-		})
-	}
-}
-
 // TestLookupASNFailsWhenTheRIRGivesNoAnswer looks up an AS IANA refers to
 // RIPE when RIPE's server refuses the connection, and when it answers only
 // with an error. Neither is a lookup that found the AS.