Checkpoint without waiting on readers and keep the WAL file small (closes #102)

The 5-second checkpoint ran in TRUNCATE mode, which waits up to the
5-second busy timeout for every open read while holding the database
lock every batch write needs. It now runs in PASSIVE mode, which never
waits. Because PASSIVE alone never lets the WAL start again while reads
overlap, a WAL file over 64 MiB gets a TRUNCATE checkpoint whose wait
for reads is capped at 500 ms and logged. A checkpoint that leaves part
of the WAL uncopied returns ErrCheckpointIncomplete, which the
maintainer logs as incomplete instead of completed. The writing
connection's journal_size_limit is 64 MiB.

Model: opus-5-5

README.md

@@ -318,8 +318,14 @@ What happens at each limit:
   (`peerBatchMaxSize`, about 10 MiB) and 250,000 for the prefix handler
   (`prefixBatchMaxSize`, about 60 MiB). At that size the handler takes nothing
   more off its queue until the write ends, so the queue fills and the streamer
-  drops messages as above. With the batches being written, the three handlers'
+  drops messages as above. When the delete of a peer's live routes after its
-  batches take at most about 0.25 GiB of the Go heap's 1.5 GiB.
+  session went down fails, the prefix handler keeps the updates the peer sent
+  after the down for the next write, which retries the delete first; on top of
+  its 250,000 it keeps at most the last 25,000 of them (`prefixBatchSize`, about
+  6 MiB) and drops and logs the older ones, so while the delete keeps failing
+  the other peers' updates still come off the queue. With the batches being
+  written, the three handlers' batches take at most about 0.25 GiB of the Go
+  heap's 1.5 GiB.
 
 With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
 with the goroutine count and Go memory figures.

internal/database/counts_internal_test.go

@@ -2,6 +2,7 @@ package database
 
 import (
 	"context"
+	"fmt"
 	"slices"
 	"sync"
 	"testing"
@@ -494,6 +495,124 @@ func TestDeleteLiveRoutesOfPeer(t *testing.T) {
 	checkOneLiveRoutePerIPVersion(t, db)
 }
 
+// TestDeleteLiveRoutesOfPeerDeletesInChunks gives a peer one IPv4 route more
+// than a chunk holds, two of them for prefixes another peer also has. One chunk
+// deletes peerRouteChunkSize of the peer's routes and commits, leaving the last
+// one. The routes are written again, and DeleteLiveRoutesOfPeer deletes all of
+// them, a chunk and then the last one. After each step the route count and
+// prefix distribution held in memory and those read from the route table are
+// the ones wanted.
+func TestDeleteLiveRoutesOfPeerDeletesInChunks(t *testing.T) {
+	t.Parallel()
+
+	const (
+		peerIP      = "192.0.2.1"
+		otherPeerIP = "192.0.2.2"
+		// How many of the peer's prefixes the other peer also has, the first
+		// ones in prefix order.
+		sharedPrefixes = 2
+	)
+
+	db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
+
+	peerRoutes := make([]*LiveRoute, 0, peerRouteChunkSize+1)
+	otherPeerRoutes := make([]*LiveRoute, 0, sharedPrefixes)
+
+	for i := range peerRouteChunkSize + 1 {
+		prefix := fmt.Sprintf("10.%d.%d.0/24", i/256, i%256)
+
+		route := mkV4Route(t, prefix, 64500, ts)
+		route.PeerIP = peerIP
+		peerRoutes = append(peerRoutes, route)
+
+		if i < sharedPrefixes {
+			route := mkV4Route(t, prefix, 64500, ts)
+			route.PeerIP = otherPeerIP
+			otherPeerRoutes = append(otherPeerRoutes, route)
+		}
+	}
+
+	err = db.UpsertLiveRouteBatch(slices.Concat(peerRoutes, otherPeerRoutes))
+	if err != nil {
+		t.Fatalf("UpsertLiveRouteBatch: %v", err)
+	}
+
+	deletedV4, deletedV6, err := db.deletePeerRouteChunk(peerIP)
+	if err != nil {
+		t.Fatalf("deletePeerRouteChunk: %v", err)
+	}
+
+	if deletedV4 != peerRouteChunkSize || deletedV6 != 0 {
+		t.Errorf("deletePeerRouteChunk deleted (%d, %d) routes, want (%d, 0)",
+			deletedV4, deletedV6, peerRouteChunkSize)
+	}
+
+	// The shared prefixes went with the chunk, so each route left has a
+	// prefix of its own.
+	checkV4RoutesOfOwnPrefixes(t, db, "after one chunk", 1+sharedPrefixes)
+
+	err = db.UpsertLiveRouteBatch(peerRoutes)
+	if err != nil {
+		t.Fatalf("UpsertLiveRouteBatch: %v", err)
+	}
+
+	deleted, err := db.DeleteLiveRoutesOfPeer(peerIP)
+	if err != nil {
+		t.Fatalf("DeleteLiveRoutesOfPeer: %v", err)
+	}
+
+	if deleted != len(peerRoutes) {
+		t.Errorf("DeleteLiveRoutesOfPeer deleted %d routes, want %d",
+			deleted, len(peerRoutes))
+	}
+
+	checkV4RoutesOfOwnPrefixes(t, db, "after DeleteLiveRoutesOfPeer",
+		sharedPrefixes)
+}
+
+// checkV4RoutesOfOwnPrefixes checks that the database has wantRoutes IPv4 live
+// routes, each for a /24 no other live route has, both as counted in memory and
+// as read from the route table. when says which step of the test checks.
+func checkV4RoutesOfOwnPrefixes(
+	t *testing.T, db *Database, when string, wantRoutes int,
+) {
+	t.Helper()
+
+	ctx := context.Background()
+
+	stats, err := db.GetStatsContext(ctx)
+	if err != nil {
+		t.Fatalf("GetStatsContext: %v", err)
+	}
+
+	tableRoutes, _, err := db.GetLiveRouteCountsContext(ctx)
+	if err != nil {
+		t.Fatalf("GetLiveRouteCountsContext: %v", err)
+	}
+
+	if stats.IPv4Routes != wantRoutes || tableRoutes != wantRoutes {
+		t.Errorf("%s: IPv4 live routes: %d in memory, %d in the table, want %d",
+			when, stats.IPv4Routes, tableRoutes, wantRoutes)
+	}
+
+	queryDistribution, _, err := db.GetPrefixDistributionContext(ctx)
+	if err != nil {
+		t.Fatalf("GetPrefixDistributionContext: %v", err)
+	}
+
+	want := []PrefixDistribution{{MaskLength: 24, Count: wantRoutes}}
+	assertDistribution(t, when+": IPv4 distribution",
+		stats.IPv4PrefixDistribution, want)
+	assertDistribution(t, when+": IPv4 distribution query",
+		queryDistribution, want)
+}
+
 // checkOneLiveRoutePerIPVersion checks that the database has one IPv4 live
 // route, for a /24, and one IPv6 live route, for a /32, both as counted in
 // memory and as read from the route tables.

internal/database/database.go

@@ -81,6 +81,21 @@ const (
 	sqliteHardHeapLimitBytes = 2147483648
 )
 
+// Settings that keep the WAL file small (see Checkpoint).
+const (
+	// sqliteWALSizeLimitBytes (64 MiB) is the WAL file size past which
+	// Checkpoint truncates the file. It is also the journal_size_limit of the
+	// connection kept for writing: when SQLite starts the WAL again from the
+	// beginning, it cuts a larger WAL file down to this size.
+	sqliteWALSizeLimitBytes = 67108864
+	// truncateBusyTimeoutMs is the busy timeout of the TRUNCATE checkpoint:
+	// the longest it waits for open reads, and so the longest it keeps batch
+	// writes waiting beyond the copy itself.
+	truncateBusyTimeoutMs = 500
+	// walFileSuffix ends the name of the WAL file, next to the database file.
+	walFileSuffix = "-wal"
+)
+
 // Settings of the ANALYZE that Analyze runs.
 const (
 	// analysisLimit is about how many rows of each index ANALYZE reads
@@ -112,6 +127,11 @@ var (
 	ErrNoRoute = errors.New("no route found")
 	// ErrNoStaleASN is returned when no ASN needs WHOIS refresh
 	ErrNoStaleASN = errors.New("no stale ASN found")
+	// ErrCheckpointIncomplete is returned by Checkpoint when it left part of
+	// the WAL uncopied, or left a WAL file over sqliteWALSizeLimitBytes
+	// untruncated, because reads still needed the WAL or SQLite reported the
+	// checkpoint busy. A later checkpoint finishes the work.
+	ErrCheckpointIncomplete = errors.New("WAL checkpoint incomplete")
 
 	errInvalidASN       = errors.New("invalid ASN")
 	errSameASNInPeering = errors.New("cannot create peering with same ASN")
@@ -194,6 +214,13 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 			err)
 	}
 
+	_, err = writer.ExecContext(ctx,
+		fmt.Sprintf("PRAGMA journal_size_limit=%d", sqliteWALSizeLimitBytes))
+	if err != nil {
+		return nil, fmt.Errorf(
+			"failed to set the writing connection's WAL size limit: %w", err)
+	}
+
 	database := &Database{
 		db: db, writer: writer, logger: logger, path: dbPath,
 		counts: &liveCounts{},
@@ -658,48 +685,71 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 	return nil
 }
 
-// Deleting all live routes of a peer. The first query of each table selects the
+// peerRouteChunkSize is the most routes of a peer that one transaction of
-// mask length of each of the peer's routes whose prefix no other peer has a
+// DeleteLiveRoutesOfPeer deletes from each route table. A peer can have a
-// live route for: the delete takes those prefixes out of the prefix
+// million routes; deleted in one transaction, they would hold the database lock
-// distribution.
+// for most of a minute.
-const (
+const peerRouteChunkSize = 10000
-	peerOnlyRouteMaskLengthsV4SQL = `SELECT r.mask_length
-		FROM live_routes_v4 AS r WHERE r.peer_ip = ?
-		AND NOT EXISTS (SELECT 1 FROM live_routes_v4 AS o
-		WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)`
-	deletePeerLiveRoutesV4SQL = `DELETE FROM live_routes_v4 WHERE peer_ip = ?`
 
-	peerOnlyRouteMaskLengthsV6SQL = `SELECT r.mask_length
+// Deleting a chunk of a peer's live routes. The first query of each table
+// selects the peer's first routes in prefix order, through the
+// UNIQUE(peer_ip, prefix) key, each with its mask length and whether another
+// peer has a live route for its prefix: the delete takes the prefixes no other
+// peer has out of the prefix distribution. The second query deletes the peer's
+// routes up to the last prefix the first one selected.
+const (
+	peerRouteChunkV4SQL = `SELECT r.prefix, r.mask_length,
+		EXISTS (SELECT 1 FROM live_routes_v4 AS o
+		WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)
+		FROM live_routes_v4 AS r WHERE r.peer_ip = ?
+		ORDER BY r.prefix LIMIT ?`
+	deletePeerRouteChunkV4SQL = `DELETE FROM live_routes_v4
+		WHERE peer_ip = ? AND prefix <= ?`
+
+	peerRouteChunkV6SQL = `SELECT r.prefix, r.mask_length,
+		EXISTS (SELECT 1 FROM live_routes_v6 AS o
+		WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)
 		FROM live_routes_v6 AS r WHERE r.peer_ip = ?
-		AND NOT EXISTS (SELECT 1 FROM live_routes_v6 AS o
+		ORDER BY r.prefix LIMIT ?`
-		WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)`
+	deletePeerRouteChunkV6SQL = `DELETE FROM live_routes_v6
-	deletePeerLiveRoutesV6SQL = `DELETE FROM live_routes_v6 WHERE peer_ip = ?`
+		WHERE peer_ip = ? AND prefix <= ?`
 )
 
-// deletePeerRoutes deletes all live routes of peerIP from one table, given that
+// deletePeerRoutes deletes the first peerRouteChunkSize live routes of peerIP,
-// table's two statements above. It returns how many routes it deleted and the
+// or as many as it has, from one table, given that table's two statements
-// mask length of each prefix it left with no live route.
+// above. It returns how many routes it deleted and the mask length of each
+// prefix it left with no live route.
 func deletePeerRoutes(
-	ctx context.Context, tx *loggingTx,
+	ctx context.Context, tx *loggingTx, chunkSQL, deleteSQL, peerIP string,
-	onlyRouteMaskLengthsSQL, deleteSQL, peerIP string,
 ) (int64, []int, error) {
-	rows, err := tx.QueryContext(ctx, onlyRouteMaskLengthsSQL, peerIP)
+	rows, err := tx.QueryContext(ctx, chunkSQL, peerIP, peerRouteChunkSize)
 	if err != nil {
 		return 0, nil, err
 	}
 	defer func() { _ = rows.Close() }()
 
-	var goneMaskLengths []int
+	var (
+		selected        int
+		lastPrefix      string
+		goneMaskLengths []int
+	)
 
 	for rows.Next() {
-		var maskLength int
+		var (
+			maskLength         int
+			otherPeerHasARoute bool
+		)
 
-		err := rows.Scan(&maskLength)
+		err := rows.Scan(&lastPrefix, &maskLength, &otherPeerHasARoute)
 		if err != nil {
 			return 0, nil, err
 		}
 
-		goneMaskLengths = append(goneMaskLengths, maskLength)
+		selected++
+
+		if !otherPeerHasARoute {
+			goneMaskLengths = append(goneMaskLengths, maskLength)
+		}
 	}
 
 	err = rows.Err()
@@ -707,7 +757,11 @@ func deletePeerRoutes(
 		return 0, nil, err
 	}
 
-	res, err := tx.ExecContext(ctx, deleteSQL, peerIP)
+	if selected == 0 {
+		return 0, nil, nil
+	}
+
+	res, err := tx.ExecContext(ctx, deleteSQL, peerIP, lastPrefix)
 	if err != nil {
 		return 0, nil, err
 	}
@@ -721,43 +775,26 @@ func deletePeerRoutes(
 }
 
 // DeleteLiveRoutesOfPeer deletes every live route of the peer, IPv4 and IPv6,
-// in one transaction, as when the peer's BGP session goes down. It returns how
+// as when the peer's BGP session goes down. It deletes them a chunk at a time
-// many routes it deleted.
+// with deletePeerRouteChunk, each chunk in a transaction of its own, and
+// releases the database lock between chunks so other writes go on. It returns
+// how many routes it deleted; after an error, the chunks it deleted before stay
+// deleted.
 func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
-	d.lock("DeleteLiveRoutesOfPeer")
+	total := 0
-	defer d.unlock()
 
-	tx, err := d.beginTx()
+	for {
-	if err != nil {
+		deletedV4, deletedV6, err := d.deletePeerRouteChunk(peerIP)
-		return 0, fmt.Errorf("failed to begin transaction: %w", err)
+		if err != nil {
+			return total, err
+		}
+
+		total += deletedV4 + deletedV6
+
+		if deletedV4 < peerRouteChunkSize && deletedV6 < peerRouteChunkSize {
+			return total, nil
+		}
 	}
-	defer d.rollback(tx)
-
-	ctx := context.Background()
-
-	deletedV4, goneMaskLengthsV4, err := deletePeerRoutes(ctx, tx,
-		peerOnlyRouteMaskLengthsV4SQL, deletePeerLiveRoutesV4SQL, peerIP)
-	if err != nil {
-		return 0, fmt.Errorf("failed to delete the IPv4 routes of peer %s: %w",
-			peerIP, err)
-	}
-
-	deletedV6, goneMaskLengthsV6, err := deletePeerRoutes(ctx, tx,
-		peerOnlyRouteMaskLengthsV6SQL, deletePeerLiveRoutesV6SQL, peerIP)
-	if err != nil {
-		return 0, fmt.Errorf("failed to delete the IPv6 routes of peer %s: %w",
-			peerIP, err)
-	}
-
-	err = tx.Commit()
-	if err != nil {
-		return 0, fmt.Errorf("failed to commit transaction: %w", err)
-	}
-
-	d.counts.addRoutes(-int(deletedV4), -int(deletedV6))
-	d.counts.addToDistribution(goneMaskLengthsV4, goneMaskLengthsV6, -1)
-
-	return int(deletedV4 + deletedV6), nil
 }
 
 // insertPrefixesInTable inserts the prefixes, all of one IP version, that the
@@ -2245,9 +2282,25 @@ func (d *Database) Analyze(ctx context.Context) error {
 	return nil
 }
 
-// Checkpoint runs a WAL checkpoint to transfer data from the WAL to the main database.
+// Checkpoint copies the WAL into the database file and keeps the WAL file
-// Uses TRUNCATE mode which blocks writers briefly but ensures complete checkpoint,
+// small. It holds the database lock, so batch writes wait while it runs.
-// keeping the WAL small for fast read performance. It holds the database lock.
+//
+// It runs a PASSIVE checkpoint, which never waits: it copies what no open read
+// still needs. That alone does not keep the file small: SQLite starts the WAL
+// again from the beginning only once a checkpoint has copied all of it and no
+// read that began before is still open, which never happens while reads keep
+// overlapping. So when the file is over sqliteWALSizeLimitBytes, Checkpoint
+// runs a TRUNCATE checkpoint instead, which waits for the reads that still need
+// the WAL, copies all of it and empties the file. That checkpoint waits at most
+// truncateBusyTimeoutMs, and it is logged with how long it took.
+//
+// So while reads overlap, the WAL file stays under sqliteWALSizeLimitBytes
+// plus what batch writes add between two checkpoints, as long as no read stays
+// open longer than that wait. A longer read makes the TRUNCATE checkpoint give
+// up, and the file grows until a later one empties it.
+//
+// It returns ErrCheckpointIncomplete when it leaves part of the WAL uncopied,
+// or a file over the limit untruncated.
 func (d *Database) Checkpoint(ctx context.Context) error {
 	d.lock("Checkpoint")
 	defer d.unlock()
@@ -2257,11 +2310,32 @@ func (d *Database) Checkpoint(ctx context.Context) error {
 		return fmt.Errorf("failed to checkpoint WAL: %w", err)
 	}
 
-	_, err = d.writer.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)")
+	wal, err := os.Stat(d.path + walFileSuffix)
+	if err != nil {
+		return fmt.Errorf("failed to read the WAL file's size: %w", err)
+	}
+
+	if wal.Size() > sqliteWALSizeLimitBytes {
+		return d.truncateWAL(ctx, wal.Size())
+	}
+
+	var busy, walPages, copiedPages int
+
+	err = d.writer.QueryRowContext(ctx, "PRAGMA wal_checkpoint(PASSIVE)").
+		Scan(&busy, &walPages, &copiedPages)
 	if err != nil {
 		return fmt.Errorf("failed to checkpoint WAL: %w", err)
 	}
 
+	if busy != 0 {
+		return fmt.Errorf("%w: SQLite reported it busy", ErrCheckpointIncomplete)
+	}
+
+	if copiedPages < walPages {
+		return fmt.Errorf("%w: copied %d of %d WAL pages",
+			ErrCheckpointIncomplete, copiedPages, walPages)
+	}
+
 	return nil
 }
 
@@ -2277,6 +2351,50 @@ func (d *Database) Ping(ctx context.Context) error {
 	return nil
 }
 
+// truncateWAL runs the TRUNCATE checkpoint of Checkpoint on a WAL file of
+// walBytes, with the writing connection's busy timeout lowered to
+// truncateBusyTimeoutMs for it, and logs how long it took. The caller holds the
+// database lock.
+func (d *Database) truncateWAL(ctx context.Context, walBytes int64) error {
+	_, err := d.writer.ExecContext(ctx,
+		fmt.Sprintf("PRAGMA busy_timeout=%d", truncateBusyTimeoutMs))
+	if err != nil {
+		return fmt.Errorf("failed to lower the busy timeout to truncate WAL: %w",
+			err)
+	}
+
+	started := time.Now()
+
+	var busy, walPages, copiedPages int
+
+	checkpointErr := d.writer.QueryRowContext(ctx,
+		"PRAGMA wal_checkpoint(TRUNCATE)").Scan(&busy, &walPages, &copiedPages)
+	took := time.Since(started)
+
+	// Restored even when ctx has ended, so that no batch write runs with the
+	// short timeout.
+	_, restoreErr := d.writer.ExecContext(context.WithoutCancel(ctx),
+		fmt.Sprintf("PRAGMA busy_timeout=%d", sqliteBusyTimeoutMs))
+	if checkpointErr != nil || restoreErr != nil {
+		return fmt.Errorf("failed to truncate WAL: %w",
+			errors.Join(checkpointErr, restoreErr))
+	}
+
+	if busy != 0 {
+		d.logger.Warn("WAL file over its size limit not truncated",
+			"wal_bytes", walBytes, "wal_pages", walPages,
+			"copied_pages", copiedPages, "duration", took)
+
+		return fmt.Errorf("%w: WAL file of %d bytes not truncated",
+			ErrCheckpointIncomplete, walBytes)
+	}
+
+	d.logger.Info("WAL file over its size limit truncated",
+		"wal_bytes", walBytes, "duration", took)
+
+	return nil
+}
+
 // The debug messages lock logs, with the operation's name, when it starts to
 // wait for the database mutex and when it holds it.
 const (
@@ -2332,6 +2450,46 @@ func (d *Database) rollback(tx *loggingTx) {
 	}
 }
 
+// deletePeerRouteChunk deletes, in one transaction, the first
+// peerRouteChunkSize IPv4 and the first peerRouteChunkSize IPv6 live routes of
+// the peer, or as many as it has. It returns how many of each it deleted.
+func (d *Database) deletePeerRouteChunk(peerIP string) (int, int, error) {
+	d.lock("DeleteLiveRoutesOfPeer")
+	defer d.unlock()
+
+	tx, err := d.beginTx()
+	if err != nil {
+		return 0, 0, fmt.Errorf("failed to begin transaction: %w", err)
+	}
+	defer d.rollback(tx)
+
+	ctx := context.Background()
+
+	deletedV4, goneMaskLengthsV4, err := deletePeerRoutes(ctx, tx,
+		peerRouteChunkV4SQL, deletePeerRouteChunkV4SQL, peerIP)
+	if err != nil {
+		return 0, 0, fmt.Errorf("failed to delete the IPv4 routes of peer %s: %w",
+			peerIP, err)
+	}
+
+	deletedV6, goneMaskLengthsV6, err := deletePeerRoutes(ctx, tx,
+		peerRouteChunkV6SQL, deletePeerRouteChunkV6SQL, peerIP)
+	if err != nil {
+		return 0, 0, fmt.Errorf("failed to delete the IPv6 routes of peer %s: %w",
+			peerIP, err)
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		return 0, 0, fmt.Errorf("failed to commit transaction: %w", err)
+	}
+
+	d.counts.addRoutes(-int(deletedV4), -int(deletedV6))
+	d.counts.addToDistribution(goneMaskLengthsV4, goneMaskLengthsV6, -1)
+
+	return int(deletedV4), int(deletedV6), nil
+}
+
 // routeTimestampRange returns the earliest and latest last_updated across both
 // live route tables, or nil values when both tables are empty. Each query reads
 // one row from an end of the last_updated index rather than scanning the tables.

internal/database/database_internal_test.go

@@ -8,6 +8,7 @@ import (
 	"log/slog"
 	"math/rand/v2"
 	"net/netip"
+	"os"
 	"slices"
 	"strconv"
 	"sync"
@@ -67,6 +68,28 @@ const (
 	lockWaitTimeout = time.Millisecond
 	// lockTestASN is the AS the batch write in the lock tests stores.
 	lockTestASN = 64500
+	// readerCheckpointLimit is how long a checkpoint, and a batch write
+	// waiting for it, may take while a reader is open: well under the busy
+	// timeout, which a checkpoint that waits for the reader takes.
+	readerCheckpointLimit = time.Second
+)
+
+// Parameters of the test of the WAL file's size while reads overlap.
+const (
+	// walFillStep is how much the test writes between two checkpoints.
+	walFillStep = sqliteWALSizeLimitBytes / 16
+	// walFillTotal is how much it writes in all: past sqliteWALSizeLimitBytes,
+	// and on until a WAL file that no checkpoint empties ends over walFillBound.
+	walFillTotal = sqliteWALSizeLimitBytes + 4*walFillStep
+	// walFillBound is the size the WAL file must stay under: the limit, plus
+	// the writes since the last checkpoint, plus one more step for SQLite's
+	// own pages besides the written data.
+	walFillBound = sqliteWALSizeLimitBytes + 2*walFillStep
+	// overlappingReads is how many reads the test keeps open at once.
+	overlappingReads = 2
+	// overlappingReadDuration is how long each read stays open: well under
+	// truncateBusyTimeoutMs, so the TRUNCATE checkpoint can wait for it.
+	overlappingReadDuration = 50 * time.Millisecond
 )
 
 // Parameters for the prefix-length sample tests.
@@ -457,6 +480,33 @@ func TestWritingConnectionCache(t *testing.T) {
 	}
 }
 
+// TestWritingConnectionWALSizeLimit checks that the connection kept for
+// writing, the one that writes the WAL, has the WAL size limit, which cuts the
+// WAL file back when SQLite starts the WAL again from the beginning.
+func TestWritingConnectionWALSizeLimit(t *testing.T) {
+	t.Parallel()
+
+	db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	var limit int64
+
+	err = db.writer.QueryRowContext(t.Context(), "PRAGMA journal_size_limit").
+		Scan(&limit)
+	if err != nil {
+		t.Fatalf("failed to read the writing connection's journal_size_limit: %v",
+			err)
+	}
+
+	if limit != sqliteWALSizeLimitBytes {
+		t.Errorf("writing connection: journal_size_limit = %d, want %d",
+			limit, sqliteWALSizeLimitBytes)
+	}
+}
+
 // readPragma reads the integer value of PRAGMA name on connection i, c.
 func readPragma(
 	ctx context.Context, t *testing.T, c *sql.Conn, i int, name string,
@@ -799,6 +849,229 @@ func TestCheckpointLockWaitEndsWithContext(t *testing.T) {
 	db.mu.Unlock()
 }
 
+// TestCheckpointDoesNotWaitForReader keeps a read transaction on
+// live_routes_v4 open while a route is committed, then starts Checkpoint and a
+// batch write together. It checks that both end within readerCheckpointLimit,
+// whichever of them takes the database lock first, and that Checkpoint reports
+// the part of the WAL the reader still needs as left uncopied. A TRUNCATE
+// checkpoint waited the busy timeout for the reader while holding the lock
+// (https://git.eeqj.de/sneak/routewatch/issues/102).
+func TestCheckpointDoesNotWaitForReader(t *testing.T) {
+	t.Parallel()
+
+	db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	ctx := context.Background()
+	ts := time.Now().UTC()
+
+	err = db.UpsertLiveRouteBatch(
+		[]*LiveRoute{mkV4Route(t, "198.51.100.0/24", lockTestASN, ts)})
+	if err != nil {
+		t.Fatalf("failed to store the route the reader reads: %v", err)
+	}
+
+	reader, err := db.db.Conn(ctx)
+	if err != nil {
+		t.Fatalf("failed to open the reading connection: %v", err)
+	}
+	defer func() { _ = reader.Close() }()
+
+	// A plain BEGIN starts a transaction that only reads, unlike the BEGIN
+	// IMMEDIATE of _txlock. It keeps the snapshot its first read takes until
+	// it ends.
+	_, err = reader.ExecContext(ctx, "BEGIN")
+	if err != nil {
+		t.Fatalf("failed to begin the read transaction: %v", err)
+	}
+	defer func() { _, _ = reader.ExecContext(ctx, "ROLLBACK") }()
+
+	var routes int
+
+	err = reader.QueryRowContext(ctx,
+		"SELECT count(*) FROM live_routes_v4").Scan(&routes)
+	if err != nil {
+		t.Fatalf("failed to read the live routes: %v", err)
+	}
+
+	err = db.UpsertLiveRouteBatch(
+		[]*LiveRoute{mkV4Route(t, "203.0.113.0/24", lockTestASN, ts)})
+	if err != nil {
+		t.Fatalf("failed to store a route after the read: %v", err)
+	}
+
+	started := time.Now()
+	checkpointed := make(chan error, 1)
+
+	go func() { checkpointed <- db.Checkpoint(ctx) }()
+
+	// A route write, not an AS write: the first new AS loads the embedded AS
+	// names, which takes seconds under -race.
+	route := mkV4Route(t, "192.0.2.0/24", lockTestASN, ts)
+	written := make(chan error, 1)
+
+	go func() { written <- db.UpsertLiveRouteBatch([]*LiveRoute{route}) }()
+
+	err = <-checkpointed
+	if !errors.Is(err, ErrCheckpointIncomplete) {
+		t.Errorf("Checkpoint = %v with a reader open, want %v",
+			err, ErrCheckpointIncomplete)
+	}
+
+	took := time.Since(started)
+	if took >= readerCheckpointLimit {
+		t.Errorf("Checkpoint took %v with a reader open, want under %v",
+			took, readerCheckpointLimit)
+	}
+
+	err = <-written
+	if err != nil {
+		t.Errorf("batch write during Checkpoint: %v", err)
+	}
+
+	took = time.Since(started)
+	if took >= readerCheckpointLimit {
+		t.Errorf("batch write during Checkpoint ended after %v, want under %v",
+			took, readerCheckpointLimit)
+	}
+}
+
+// TestCheckpointKeepsWALSmallWhileReadsOverlap writes walFillTotal while
+// overlappingReads reads keep overlapping, so that one is open at every write,
+// and runs Checkpoint after every walFillStep, as the maintainer does between
+// batch writes. It checks that the WAL file stays under walFillBound and that
+// the writing connection has its busy timeout back afterwards. With only a
+// PASSIVE checkpoint, SQLite never started the WAL again from the beginning
+// while reads overlapped, and the file grew for as long as they did
+// (https://git.eeqj.de/sneak/routewatch/issues/102).
+func TestCheckpointKeepsWALSmallWhileReadsOverlap(t *testing.T) {
+	t.Parallel()
+
+	db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	ctx := t.Context()
+
+	// The test writes to a table of its own: each write replaces the one
+	// row's data with walFillStep of new random bytes, so the WAL grows by that
+	// much while the database file stays small. SQLite writes no page whose
+	// content stays the same, so the bytes must change. Like batch writes,
+	// the writes use the writing connection between the checkpoints.
+	_, err = db.writer.ExecContext(ctx,
+		"CREATE TABLE wal_fill (data BLOB); INSERT INTO wal_fill VALUES (NULL)")
+	if err != nil {
+		t.Fatalf("failed to create the table the test writes to: %v", err)
+	}
+
+	stop := make(chan struct{})
+
+	var readers sync.WaitGroup
+
+	for i := range overlappingReads {
+		readers.Go(func() {
+			// Reader i starts i/overlappingReads of a read after the first,
+			// so that the reads overlap.
+			time.Sleep(time.Duration(i) * overlappingReadDuration /
+				overlappingReads)
+			keepReading(ctx, t, db, stop)
+		})
+	}
+
+	defer func() {
+		close(stop)
+		readers.Wait()
+	}()
+
+	for written := 0; written < walFillTotal; written += walFillStep {
+		_, err = db.writer.ExecContext(ctx,
+			"UPDATE wal_fill SET data = randomblob(?)", walFillStep)
+		if err != nil {
+			t.Fatalf("failed to write: %v", err)
+		}
+
+		wal, statErr := os.Stat(db.path + walFileSuffix)
+		if statErr != nil {
+			t.Fatalf("failed to read the WAL file's size: %v", statErr)
+		}
+
+		if wal.Size() > walFillBound {
+			t.Fatalf("WAL file of %d bytes after writing %d with reads "+
+				"overlapping, want at most %d", wal.Size(), written+walFillStep,
+				walFillBound)
+		}
+
+		err = db.Checkpoint(ctx)
+		if err != nil && !errors.Is(err, ErrCheckpointIncomplete) {
+			t.Fatalf("Checkpoint: %v", err)
+		}
+	}
+
+	busyTimeout := readPragma(ctx, t, db.writer, 0, "busy_timeout")
+	if busyTimeout != sqliteBusyTimeoutMs {
+		t.Errorf("writing connection: busy_timeout = %d after the checkpoints, "+
+			"want %d", busyTimeout, sqliteBusyTimeoutMs)
+	}
+}
+
+// keepReading runs one read transaction after another on a connection of its
+// own, each open for overlappingReadDuration, until stop is closed.
+func keepReading(
+	ctx context.Context, t *testing.T, db *Database, stop <-chan struct{},
+) {
+	t.Helper()
+
+	conn, err := db.db.Conn(ctx)
+	if err != nil {
+		t.Errorf("failed to open a reading connection: %v", err)
+
+		return
+	}
+	defer func() { _ = conn.Close() }()
+
+	for {
+		select {
+		case <-stop:
+			return
+		default:
+		}
+
+		// A plain BEGIN starts a transaction that only reads, unlike the
+		// BEGIN IMMEDIATE of _txlock. It keeps the snapshot its first read
+		// takes until it ends.
+		_, err = conn.ExecContext(ctx, "BEGIN")
+		if err != nil {
+			t.Errorf("failed to begin a read transaction: %v", err)
+
+			return
+		}
+
+		var rows int
+
+		err = conn.QueryRowContext(ctx, "SELECT count(*) FROM wal_fill").
+			Scan(&rows)
+		if err != nil {
+			t.Errorf("failed to read: %v", err)
+
+			return
+		}
+
+		time.Sleep(overlappingReadDuration)
+
+		_, err = conn.ExecContext(ctx, "ROLLBACK")
+		if err != nil {
+			t.Errorf("failed to end a read transaction: %v", err)
+
+			return
+		}
+	}
+}
+
 // TestGetRandomPrefixesByLengthListsPrefixOnce stores one prefix of each
 // family learned from two peers in the same second and checks that the
 // prefix-length sample lists each of them once.

internal/routewatch/asnfetcher.go

@@ -264,7 +264,10 @@ func (f *ASNFetcher) fetchNextStale(ctx context.Context) {
 }
 
 // fetchAndUpdate performs a WHOIS lookup and updates the database.
-// Returns true if successful.
+// Returns true if successful. A failed lookup leaves the AS's row as it was,
+// 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.
 func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool {
 	ctx, cancel := context.WithTimeout(ctx, time.Minute)
 	defer cancel()
@@ -272,6 +275,19 @@ func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool {
 	f.logger.Info("Fetching WHOIS data", "asn", asn)
 
 	info, err := f.whoisClient.LookupASN(ctx, asn)
+	if errors.Is(err, whois.ErrNoReferral) {
+		f.logger.Info("IANA refers the AS to no RIR", "asn", asn)
+
+		err = f.db.UpdateASNWHOIS(ctx, &database.ASNWHOISUpdate{ASN: asn})
+		if err != nil {
+			f.logger.Error("Failed to update ASN WHOIS data",
+				"asn", asn, "error", err)
+			f.recordFailure()
+		}
+
+		return false
+	}
+
 	if err != nil {
 		f.logger.Error("WHOIS lookup failed", "asn", asn, "error", err)
 		f.recordFailure()

internal/routewatch/asnfetcher_internal_test.go

@@ -0,0 +1,157 @@
+package routewatch
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"io"
+	"net"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/database"
+	"git.eeqj.de/sneak/routewatch/internal/logger"
+)
+
+// The WHOIS servers a lookup asks: IANA first, then the RIR IANA refers it
+// to.
+const (
+	ianaWHOISServer = "whois.iana.org:43"
+	ripeWHOISServer = "whois.ripe.net:43"
+)
+
+// ianaRefersToRIPE is IANA's answer for an AS in a block it gave to RIPE.
+const ianaRefersToRIPE = `refer:        whois.ripe.net
+
+as-block:     64496-64511
+organisation: Assigned by RIPE NCC
+changed:      2002-09-01
+source:       IANA
+`
+
+// ripeAccessDenied is RIPE's answer to a host it no longer serves.
+const ripeAccessDenied = `% This is the RIPE Database query service.
+%ERROR:201: access denied for 192.0.2.1
+`
+
+// ianaPrivateUseBlock is IANA's answer for a private-use AS: its block and
+// no referral.
+const ianaPrivateUseBlock = `as-block:     64512-65534
+organisation: Reserved for Private Use
+source:       IANA
+`
+
+// newFetcherWithWHOISAnswers returns an ASN fetcher on a new database that
+// holds testASNA, never looked up. The fetcher's WHOIS client connects to a
+// fake server for each address in answers, which reads the query and sends
+// answers[address]; connecting to any other address is refused.
+func newFetcherWithWHOISAnswers(
+	t *testing.T, answers map[string]string,
+) (*ASNFetcher, *database.Database) {
+	t.Helper()
+
+	db := newTestDatabase(t)
+
+	_, err := db.GetOrCreateASN(testASNA, time.Now())
+	if err != nil {
+		t.Fatalf("GetOrCreateASN: %v", err)
+	}
+
+	var servers sync.WaitGroup
+
+	t.Cleanup(servers.Wait)
+
+	f := NewASNFetcher(db, logger.New().Logger)
+	f.whoisClient.DialContext = func(
+		_ context.Context, _, address string,
+	) (net.Conn, error) {
+		answer, ok := answers[address]
+		if !ok {
+			return nil, syscall.ECONNREFUSED
+		}
+
+		client, server := net.Pipe()
+
+		servers.Go(func() {
+			defer func() { _ = server.Close() }()
+
+			_, _ = bufio.NewReader(server).ReadString('\n')
+			_, _ = io.WriteString(server, answer)
+		})
+
+		return client, nil
+	}
+
+	return f, db
+}
+
+// TestFetchAndUpdateKeepsTheRowWhenTheLookupFails looks up an AS IANA refers
+// to RIPE when RIPE's server refuses the connection, and when it answers only
+// with an error. The AS must stay never looked up, and the fetcher must back
+// off.
+func TestFetchAndUpdateKeepsTheRowWhenTheLookupFails(t *testing.T) {
+	t.Parallel()
+
+	for name, answers := range map[string]map[string]string{
+		"connection refused": {ianaWHOISServer: ianaRefersToRIPE},
+		"only an error": {
+			ianaWHOISServer: ianaRefersToRIPE,
+			ripeWHOISServer: ripeAccessDenied,
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			f, db := newFetcherWithWHOISAnswers(t, answers)
+
+			if f.fetchAndUpdate(t.Context(), testASNA) {
+				t.Error("fetchAndUpdate reported success")
+			}
+
+			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: the failed lookup "+
+					"was stored", stats.NeverFetched)
+			}
+
+			got := f.GetStats()
+			if got.CurrentInterval != baseInterval*backoffMultiplier ||
+				got.ConsecutiveFails != 1 {
+				t.Errorf("interval %v after %d failures, want %v after 1",
+					got.CurrentInterval, got.ConsecutiveFails,
+					baseInterval*backoffMultiplier)
+			}
+		})
+	}
+}
+
+// TestFetchAndUpdateMarksAnASWithoutReferral looks up a private-use AS, which
+// IANA refers to no RIR. The AS must not be looked up again until it is
+// stale, and the lookup counts as neither a success nor a failure.
+func TestFetchAndUpdateMarksAnASWithoutReferral(t *testing.T) {
+	t.Parallel()
+
+	f, db := newFetcherWithWHOISAnswers(t,
+		map[string]string{ianaWHOISServer: ianaPrivateUseBlock})
+
+	f.fetchAndUpdate(t.Context(), testASNA)
+
+	asn, err := db.GetNextStaleASN(t.Context(), whoisStaleThreshold)
+	if !errors.Is(err, database.ErrNoStaleASN) {
+		t.Errorf("GetNextStaleASN = %d, %v; want ErrNoStaleASN", asn, err)
+	}
+
+	got := f.GetStats()
+	if got.CurrentInterval != baseInterval || got.SuccessesLastHour != 0 ||
+		got.ErrorsLastHour != 0 {
+		t.Errorf("interval %v, %d successes, %d errors; want %v, 0, 0",
+			got.CurrentInterval, got.SuccessesLastHour, got.ErrorsLastHour,
+			baseInterval)
+	}
+}

internal/routewatch/dbmaintainer.go

@@ -2,6 +2,7 @@ package routewatch
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"sync"
 	"time"
@@ -34,6 +35,12 @@ const (
 	analyzeTimeout = 5 * time.Minute
 )
 
+// checkpointIncompleteMessage is logged, at debug level like a completed
+// checkpoint, when a checkpoint leaves part of the WAL for a later one: while
+// reads overlap, most do. The database warns when the WAL file stays over its
+// size limit.
+const checkpointIncompleteMessage = "WAL checkpoint incomplete"
+
 // DBMaintainer handles background database maintenance tasks.
 type DBMaintainer struct {
 	db     database.Store
@@ -123,9 +130,13 @@ func (m *DBMaintainer) runCheckpoint(ctx context.Context) {
 	duration, err := runWithTimeout(ctx, checkpointTimeout, m.db.Checkpoint)
 	m.recordResult(&m.lastCheckpoint, &m.lastCheckpointError, &m.checkpointCount, err)
 
-	if err != nil {
+	switch {
+	case errors.Is(err, database.ErrCheckpointIncomplete):
+		m.logger.Debug(checkpointIncompleteMessage,
+			"error", err, "duration", duration)
+	case err != nil:
 		m.logger.Error("WAL checkpoint failed", "error", err, "duration", duration)
-	} else {
+	default:
 		m.logger.Debug("WAL checkpoint completed", "duration", duration)
 	}
 }

internal/routewatch/dbmaintainer_internal_test.go

@@ -0,0 +1,54 @@
+package routewatch
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"log/slog"
+	"testing"
+
+	"git.eeqj.de/sneak/routewatch/internal/database"
+)
+
+// incompleteCheckpointStore is a store whose every checkpoint leaves part of
+// the WAL uncopied.
+type incompleteCheckpointStore struct{ *mockStore }
+
+// Checkpoint reports the checkpoint incomplete.
+func (incompleteCheckpointStore) Checkpoint(context.Context) error {
+	return database.ErrCheckpointIncomplete
+}
+
+// TestIncompleteCheckpointNotLoggedAsCompleted runs a checkpoint that leaves
+// part of the WAL uncopied and checks that the maintainer logs it as
+// incomplete, not as completed, and does not count it as a checkpoint that
+// completed.
+func TestIncompleteCheckpointNotLoggedAsCompleted(t *testing.T) {
+	t.Parallel()
+
+	var logs bytes.Buffer
+
+	logger := slog.New(slog.NewJSONHandler(&logs,
+		&slog.HandlerOptions{Level: slog.LevelDebug}))
+	m := NewDBMaintainer(incompleteCheckpointStore{newMockStore()}, logger)
+
+	m.runCheckpoint(t.Context())
+
+	var logged struct {
+		Msg string `json:"msg"`
+	}
+
+	err := json.Unmarshal(logs.Bytes(), &logged)
+	if err != nil {
+		t.Fatalf("want one log message, got %q: %v", logs.String(), err)
+	}
+
+	if logged.Msg != checkpointIncompleteMessage {
+		t.Errorf("logged %q, want %q", logged.Msg, checkpointIncompleteMessage)
+	}
+
+	if m.checkpointCount != 0 {
+		t.Errorf("checkpointCount = %d after an incomplete checkpoint, want 0",
+			m.checkpointCount)
+	}
+}

internal/routewatch/prefixhandler.go

@@ -28,7 +28,9 @@ const (
 	// prefixBatchMaxSize is the most prefix updates the batch holds while the
 	// flush goroutine writes the previous one: ten batches, about 60 MiB
 	// (README "Memory"). At this size HandleMessage waits until the flush
-	// goroutine takes the batch.
+	// goroutine takes the batch. On top of these, the handler holds at most
+	// prefixBatchSize updates, about 6 MiB, that a flush kept for the peers
+	// whose route delete failed (held).
 	prefixBatchMaxSize = 250000
 
 	// prefixBatchTimeout is the maximum time to wait before flushing a batch
@@ -48,13 +50,21 @@ type PrefixHandler struct {
 	logger  *logger.Logger
 	metrics *metrics.Tracker
 
-	// Batching. mu guards batch and peersDown only: the flush goroutine takes
+	// Batching. mu guards batch, held and peersDown only: the flush goroutine
-	// both under mu and writes them after releasing mu, so HandleMessage
+	// takes them under mu and writes them after releasing mu, so
-	// waits for a database write only while batch is at prefixBatchMaxSize.
+	// HandleMessage waits for a database write only while batch is at
+	// prefixBatchMaxSize.
 	mu    sync.Mutex
 	batch []prefixUpdate
-	// peersDown holds the peers whose BGP session went down since the batch
+	// held holds the updates that the peers whose route delete failed in the
-	// was taken.
+	// last flush sent after their session went down, for the next flush to
+	// write after it retries the delete. They are not in batch, so they do
+	// not count toward a full batch: a delete that keeps failing is retried
+	// at the pace of the flush timer.
+	held []prefixUpdate
+	// peersDown holds, each once, the peers whose BGP session went down
+	// since the batch was taken, and those whose route delete failed in the
+	// last flush.
 	peersDown []string
 	// batchTaken is signalled when the flush goroutine takes batch; its L is
 	// &mu.
@@ -200,17 +210,21 @@ func (h *PrefixHandler) Stop() {
 
 // peerSessionDown notes that the peer's BGP session went down, so that the next
 // flush deletes all of the peer's live routes before it writes the batch. The
-// peer's updates still in the batch came before the down, which replaces them,
+// peer's updates still in the batch or held came before the down, which
-// so they are dropped; updates the peer sends later stay in the batch and are
+// replaces them, so they are dropped; updates the peer sends later stay in the
-// written after the delete.
+// batch and are written after the delete. A peer already in peersDown is not
+// added again: one delete removes all of its routes.
 func (h *PrefixHandler) peerSessionDown(peer string) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
-	h.batch = slices.DeleteFunc(h.batch, func(update prefixUpdate) bool {
+	isPeers := func(update prefixUpdate) bool { return update.peer == peer }
-		return update.peer == peer
+	h.batch = slices.DeleteFunc(h.batch, isPeers)
-	})
+	h.held = slices.DeleteFunc(h.held, isPeers)
-	h.peersDown = append(h.peersDown, peer)
+
+	if !slices.Contains(h.peersDown, peer) {
+		h.peersDown = append(h.peersDown, peer)
+	}
 }
 
 // flushLoop runs in a goroutine, the only one that writes the batch: it
@@ -238,18 +252,23 @@ func (h *PrefixHandler) flushLoop() {
 	}
 }
 
-// takeBatch returns the pending updates and the peers whose session went down
+// takeBatch returns the pending updates, the held ones first, and the peers in
-// since the last flush, and leaves an empty batch and no peers in their place
+// peersDown, and leaves an empty batch, no held updates and no peers in their
-// for HandleMessage to fill.
+// place for HandleMessage to fill.
 func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
-	if len(h.batch) == 0 && len(h.peersDown) == 0 {
+	if len(h.held) == 0 && len(h.batch) == 0 && len(h.peersDown) == 0 {
 		return nil, nil
 	}
 
 	batch, peersDown := h.batch, h.peersDown
+	if len(h.held) > 0 {
+		batch = slices.Concat(h.held, h.batch)
+	}
+
+	h.held = nil
 	h.batch = make([]prefixUpdate, 0, prefixBatchSize)
 	h.peersDown = nil
 	h.batchTaken.Broadcast()
@@ -269,8 +288,12 @@ func (h *PrefixHandler) flushBatch() {
 	batchSize := len(batch)
 
 	// The routes of a peer whose session went down are deleted first: the
-	// batch holds only the updates the peer sent after the down.
+	// batch holds only the updates the peer sent after the down. A peer whose
-	h.deleteLiveRoutesOfPeers(peersDown)
+	// delete failed is kept, with those updates, for the next flush to retry.
+	failedPeers := h.deleteLiveRoutesOfPeers(peersDown)
+	if len(failedPeers) > 0 {
+		batch = h.keepForNextFlush(batch, failedPeers)
+	}
 
 	// For each prefix and peer, keep only the last update the feed delivered.
 	// Updates from different peers for the same prefix are all kept. Arrival
@@ -344,15 +367,21 @@ func (h *PrefixHandler) flushBatch() {
 }
 
 // deleteLiveRoutesOfPeers deletes all live routes of each of peers, the peers
-// whose BGP session went down. Most downs come from peers that have no routes,
+// whose BGP session went down, and returns those whose delete failed. Most
-// so only a delete that removed routes is logged.
+// downs come from peers that have no routes, so only a delete that removed
-func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) {
+// routes is logged.
+func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) []string {
+	var failed []string
+
 	for _, peer := range peers {
 		deleted, err := h.db.DeleteLiveRoutesOfPeer(peer)
 		if err != nil {
 			h.logger.Error(
-				"Failed to delete the live routes of a peer whose session went down",
+				"Failed to delete the live routes of a peer whose session "+
-				"peer", peer, "error", err)
+					"went down; the next flush retries",
+				"peer", peer, "routes_deleted", deleted, "error", err)
+
+			failed = append(failed, peer)
 
 			continue
 		}
@@ -363,6 +392,63 @@ func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) {
 				"peer", peer, "routes", deleted)
 		}
 	}
+
+	return failed
+}
+
+// keepForNextFlush puts the peers whose route delete failed back in peersDown,
+// and their updates from batch in held, so that the next flush retries each
+// delete before it writes the peer's updates; written now, they would be
+// deleted by the retry. The updates of a peer whose session went down again
+// since the batch was taken came before that down, so they are dropped, as
+// peerSessionDown drops them, and the peer is already in peersDown. It returns
+// the updates of the other peers, for this flush to write.
+//
+// While a delete keeps failing, the held updates would grow with every flush,
+// so only the last prefixBatchSize of them are kept and the older ones are
+// dropped. This never leaves in place a route the peer withdrew: a prefix with
+// a kept update keeps its last one, and the retry deletes all of the peer's
+// routes, so a prefix whose updates were all dropped has no route from the peer
+// until the peer announces it again.
+func (h *PrefixHandler) keepForNextFlush(
+	batch []prefixUpdate, failedPeers []string,
+) []prefixUpdate {
+	var kept, toWrite []prefixUpdate
+
+	for _, update := range batch {
+		if slices.Contains(failedPeers, update.peer) {
+			kept = append(kept, update)
+		} else {
+			toWrite = append(toWrite, update)
+		}
+	}
+
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	kept = slices.DeleteFunc(kept, func(update prefixUpdate) bool {
+		return slices.Contains(h.peersDown, update.peer)
+	})
+
+	if len(kept) > prefixBatchSize {
+		h.logger.Error(
+			"Dropped the oldest updates of peers whose route delete failed",
+			"peers", failedPeers, "dropped", len(kept)-prefixBatchSize)
+
+		// A copy, so that the dropped updates are freed.
+		kept = slices.Clone(kept[len(kept)-prefixBatchSize:])
+	}
+
+	// takeBatch left held empty, and only this flush fills it.
+	h.held = kept
+
+	for _, peer := range failedPeers {
+		if !slices.Contains(h.peersDown, peer) {
+			h.peersDown = append(h.peersDown, peer)
+		}
+	}
+
+	return toWrite
 }
 
 // writeBatch writes the collected route upserts, route deletions and prefix

internal/routewatch/prefixhandler_internal_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/netip"
 	"slices"
 	"sync"
 	"testing"
@@ -560,6 +561,211 @@ func TestPrefixHandlerHandlesADownWhileAFlushWrites(t *testing.T) {
 	})
 }
 
+// errTestPeerDelete is the error a test store's failing delete of a peer's
+// routes returns.
+var errTestPeerDelete = errors.New("test: the delete of a peer's routes failed")
+
+// firstPeerDeleteFailsStore is a real database whose first delete of a peer's
+// live routes fails, deleting nothing. deletes counts the deletes asked for.
+type firstPeerDeleteFailsStore struct {
+	*database.Database
+
+	// whileFailing, if not nil, runs during the delete that fails.
+	whileFailing func()
+	deletes      int
+}
+
+// DeleteLiveRoutesOfPeer fails the first call and deletes the peer's routes on
+// every later one.
+func (s *firstPeerDeleteFailsStore) DeleteLiveRoutesOfPeer(
+	peerIP string,
+) (int, error) {
+	s.deletes++
+	if s.deletes > 1 {
+		return s.Database.DeleteLiveRoutesOfPeer(peerIP)
+	}
+
+	if s.whileFailing != nil {
+		s.whileFailing()
+	}
+
+	return 0, errTestPeerDelete
+}
+
+// TestPrefixHandlerRetriesAFailedDeleteOfAPeersRoutes runs, through the prefix
+// handler into a database whose first delete of a peer's routes fails, one flush
+// after another: a peer's announcement, then its session going down and a new
+// announcement from the peer, then nothing. The failed delete must be retried
+// on the next flush, and the announcement the peer sent after the down written
+// only after that retry, so it is kept. In the other cases the peer's session
+// goes down again, and the peer announces another prefix, while the delete
+// fails or after the flush whose delete failed: the announcement sent between
+// the two downs must not be written, and the next flush must delete the peer's
+// routes once, not once for the retry and once for the down.
+func TestPrefixHandlerRetriesAFailedDeleteOfAPeersRoutes(t *testing.T) {
+	t.Parallel()
+
+	const (
+		beforeDown      = "192.0.2.0/24"
+		afterDown       = "198.51.100.0/24"
+		afterSecondDown = "203.0.113.0/24"
+		maskLength      = 24
+		// The delete that fails and its retry, which is also the delete for
+		// a second down.
+		wantDeletes = 2
+	)
+
+	tests := []struct {
+		name string
+		// Handled while the delete fails, if not nil.
+		whileFailing []*ristypes.RISMessage
+		// Handled after the flush whose delete fails, if not nil.
+		afterFailing []*ristypes.RISMessage
+		// The peers left with a live route for each prefix.
+		wantPeers map[string][]string
+	}{
+		{
+			name: "the delete fails once",
+			wantPeers: map[string][]string{
+				beforeDown: nil, afterDown: {testPeerIP}, afterSecondDown: nil,
+			},
+		},
+		{
+			name: "the session goes down again while the delete fails",
+			whileFailing: []*ristypes.RISMessage{
+				sessionDown(), peerAnnouncement(testPeerIP, afterSecondDown),
+			},
+			wantPeers: map[string][]string{
+				beforeDown: nil, afterDown: nil, afterSecondDown: {testPeerIP},
+			},
+		},
+		{
+			name: "the session goes down again after the delete failed",
+			afterFailing: []*ristypes.RISMessage{
+				sessionDown(), peerAnnouncement(testPeerIP, afterSecondDown),
+			},
+			wantPeers: map[string][]string{
+				beforeDown: nil, afterDown: nil, afterSecondDown: {testPeerIP},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			db := newTestDatabase(t)
+			store := &firstPeerDeleteFailsStore{Database: db}
+
+			// Built without NewPrefixHandler's flush timer, so the test
+			// flushes itself.
+			h := &PrefixHandler{db: store, logger: logger.New()}
+			store.whileFailing = func() {
+				for _, msg := range tt.whileFailing {
+					h.HandleMessage(msg)
+				}
+			}
+
+			h.HandleMessage(peerAnnouncement(testPeerIP, beforeDown))
+			h.flushBatch()
+			h.HandleMessage(sessionDown())
+			h.HandleMessage(peerAnnouncement(testPeerIP, afterDown))
+			h.flushBatch() // the delete fails
+
+			for _, msg := range tt.afterFailing {
+				h.HandleMessage(msg)
+			}
+
+			h.flushBatch()
+
+			if store.deletes != wantDeletes {
+				t.Errorf("the store was asked for %d deletes of the peer's "+
+					"routes, want %d", store.deletes, wantDeletes)
+			}
+
+			assertPeers(t, db, tt.wantPeers)
+			assertCounts(t, db, ipv4Version, 1, []database.PrefixDistribution{
+				{MaskLength: maskLength, Count: 1},
+			})
+		})
+	}
+}
+
+// peerDeleteAlwaysFailsStore is a real database whose every delete of a peer's
+// live routes fails, deleting nothing.
+type peerDeleteAlwaysFailsStore struct {
+	*database.Database
+}
+
+// DeleteLiveRoutesOfPeer fails.
+func (peerDeleteAlwaysFailsStore) DeleteLiveRoutesOfPeer(string) (int, error) {
+	return 0, errTestPeerDelete
+}
+
+// TestPrefixHandlerKeepsAtMostABatchOfUpdatesWhileADeleteKeepsFailing runs,
+// through the prefix handler into a database whose every delete of a peer's
+// routes fails, a peer's session going down and then three flushes. Before each
+// flush the peer announces prefixBatchSize new prefixes and another peer
+// announces one. After each flush the handler must hold only the peer's last
+// prefixBatchSize updates, not every update since the down, and the other
+// peer's route must be written. After the last flush, one more update must not
+// wake the flush goroutine: held updates do not count toward a full batch.
+func TestPrefixHandlerKeepsAtMostABatchOfUpdatesWhileADeleteKeepsFailing(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	const otherPeerIP = "2001:db8:ffff::2"
+
+	// One prefix the other peer announces before each flush.
+	otherPeersPrefixes := []string{testV4Prefix, testV6Prefix, "2001:db8:2::/48"}
+
+	db := newTestDatabase(t)
+
+	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
+	h := &PrefixHandler{
+		db: peerDeleteAlwaysFailsStore{Database: db}, logger: logger.New(),
+	}
+	h.HandleMessage(sessionDown())
+
+	// The peer announces a /32 for each address from 10.0.0.0 on, each
+	// address once.
+	addr := netip.MustParseAddr("10.0.0.0")
+
+	for flush, otherPeersPrefix := range otherPeersPrefixes {
+		prefixes := make([]string, prefixBatchSize)
+		for i := range prefixes {
+			prefixes[i] = netip.PrefixFrom(addr, addr.BitLen()).String()
+			addr = addr.Next()
+		}
+
+		h.HandleMessage(peerAnnouncement(testPeerIP, prefixes...))
+		h.HandleMessage(peerAnnouncement(otherPeerIP, otherPeersPrefix))
+		h.flushBatch()
+
+		kept := make([]string, 0, len(h.held))
+		for _, update := range h.held {
+			kept = append(kept, update.prefix)
+		}
+
+		if !slices.Equal(kept, prefixes) {
+			t.Fatalf("after flush %d the handler holds %d updates, not just "+
+				"the peer's last %d", flush+1, len(kept), len(prefixes))
+		}
+
+		assertPeers(t, db, map[string][]string{otherPeersPrefix: {otherPeerIP}})
+	}
+
+	// A new channel, so that a wake-up in it comes from the update below.
+	h.batchFullCh = make(chan struct{}, 1)
+	h.HandleMessage(peerAnnouncement(otherPeerIP, testV4Prefix))
+
+	if len(h.batchFullCh) > 0 {
+		t.Error("one update after a flush that held a batch of updates " +
+			"woke the flush goroutine")
+	}
+}
+
 // newTestDatabase opens a database in a new temporary directory. It is closed
 // when the test ends.
 func newTestDatabase(t *testing.T) *database.Database {

internal/whois/whois.go

@@ -4,7 +4,9 @@ package whois
 import (
 	"bufio"
 	"context"
+	"errors"
 	"fmt"
+	"io"
 	"net"
 	"regexp"
 	"strings"
@@ -18,6 +20,10 @@ const (
 	writeTimeout = 5 * time.Second
 )
 
+// maxAnswerSize caps the bytes read of one WHOIS answer: a real one is a
+// few KiB, so one this large is from a server that does not stop sending.
+const maxAnswerSize = 1 << 20
+
 // Parsing constants.
 const (
 	keyValueParts       = 2 // Expected parts when splitting "key: value"
@@ -43,6 +49,19 @@ const (
 	RIRAFRNIC = "AFRINIC"
 )
 
+// Errors LookupASN returns when a WHOIS answer has nothing to store.
+var (
+	// ErrNoReferral is returned when IANA refers the AS to no RIR, as it
+	// does for a reserved, private-use or unallocated AS number.
+	ErrNoReferral = errors.New("IANA refers the AS to no RIR")
+
+	errNoASBlock = errors.New("IANA answered with neither a referral " +
+		"nor an AS block")
+	errNoFields       = errors.New("RIR answered with no WHOIS fields")
+	errUnknownServer  = errors.New("IANA refers the AS to a server no RIR runs")
+	errAnswerTooLarge = errors.New("WHOIS answer reached the size cap")
+)
+
 // ASNInfo contains parsed WHOIS information for an ASN.
 type ASNInfo struct {
 	ASN         int
@@ -63,20 +82,25 @@ type ASNInfo struct {
 
 // Client performs WHOIS lookups for ASNs.
 type Client struct {
-	// Dialer for creating connections (can be overridden for testing)
+	// DialContext opens the connection to a WHOIS server, given as
-	dialer *net.Dialer
+	// host:port. NewClient sets a net.Dialer's; tests replace it to reach
+	// local servers instead.
+	DialContext func(ctx context.Context, network, address string) (net.Conn, error)
 }
 
 // NewClient creates a new WHOIS client.
 func NewClient() *Client {
-	return &Client{
+	dialer := &net.Dialer{Timeout: dialTimeout}
-		dialer: &net.Dialer{
+
-			Timeout: dialTimeout,
+	return &Client{DialContext: dialer.DialContext}
-		},
-	}
 }
 
-// LookupASN queries WHOIS for the given ASN and returns parsed information.
+// LookupASN asks IANA which RIR holds the given ASN, asks that RIR, and
+// returns what the RIR's answer says. It returns ErrNoReferral when IANA
+// refers the AS to no RIR, and another error when a query fails, 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 RIR's answer carries
+// none of ASNInfo's fields, as a refusal of service does.
 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)
@@ -87,34 +111,53 @@ func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
 	}
 
 	// Determine RIR from IANA response
-	rir, whoisServer := c.parseIANAReferral(ianaResp)
+	rir, whoisServer, err := c.parseIANAReferral(ianaResp)
+	if err != nil {
+		return nil, fmt.Errorf("%s: %w", query, err)
+	}
+
 	if whoisServer == "" {
-		// No referral, try to parse what we have
+		// IANA names the block of an AS it refers to no RIR. An answer
-		return c.parseResponse(asn, rir, ianaResp), nil
+		// without that, such as an empty one, says nothing about the AS.
+		if !strings.Contains(ianaResp, "as-block:") {
+			return nil, fmt.Errorf("%s: %w", query, errNoASBlock)
+		}
+
+		return nil, fmt.Errorf("%s: %w", query, ErrNoReferral)
 	}
 
 	// Query the authoritative RIR
 	rirResp, err := c.query(ctx, whoisServer, query)
 	if err != nil {
-		// Return partial data from IANA if RIR query fails
+		return nil, fmt.Errorf("RIR query failed: %w", err)
-		info := c.parseResponse(asn, rir, ianaResp)
-		info.RawResponse = fmt.Sprintf("%s\n--- RIR query failed: %v ---\n",
-			ianaResp, err)
-
-		return info, nil
 	}
 
-	// Combine responses and parse
+	// The fields come from the RIR's answer alone: IANA's describes the
-	fullResponse := ianaResp + "\n" + rirResp
+	// block the AS is in.
-	info := c.parseResponse(asn, rir, fullResponse)
+	info := c.parseResponse(asn, rir, rirResp)
-	info.RawResponse = fullResponse
+	if info.isEmpty() {
+		return nil, fmt.Errorf("%s from %s: %w", query, whoisServer, errNoFields)
+	}
+
+	info.RawResponse = ianaResp + "\n" + rirResp
 
 	return info, nil
 }
 
-// query performs a raw WHOIS query to the specified server.
+// isEmpty reports whether info holds none of the fields a WHOIS answer
+// fills, as when the answer is only comment and error lines.
+func (info *ASNInfo) isEmpty() bool {
+	return info.ASName == "" && info.OrgName == "" && info.OrgID == "" &&
+		info.Address == "" && info.CountryCode == "" &&
+		info.AbuseEmail == "" && info.AbusePhone == "" &&
+		info.TechEmail == "" && info.TechPhone == "" &&
+		info.RegDate == nil && info.LastMod == nil
+}
+
+// query performs a raw WHOIS query to the specified server. An answer that
+// reaches maxAnswerSize is an error.
 func (c *Client) query(ctx context.Context, server, query string) (string, error) {
-	conn, err := c.dialer.DialContext(ctx, "tcp", server)
+	conn, err := c.DialContext(ctx, "tcp", server)
 	if err != nil {
 		return "", fmt.Errorf("dial %s: %w", server, err)
 	}
@@ -140,7 +183,11 @@ func (c *Client) query(ctx context.Context, server, query string) (string, error
 
 	var sb strings.Builder
 
-	scanner := bufio.NewScanner(conn)
+	// The reader ends at the cap; N, the bytes it has left, is 0 once the
+	// answer has reached it.
+	answer := &io.LimitedReader{R: conn, N: maxAnswerSize}
+
+	scanner := bufio.NewScanner(answer)
 	for scanner.Scan() {
 		sb.WriteString(scanner.Text())
 		sb.WriteString("\n")
@@ -151,13 +198,18 @@ func (c *Client) query(ctx context.Context, server, query string) (string, error
 		return sb.String(), fmt.Errorf("read response: %w", err)
 	}
 
+	if answer.N == 0 {
+		return "", fmt.Errorf("read response: %w", errAnswerTooLarge)
+	}
+
 	return sb.String(), nil
 }
 
 // parseIANAReferral extracts the RIR and WHOIS server from an IANA response.
-func (c *Client) parseIANAReferral(response string) (string, string) {
+// The server is "" when the response has no referral. A referral to a server
-	var rir string
+// that is none of the five RIRs' is an error, so a forged or broken answer
-
+// cannot send the client to any other host or port.
+func (c *Client) parseIANAReferral(response string) (string, string, error) {
 	for line := range strings.SplitSeq(response, "\n") {
 		line = strings.TrimSpace(line)
 
@@ -168,54 +220,22 @@ func (c *Client) parseIANAReferral(response string) (string, string) {
 
 			switch {
 			case strings.Contains(server, "arin"):
-				return RIRARIN, whoisServerARIN
+				return RIRARIN, whoisServerARIN, nil
 			case strings.Contains(server, "ripe"):
-				return RIRRIPE, whoisServerRIPE
+				return RIRRIPE, whoisServerRIPE, nil
 			case strings.Contains(server, "apnic"):
-				return RIRAPNIC, whoisServerAPNIC
+				return RIRAPNIC, whoisServerAPNIC, nil
 			case strings.Contains(server, "lacnic"):
-				return RIRLACNIC, whoisServerLACNIC
+				return RIRLACNIC, whoisServerLACNIC, nil
 			case strings.Contains(server, "afrinic"):
-				return RIRAFRNIC, whoisServerAFRINIC
+				return RIRAFRNIC, whoisServerAFRINIC, nil
 			default:
-				// Unknown server, add port if missing
+				return "", "", fmt.Errorf("%w: %q", errUnknownServer, server)
-				if !strings.Contains(server, ":") {
-					server += ":43"
-				}
-
-				return "", server
-			}
-		}
-
-		// Also check organisation line for RIR hints
-		if strings.HasPrefix(strings.ToLower(line), "organisation:") {
-			orgRIR := rirInOrganisationLine(strings.ToLower(line))
-			if orgRIR != "" {
-				rir = orgRIR
 			}
 		}
 	}
 
-	return rir, ""
+	return "", "", nil
-}
-
-// rirInOrganisationLine returns the RIR a lowercased IANA "organisation:"
-// line names, or "" when it names none.
-func rirInOrganisationLine(org string) string {
-	switch {
-	case strings.Contains(org, "arin"):
-		return RIRARIN
-	case strings.Contains(org, "ripe"):
-		return RIRRIPE
-	case strings.Contains(org, "apnic"):
-		return RIRAPNIC
-	case strings.Contains(org, "lacnic"):
-		return RIRLACNIC
-	case strings.Contains(org, "afrinic"):
-		return RIRAFRNIC
-	default:
-		return ""
-	}
 }
 
 // parseResponse extracts ASN information from a WHOIS response.

internal/whois/whois_test.go

@@ -0,0 +1,246 @@
+package whois_test
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"io"
+	"net"
+	"strings"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/whois"
+)
+
+// The servers LookupASN asks: IANA first, then the RIR IANA refers it to.
+const (
+	ianaServer = "whois.iana.org:43"
+	ripeServer = "whois.ripe.net:43"
+)
+
+// testASN is the AS the tests look up.
+const testASN = 3333
+
+// ianaReferral is IANA's answer for an AS in a block it gave to RIPE.
+const ianaReferral = `% IANA WHOIS server
+
+refer:        whois.ripe.net
+
+as-block:     3154-3353
+organisation: Assigned by RIPE NCC
+whois:        whois.ripe.net
+changed:      2002-09-01
+source:       IANA
+`
+
+// ripeRecord is RIPE's answer for the AS.
+const ripeRecord = `% This is the RIPE Database query service.
+
+aut-num:        AS3333
+as-name:        RIPE-NCC-AS
+org:            ORG-RIEN1-RIPE
+created:        2002-09-17T13:08:32Z
+last-modified:  2024-01-09T08:55:41Z
+source:         RIPE
+`
+
+// ripeRefusal is RIPE's answer to a host it no longer serves.
+const ripeRefusal = `% This is the RIPE Database query service.
+% The objects are in RPSL format.
+%
+%ERROR:201: access denied for 192.0.2.1
+%
+% Sorry, access from your host has been permanently
+% denied because of a repeated excessive querying.
+`
+
+// ianaPrivateUse is IANA's answer for a private-use AS: its block and no
+// referral.
+const ianaPrivateUse = `% IANA WHOIS server
+
+as-block:     64512-65534
+organisation: Reserved for Private Use
+descr:        Reserved for Private Use
+source:       IANA
+`
+
+// ianaNoObject is IANA's answer when it has nothing for the query.
+const ianaNoObject = `% IANA WHOIS server
+% This query returned 0 objects.
+`
+
+// fillerLine is one of the lines a server that does not stop keeps sending.
+const fillerLine = "remarks:        and more\n"
+
+// endlessAnswerSize is how many bytes of filler lines that server sends:
+// four times the client's cap on an answer.
+const endlessAnswerSize = 4 << 20
+
+// fakeServers returns a DialContext for a whois.Client that connects to a
+// fake server for each address in answers, which reads the query and sends
+// answers[address]. Connecting to any other address is refused.
+func fakeServers(
+	t *testing.T, answers map[string]string,
+) func(context.Context, string, string) (net.Conn, error) {
+	t.Helper()
+
+	var servers sync.WaitGroup
+
+	t.Cleanup(servers.Wait)
+
+	return func(_ context.Context, _, address string) (net.Conn, error) {
+		answer, ok := answers[address]
+		if !ok {
+			return nil, syscall.ECONNREFUSED
+		}
+
+		client, server := net.Pipe()
+
+		servers.Go(func() {
+			defer func() { _ = server.Close() }()
+
+			_, _ = bufio.NewReader(server).ReadString('\n')
+			_, _ = io.WriteString(server, answer)
+		})
+
+		return client, nil
+	}
+}
+
+// lookUp looks testASN up with fake servers sending answers.
+func lookUp(
+	t *testing.T, answers map[string]string,
+) (*whois.ASNInfo, error) {
+	t.Helper()
+
+	client := whois.NewClient()
+	client.DialContext = fakeServers(t, answers)
+
+	return client.LookupASN(t.Context(), testASN)
+}
+
+// TestLookupASNReadsTheRIRAnswer looks up an AS IANA refers to RIPE. The
+// fields come from RIPE's answer, the last-modified date included: IANA's
+// "changed:" line dates its block, not the AS.
+func TestLookupASNReadsTheRIRAnswer(t *testing.T) {
+	t.Parallel()
+
+	info, err := lookUp(t, map[string]string{
+		ianaServer: ianaReferral,
+		ripeServer: ripeRecord,
+	})
+	if err != nil {
+		t.Fatalf("LookupASN: %v", err)
+	}
+
+	if info.ASName != "RIPE-NCC-AS" || info.OrgID != "ORG-RIEN1-RIPE" ||
+		info.RIR != whois.RIRRIPE {
+		t.Errorf("LookupASN = AS name %q, org %q, RIR %q; "+
+			"want RIPE-NCC-AS, ORG-RIEN1-RIPE, RIPE",
+			info.ASName, info.OrgID, info.RIR)
+	}
+
+	wantLastMod := time.Date(2024, 1, 9, 8, 55, 41, 0, time.UTC)
+	if info.LastMod == nil || !info.LastMod.Equal(wantLastMod) {
+		t.Errorf("LookupASN last modified = %v, want %v",
+			info.LastMod, wantLastMod)
+	}
+}
+
+// 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.
+func TestLookupASNFailsWhenTheRIRGivesNoAnswer(t *testing.T) {
+	t.Parallel()
+
+	for name, answers := range map[string]map[string]string{
+		"connection refused": {ianaServer: ianaReferral},
+		"only an error":      {ianaServer: ianaReferral, ripeServer: ripeRefusal},
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			_, err := lookUp(t, answers)
+			if err == nil || errors.Is(err, whois.ErrNoReferral) {
+				t.Errorf("LookupASN error = %v, want one that is not "+
+					"ErrNoReferral", err)
+			}
+		})
+	}
+}
+
+// TestLookupASNReportsNoReferral looks up a private-use AS, which IANA
+// refers to no RIR.
+func TestLookupASNReportsNoReferral(t *testing.T) {
+	t.Parallel()
+
+	_, err := lookUp(t, map[string]string{ianaServer: ianaPrivateUse})
+	if !errors.Is(err, whois.ErrNoReferral) {
+		t.Errorf("LookupASN error = %v, want ErrNoReferral", err)
+	}
+}
+
+// TestLookupASNFailsWhenIANAAnswersNoBlock looks up an AS when IANA answers
+// with neither a referral nor the AS's block. That is not IANA saying no
+// RIR holds the AS.
+func TestLookupASNFailsWhenIANAAnswersNoBlock(t *testing.T) {
+	t.Parallel()
+
+	_, err := lookUp(t, map[string]string{ianaServer: ianaNoObject})
+	if err == nil || errors.Is(err, whois.ErrNoReferral) {
+		t.Errorf("LookupASN error = %v, want one that is not ErrNoReferral",
+			err)
+	}
+}
+
+// TestLookupASNFailsWhenTheRIRAnswerDoesNotEnd looks up an AS whose RIR
+// server follows the record with filler lines far past any real answer.
+// The client stops reading at its cap and fails instead of keeping them.
+func TestLookupASNFailsWhenTheRIRAnswerDoesNotEnd(t *testing.T) {
+	t.Parallel()
+
+	endless := ripeRecord +
+		strings.Repeat(fillerLine, endlessAnswerSize/len(fillerLine))
+
+	_, err := lookUp(t, map[string]string{
+		ianaServer: ianaReferral,
+		ripeServer: endless,
+	})
+	if err == nil || errors.Is(err, whois.ErrNoReferral) {
+		t.Errorf("LookupASN error = %v, want one that is not ErrNoReferral",
+			err)
+	}
+}
+
+// TestLookupASNFailsOnAReferralToAnotherServer looks up an AS when IANA's
+// answer refers it to a server none of the five RIRs run, with a port and
+// without. The client follows a referral only to an RIR's server, so it
+// fails instead of asking that server, which would answer with a record.
+func TestLookupASNFailsOnAReferralToAnotherServer(t *testing.T) {
+	t.Parallel()
+
+	// Each case maps the server on the "refer:" line to the address the
+	// client would dial to follow it. IANA's answer also names the AS's
+	// block, so a client that dropped the referral would report no RIR.
+	for refer, address := range map[string]string{
+		"whois.example.net:8080": "whois.example.net:8080",
+		"whois.example.net":      "whois.example.net:43",
+	} {
+		t.Run(refer, func(t *testing.T) {
+			t.Parallel()
+
+			_, err := lookUp(t, map[string]string{
+				ianaServer: "refer:        " + refer + "\n\n" +
+					"as-block:     3154-3353\n",
+				address: ripeRecord,
+			})
+			if err == nil || errors.Is(err, whois.ErrNoReferral) {
+				t.Errorf("LookupASN error = %v, want one that is not "+
+					"ErrNoReferral", err)
+			}
+		})
+	}
+}