Pause a registry's WHOIS server after it refuses a query (closes #218)

The fetcher kept one interval for every registry, so a registry that
refused a query was asked again within seconds. RIPE blocked the test
host's address that way.

The WHOIS client now pauses an RIR's or NIC.br's server for an hour
after the server refuses a query. A refused or reset connection is a
refusal, and so is an %ERROR:201: or %ERROR:202: line. Lookups at the
other servers go on. A lookup whose RIR is paused leaves the AS stale,
and counts as neither a success nor a failure. IANA is never paused,
because a pause of IANA would stop every lookup. NIC.br's hour-long
skip after its over-limit answer is now the same pause.

Model: opus-5-5

README.md

@@ -429,12 +429,10 @@ the routing table, and everything else from the database.
     - Parses BGP UPDATE messages from JSON stream
     - Dispatches to registered handlers based on message type
     - Each handler has its own queue with backpressure handling
-    - Handlers batch writes for efficiency. The AS, peer and prefix handlers
+    - Handlers batch writes for efficiency: the AS, peer and prefix handlers
-      write a batch when it reaches a size, or after a time:
+      write 30,000, 10,000 and 25,000 updates at a time, or after 2, 2 and 1
-        - the AS handler at 30,000 ASes, or after 2 seconds;
+      seconds; the peering handler writes the peerings it has collected every 30
-        - the peer handler at 10,000 updates, or after 2 seconds;
+      seconds
-        - the prefix handler at 25,000 updates, or after 1 second.
-    - The peering handler writes the peerings it has collected every 30 seconds
 
 4. **Handler Details**
     - **ASHandler**: Tracks all ASNs seen in AS paths
@@ -634,19 +632,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.
 
 With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
 with the goroutine count and Go memory figures.

TODO.md

@@ -43,11 +43,6 @@ tested.
   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

internal/routewatch/app_integration_internal_test.go

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

internal/routewatch/ashandler.go

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

internal/routewatch/ashandler_internal_test.go

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

internal/routewatch/batchwrite_internal_test.go

@@ -178,7 +178,6 @@ type stopLog struct {
 	Updates      int    `json:"updates"`
 	TableChanges int    `json:"table_changes"`
 	Paths        int    `json:"paths"`
-	BatchEntries int    `json:"batch_entries"`
 }
 
 // newErrorLogger returns a logger that writes only errors, as JSON lines, to
@@ -496,20 +495,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()
@@ -519,27 +518,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()
@@ -705,8 +703,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()
@@ -763,41 +762,25 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
 	}
 }
 
-// TestASHandlerNeverWaitsForAWrite starts a write that the store holds. Then it
+func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
-// hands the AS handler twenty messages whose paths each hold one AS
-// asnBatchSize times. HandleMessage must return each time, and the batch must
-// hold one entry, for that AS.
-func TestASHandlerNeverWaitsForAWrite(t *testing.T) {
 	t.Parallel()
 
-	const messages = 20
-
 	store := newBlockingStore()
 	h := newASHandler(store, logger.New(), longerThanAnyTest)
+	h.batchMaxSize = asnBatchSize
+	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
+	toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
+	handleAtMaxBatchSize(t, store, h,
+		[]*ristypes.RISMessage{{Path: fill}},
+		[]*ristypes.RISMessage{{Path: toMax}},
+		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
 
-	defer stopInTime(t, h)
+	store.mu.Lock()
-	defer store.releaseWrites()
+	defer store.mu.Unlock()
 
-	startBlockedWrite(t, store, h,
+	if !store.asns[testASNB] {
-		[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}})
+		t.Errorf("Stop did not write AS%d, handled after the batch reached its "+
-
+			"maximum size", testASNB)
-	oneAS := &ristypes.RISMessage{
-		Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)},
-	}
-
-	if !returnsInTime(func() {
-		for range messages {
-			h.HandleMessage(oneAS)
-		}
-	}) {
-		t.Fatal("HandleMessage did not return while a database write was blocked")
-	}
-
-	h.mu.Lock()
-	defer h.mu.Unlock()
-
-	if len(h.batch) != 1 {
-		t.Errorf("the batch holds %d entries for one AS, want 1", len(h.batch))
 	}
 }
 
@@ -861,11 +844,13 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
 			batchSize: asnBatchSize,
 			handle: func(n int) chan struct{} {
 				h := &ASHandler{
-					batch:       make(map[int]database.SeenTimes),
+					batchMaxSize: asnBatchMaxSize,
-					batchFullCh: make(chan struct{}, 1),
+					batchFullCh:  make(chan struct{}, 1),
 				}
-				// Each of the path's n ASes is one entry of the batch.
+				// The origin AS and each AS in the path are one update each.
-				h.HandleMessage(&ristypes.RISMessage{Path: pathOfASes(n)})
+				h.HandleMessage(&ristypes.RISMessage{
+					Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)},
+				})
 
 				return h.batchFullCh
 			},