Write each handler batch from its flush goroutine, outside the mutex (closes #65)

The AS, peer and prefix handlers wrote their batch to the database while
holding the mutex HandleMessage needs, and HandleMessage wrote a full batch
itself. For as long as a write took, the handler took nothing off its queue,
and the streamer dropped updates once the queue was half full.

Now only the flush goroutine writes: it takes the batch under the mutex,
releases it, then writes. A full batch wakes the flush goroutine instead.
While a write runs, the next batch grows to at most ten batch sizes; at that
size HandleMessage waits until the flush goroutine takes it (README
"Memory"). Stop still writes what the handler holds.

Model: opus-5-5

README.md

@@ -234,9 +234,14 @@ What happens at each limit:
 - Handler queues: each of the four handler queues holds at most 20,000 messages.
   When a queue fills, the streamer drops messages instead of blocking.
 - Handler batches: while the AS, peer or prefix handler writes a batch to the
-  database, the updates that arrive collect in its next batch, which has no size
+  database, the updates that arrive collect in its next batch, which stops at
-  limit. A database that stays slower than the feed grows memory instead of
+  ten times the batch size: 300,000 updates for the AS handler
-  dropping updates.
+  (`asnBatchMaxSize`, about 10 MiB), 100,000 for the peer handler
+  (`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'
+  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_test.go

@@ -128,7 +128,7 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) {
 
 	// A withdrawal removes one route.
 	if err := db.DeleteLiveRouteBatch([]LiveRouteDeletion{
-		{Prefix: "203.0.113.0/24", OriginASN: 64501, PeerIP: "192.0.2.1", IPVersion: ipVersionV4},
+		{Prefix: "203.0.113.0/24", PeerIP: "192.0.2.1", IPVersion: ipVersionV4},
 	}); err != nil {
 		t.Fatalf("DeleteLiveRouteBatch: %v", err)
 	}
@@ -327,13 +327,11 @@ func announceRoutes(db *Database, batch bool, routes []*LiveRoute) error {
 }
 
 // withdrawRoutes removes routes in one DeleteLiveRouteBatch, or with one
-// DeleteLiveRoute each. It names each route by prefix and peer only, with no
+// DeleteLiveRoute each.
-// origin ASN, as a withdrawal from the feed does when its message carries no AS
-// path.
 func withdrawRoutes(db *Database, batch bool, routes []*LiveRoute) error {
 	if !batch {
 		for _, route := range routes {
-			if err := db.DeleteLiveRoute(route.Prefix, 0, route.PeerIP); err != nil {
+			if err := db.DeleteLiveRoute(route.Prefix, route.PeerIP); err != nil {
 				return err
 			}
 		}

internal/database/database.go

@@ -228,21 +228,22 @@ func (d *Database) beginTx() (*loggingTx, error) {
 }
 
 // A live-route upsert is an UPDATE followed, only when no row matched, by an
-// INSERT. The UPDATE's rows-affected count (1 for an existing key, 0 for a new
+// INSERT. A live route is keyed by prefix and peer, so an announcement replaces
-// one) is what lets the in-memory route counters stay exact without a COUNT(*).
+// the peer's route for the prefix whatever its origin AS. The UPDATE's
-// Callers hold the database write lock, so no other writer can insert the same
+// rows-affected count (1 for an existing key, 0 for a new one) is what lets the
-// key between the two statements. The id column is set only on INSERT, so an
+// in-memory route counters stay exact without a COUNT(*). Callers hold the
-// updated route keeps its original id, exactly as the previous ON CONFLICT
+// database write lock, so no other writer can insert the same key between the
-// upsert did.
+// two statements. The id column is set only on INSERT, so an updated route
+// keeps its original id, exactly as the previous ON CONFLICT upsert did.
 const (
-	updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?,
+	updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, origin_asn = ?, as_path = ?,
-		last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
+		next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
 
 	insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip,
 		as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
 
-	updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?,
+	updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, origin_asn = ?, as_path = ?,
-		last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
+		next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
 
 	insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip,
 		as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
@@ -261,8 +262,8 @@ const (
 // live route for its prefix.
 func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
 	inserted, newPrefix bool, err error) {
-	res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
+	res, err := upd.Exec(route.MaskLength, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
-		route.Prefix, route.OriginASN, route.PeerIP)
+		route.Prefix, route.PeerIP)
 	if err != nil {
 		return false, false, err
 	}
@@ -293,8 +294,8 @@ func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string
 // live route for its prefix.
 func upsertRouteRowV6(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
 	inserted, newPrefix bool, err error) {
-	res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
+	res, err := upd.Exec(route.MaskLength, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
-		route.Prefix, route.OriginASN, route.PeerIP)
+		route.Prefix, route.PeerIP)
 	if err != nil {
 		return false, false, err
 	}
@@ -441,32 +442,18 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 		}
 	}()
 
-	// No longer need to separate deletions since we handle them in the loop below
-
 	// Prepare statements for both IPv4 and IPv6 tables
-	stmtV4WithOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`)
+	deleteV4, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`)
 	if err != nil {
-		return fmt.Errorf("failed to prepare IPv4 delete with origin statement: %w", err)
+		return fmt.Errorf("failed to prepare IPv4 delete statement: %w", err)
 	}
-	defer func() { _ = stmtV4WithOrigin.Close() }()
+	defer func() { _ = deleteV4.Close() }()
 
-	stmtV4WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`)
+	deleteV6, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`)
 	if err != nil {
-		return fmt.Errorf("failed to prepare IPv4 delete without origin statement: %w", err)
+		return fmt.Errorf("failed to prepare IPv6 delete statement: %w", err)
 	}
-	defer func() { _ = stmtV4WithoutOrigin.Close() }()
+	defer func() { _ = deleteV6.Close() }()
-
-	stmtV6WithOrigin, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`)
-	if err != nil {
-		return fmt.Errorf("failed to prepare IPv6 delete with origin statement: %w", err)
-	}
-	defer func() { _ = stmtV6WithOrigin.Close() }()
-
-	stmtV6WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`)
-	if err != nil {
-		return fmt.Errorf("failed to prepare IPv6 delete without origin statement: %w", err)
-	}
-	defer func() { _ = stmtV6WithoutOrigin.Close() }()
 
 	hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL)
 	if err != nil {
@@ -485,38 +472,18 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 	// Mask lengths of the prefixes this batch leaves with no live route.
 	var gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6 []int
 	for _, del := range deletions {
-		var stmt *sql.Stmt
+		stmt := deleteV4
-
+		if del.IPVersion != ipVersionV4 {
-		// Select appropriate statement based on IP version and whether we have origin ASN
+			stmt = deleteV6
-		//nolint:nestif // Clear logic for selecting the right statement
-		if del.IPVersion == ipVersionV4 {
-			if del.OriginASN == 0 {
-				stmt = stmtV4WithoutOrigin
-			} else {
-				stmt = stmtV4WithOrigin
-			}
-		} else {
-			if del.OriginASN == 0 {
-				stmt = stmtV6WithoutOrigin
-			} else {
-				stmt = stmtV6WithOrigin
-			}
-		}
-
-		// Execute deletion
-		var res sql.Result
-		if del.OriginASN == 0 {
-			res, err = stmt.Exec(del.Prefix, del.PeerIP)
-		} else {
-			res, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP)
 		}
 
+		res, err := stmt.Exec(del.Prefix, del.PeerIP)
 		if err != nil {
 			return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err)
 		}
 
-		// A deletion with no origin ASN can remove several rows, so use the
+		// The feed often withdraws a route that is not live, which deletes
-		// exact rows-affected count to keep the in-memory route counters right.
+		// nothing, so the in-memory route counters use the rows-affected count.
 		affected, err := res.RowsAffected()
 		if err != nil {
 			return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err)
@@ -1282,9 +1249,8 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
 	return nil
 }
 
-// DeleteLiveRoute deletes a live route
+// DeleteLiveRoute deletes the live route the peer has for the prefix
-// If originASN is 0, deletes all routes for the prefix/peer combination
+func (d *Database) DeleteLiveRoute(prefix, peerIP string) error {
-func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) error {
 	d.lock("DeleteLiveRoute")
 	defer d.unlock()
 
@@ -1309,22 +1275,12 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string)
 	}()
 
 	// Literal per-table queries (rather than one formatted with the table name)
-	// so the delete carries no dynamically built SQL. A delete with no origin
+	// so the delete carries no dynamically built SQL.
-	// ASN can remove several rows.
 	var res sql.Result
-	switch {
+	if isV4 {
-	case isV4 && originASN == 0:
 		res, err = tx.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
-	case isV4:
+	} else {
-		res, err = tx.Exec(
-			`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
-			prefix, originASN, peerIP)
-	case originASN == 0:
 		res, err = tx.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
-	default:
-		res, err = tx.Exec(
-			`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
-			prefix, originASN, peerIP)
 	}
 	if err != nil {
 		return err

internal/database/database_test.go

@@ -13,6 +13,7 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/config"
 	"git.eeqj.de/sneak/routewatch/internal/logger"
 	"github.com/google/uuid"
+	"github.com/mattn/go-sqlite3"
 )
 
 // tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in
@@ -122,8 +123,8 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) {
 	lookup("2001:db9::1", "")
 
 	err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
-		{Prefix: "10.1.2.0/24", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV4},
+		{Prefix: "10.1.2.0/24", PeerIP: testPeerIP, IPVersion: ipVersionV4},
-		{Prefix: "2001:db8:1:2::/64", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV6},
+		{Prefix: "2001:db8:1:2::/64", PeerIP: testPeerIP, IPVersion: ipVersionV6},
 	})
 	if err != nil {
 		t.Fatalf("DeleteLiveRouteBatch: %v", err)
@@ -205,6 +206,40 @@ func TestGetIPInfoIgnoresShortRoutes(t *testing.T) {
 	lookup("2001:db8::1", "2001:db8::/32", 64502)
 }
 
+// TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer checks that each live route
+// table refuses a second route from a peer for a prefix, even with another
+// origin AS. An upsert updates the peer's route before it would insert one, so
+// only an insert straight into the table shows what the table allows.
+func TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer(t *testing.T) {
+	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() }()
+
+	const otherOriginASN = 64501
+	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
+	for _, tt := range []struct {
+		route     *LiveRoute
+		insertSQL string
+	}{
+		{route: mkV4Route(t, "198.51.100.0/24", 64500, ts), insertSQL: insertLiveRouteV4SQL},
+		{route: mkV6Route("2001:db8::/32", 64500, ts), insertSQL: insertLiveRouteV6SQL},
+	} {
+		if err := db.UpsertLiveRoute(tt.route); err != nil {
+			t.Fatalf("UpsertLiveRoute(%s): %v", tt.route.Prefix, err)
+		}
+
+		_, err := db.db.Exec(tt.insertSQL, uuid.NewString(), tt.route.Prefix, tt.route.MaskLength,
+			otherOriginASN, tt.route.PeerIP, "[]", tt.route.NextHop, ts)
+		var sqliteErr sqlite3.Error
+		if !errors.As(err, &sqliteErr) || sqliteErr.ExtendedCode != sqlite3.ErrConstraintUnique {
+			t.Errorf("second route from peer %s for %s: error %v, want a UNIQUE constraint error",
+				tt.route.PeerIP, tt.route.Prefix, err)
+		}
+	}
+}
+
 // TestConnectionPoolPragmas holds several pooled connections open at once and
 // checks each one carries the per-connection settings from the DSN, plus the
 // process-wide hard heap limit, and sees the new file with auto_vacuum
@@ -306,7 +341,6 @@ func TestVacuumReturnsFreePages(t *testing.T) {
 		routes = append(routes, route)
 		deletions = append(deletions, LiveRouteDeletion{
 			Prefix:    route.Prefix,
-			OriginASN: asn,
 			PeerIP:    route.PeerIP,
 			IPVersion: ipVersionV4,
 		})

internal/database/interface.go

@@ -55,7 +55,7 @@ type Store interface {
 	// Live route operations
 	UpsertLiveRoute(route *LiveRoute) error
 	UpsertLiveRouteBatch(routes []*LiveRoute) error
-	DeleteLiveRoute(prefix string, originASN int, peerIP string) error
+	DeleteLiveRoute(prefix, peerIP string) error
 	DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error
 	GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
 	GetPrefixDistributionContext(ctx context.Context) (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)

internal/database/models.go

@@ -114,7 +114,6 @@ type IPInfo struct {
 // LiveRouteDeletion represents parameters for deleting a live route
 type LiveRouteDeletion struct {
 	Prefix    string
-	OriginASN int
 	PeerIP    string
 	IPVersion int
 }

internal/database/schema.sql

@@ -107,7 +107,7 @@ CREATE TABLE IF NOT EXISTS live_routes_v4 (
 	as_path TEXT NOT NULL, -- JSON array
 	next_hop TEXT NOT NULL,
 	last_updated DATETIME NOT NULL,
-	UNIQUE(prefix, origin_asn, peer_ip)
+	UNIQUE(prefix, peer_ip) -- a peer has at most one route for a prefix
 );
 
 -- IPv6 routing table maintained by PrefixHandler
@@ -120,7 +120,7 @@ CREATE TABLE IF NOT EXISTS live_routes_v6 (
 	as_path TEXT NOT NULL, -- JSON array
 	next_hop TEXT NOT NULL,
 	last_updated DATETIME NOT NULL,
-	UNIQUE(prefix, origin_asn, peer_ip)
+	UNIQUE(prefix, peer_ip) -- a peer has at most one route for a prefix
 );
 
 -- Indexes for live_routes_v4 table

internal/routewatch/app_integration_test.go

@@ -205,7 +205,7 @@ func (m *mockStore) UpsertLiveRoute(route *database.LiveRoute) error {
 }
 
 // DeleteLiveRoute mock implementation
-func (m *mockStore) DeleteLiveRoute(prefix string, originASN int, peerIP string) error {
+func (m *mockStore) DeleteLiveRoute(prefix, peerIP string) error {
 	// Simple mock - just return nil
 	return nil
 }

internal/routewatch/ashandler.go

@@ -19,6 +19,12 @@ const (
 	// asnBatchSize is the number of ASN operations to batch together
 	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
@@ -33,10 +39,11 @@ type ASHandler struct {
 	logger *logger.Logger
 
 	// Batching. mu guards batch only: the flush goroutine takes the batch
-	// under mu and writes it after releasing mu, so HandleMessage never
+	// under mu and writes it after releasing mu, so HandleMessage waits for
-	// waits for a database write.
+	// a database write only while batch is at asnBatchMaxSize.
 	mu          sync.Mutex
 	batch       []asnOp
+	batchTaken  sync.Cond     // signalled when the flush goroutine takes batch; its L is &mu
 	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
 	lastFlush   time.Time     // used only by the flush goroutine
 	stopCh      chan struct{}
@@ -62,6 +69,7 @@ func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
 		lastFlush:   time.Now(),
 		stopCh:      make(chan struct{}),
 	}
+	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
 	h.wg.Add(1)
@@ -104,6 +112,13 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
+	// While the batch is at its maximum size, wait until the flush goroutine
+	// takes it. Meanwhile the handler takes nothing off its queue, which
+	// fills, and the streamer drops messages for it.
+	for len(h.batch) >= asnBatchMaxSize {
+		h.batchTaken.Wait()
+	}
+
 	// Queue origin ASN operation
 	if originASN > 0 {
 		h.batch = append(h.batch, asnOp{
@@ -123,7 +138,8 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 
 	// A full batch wakes the flush goroutine. If it is writing, the wake-up
-	// waits in batchFullCh and the batch keeps growing until the write ends.
+	// waits in batchFullCh and the batch keeps growing until the write ends
+	// or the batch reaches asnBatchMaxSize.
 	if len(h.batch) >= asnBatchSize {
 		select {
 		case h.batchFullCh <- struct{}{}:
@@ -167,6 +183,7 @@ func (h *ASHandler) takeBatch() []asnOp {
 	}
 	batch := h.batch
 	h.batch = make([]asnOp, 0, asnBatchSize)
+	h.batchTaken.Broadcast()
 
 	return batch
 }

internal/routewatch/batchwrite_test.go

@@ -11,17 +11,24 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/ristypes"
 )
 
-// handleMessageTimeout is how long a test waits for HandleMessage to return,
+const (
-// or for a write to start, before it fails.
+	// handleMessageTimeout is how long a test waits for HandleMessage to
-const handleMessageTimeout = 2 * time.Second
+	// return, or for a write to start, before it fails.
+	handleMessageTimeout = 2 * time.Second
+
+	// stillWaitingAfter is how long a test watches a HandleMessage that must
+	// wait. One that does not wait returns well within it.
+	stillWaitingAfter = 100 * time.Millisecond
+)
 
 // blockingStore is a mockStore whose batch writes for the AS, peer and prefix
-// handlers wait until the test closes release. writeStarted is closed when the
+// handlers wait until the test calls releaseWrites. writeStarted is closed when
-// first of them starts. It records what each write was given.
+// the first of them starts. It records what each write was given.
 type blockingStore struct {
 	*mockStore
 
 	release      chan struct{}
+	releaseOnce  sync.Once
 	writeStarted chan struct{}
 	startOnce    sync.Once
 
@@ -47,6 +54,12 @@ func (s *blockingStore) waitForRelease() {
 	<-s.release
 }
 
+// releaseWrites lets the held writes, and all later ones, finish. A second call
+// does nothing.
+func (s *blockingStore) releaseWrites() {
+	s.releaseOnce.Do(func() { close(s.release) })
+}
+
 func (s *blockingStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
 	s.waitForRelease()
 	s.mu.Lock()
@@ -96,24 +109,16 @@ func returnsInTime(f func()) bool {
 	}
 }
 
-// handleWhileAWriteIsBlocked hands h the messages in fill, which fill its
+// startBlockedWrite hands h the messages in fill, which fill its batch so that
-// batch so that the handler writes it to store, waits until the write starts,
+// the handler writes it to store, and waits until that write starts. store
-// and hands h the message extra while store holds the write. HandleMessage
+// holds it until the test calls releaseWrites.
-// must return each time. Then it lets the writes finish and stops h, which
+func startBlockedWrite(
-// writes what it holds.
-func handleWhileAWriteIsBlocked(
 	t *testing.T,
 	store *blockingStore,
-	h interface {
+	h interface{ HandleMessage(*ristypes.RISMessage) },
-		HandleMessage(*ristypes.RISMessage)
-		Stop()
-	},
 	fill []*ristypes.RISMessage,
-	extra *ristypes.RISMessage,
 ) {
 	t.Helper()
-	defer h.Stop()
-	defer close(store.release)
 
 	if !returnsInTime(func() {
 		for _, msg := range fill {
@@ -128,31 +133,98 @@ func handleWhileAWriteIsBlocked(
 	case <-time.After(handleMessageTimeout):
 		t.Fatal("the handler did not write its full batch")
 	}
+}
+
+// handleWhileAWriteIsBlocked starts a blocked write with the messages in fill
+// and hands h the message extra while store holds the write. HandleMessage must
+// return each time. Then it lets the writes finish and stops h, which writes
+// what it holds.
+func handleWhileAWriteIsBlocked(
+	t *testing.T,
+	store *blockingStore,
+	h interface {
+		HandleMessage(*ristypes.RISMessage)
+		Stop()
+	},
+	fill []*ristypes.RISMessage,
+	extra *ristypes.RISMessage,
+) {
+	t.Helper()
+	defer h.Stop()
+	defer store.releaseWrites()
+
+	startBlockedWrite(t, store, h, fill)
 
 	if !returnsInTime(func() { h.HandleMessage(extra) }) {
 		t.Fatal("HandleMessage did not return while a database write was blocked")
 	}
 }
 
+// handleAtMaxBatchSize starts a blocked write with the messages in fill, then,
+// while store holds the write, hands h the messages in toMax, which bring its
+// next batch to its maximum size, and the message extra. HandleMessage for
+// extra must wait until the test lets the write finish, and return after. Then
+// it stops h, which writes what it holds.
+func handleAtMaxBatchSize(
+	t *testing.T,
+	store *blockingStore,
+	h interface {
+		HandleMessage(*ristypes.RISMessage)
+		Stop()
+	},
+	fill, toMax []*ristypes.RISMessage,
+	extra *ristypes.RISMessage,
+) {
+	t.Helper()
+	defer h.Stop()
+	defer store.releaseWrites()
+
+	startBlockedWrite(t, store, h, fill)
+	for _, msg := range toMax {
+		h.HandleMessage(msg)
+	}
+
+	returned := make(chan struct{})
+	go func() {
+		h.HandleMessage(extra)
+		close(returned)
+	}()
+
+	select {
+	case <-returned:
+		t.Fatal("HandleMessage returned while the batch was at its maximum size and a write was blocked")
+	case <-time.After(stillWaitingAfter):
+	}
+
+	store.releaseWrites()
+	select {
+	case <-returned:
+	case <-time.After(handleMessageTimeout):
+		t.Fatal("HandleMessage did not return after the database write finished")
+	}
+}
+
+// prefixAnnouncement returns a message in which testPeerIP announces prefixes
+// with origin AS testASNA.
+func prefixAnnouncement(prefixes []string) *ristypes.RISMessage {
+	return &ristypes.RISMessage{
+		Peer:          testPeerIP,
+		Path:          ristypes.ASPath{testASNA},
+		Announcements: []ristypes.RISAnnouncement{{NextHop: testPeerIP, Prefixes: prefixes}},
+	}
+}
+
 func TestPrefixHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
 	const (
 		fillPrefix  = "192.0.2.0/24"
 		extraPrefix = "198.51.100.0/24"
 	)
 
-	announce := func(prefixes []string) *ristypes.RISMessage {
-		return &ristypes.RISMessage{
-			Peer:          testPeerIP,
-			Path:          ristypes.ASPath{testASNA},
-			Announcements: []ristypes.RISAnnouncement{{NextHop: testPeerIP, Prefixes: prefixes}},
-		}
-	}
-
 	store := newBlockingStore()
 	h := NewPrefixHandler(store, logger.New())
 	handleWhileAWriteIsBlocked(t, store, h,
-		[]*ristypes.RISMessage{announce(slices.Repeat([]string{fillPrefix}, prefixBatchSize))},
+		[]*ristypes.RISMessage{prefixAnnouncement(slices.Repeat([]string{fillPrefix}, prefixBatchSize))},
-		announce([]string{extraPrefix}))
+		prefixAnnouncement([]string{extraPrefix}))
 
 	store.mu.Lock()
 	defer store.mu.Unlock()
@@ -190,3 +262,113 @@ func TestPeerHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
 		t.Errorf("Stop did not write peer %s, handled while a write was blocked", otherPeerIP)
 	}
 }
+
+func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
+	const (
+		fillPrefix  = "192.0.2.0/24"
+		extraPrefix = "198.51.100.0/24"
+	)
+
+	store := newBlockingStore()
+	h := NewPrefixHandler(store, logger.New())
+	handleAtMaxBatchSize(t, store, h,
+		[]*ristypes.RISMessage{prefixAnnouncement(slices.Repeat([]string{fillPrefix}, prefixBatchSize))},
+		[]*ristypes.RISMessage{prefixAnnouncement(slices.Repeat([]string{fillPrefix}, prefixBatchMaxSize))},
+		prefixAnnouncement([]string{extraPrefix}))
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if !store.prefixes[extraPrefix] {
+		t.Errorf("Stop did not write %s, handled after the batch reached its maximum size", extraPrefix)
+	}
+}
+
+func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
+	store := newBlockingStore()
+	h := NewASHandler(store, logger.New())
+	handleAtMaxBatchSize(t, store, h,
+		[]*ristypes.RISMessage{{Path: slices.Repeat(ristypes.ASPath{testASNA}, asnBatchSize)}},
+		[]*ristypes.RISMessage{{Path: slices.Repeat(ristypes.ASPath{testASNA}, asnBatchMaxSize)}},
+		&ristypes.RISMessage{Path: ristypes.ASPath{testASNB}})
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if !store.asns[testASNB] {
+		t.Errorf("Stop did not write AS%d, handled after the batch reached its maximum size", testASNB)
+	}
+}
+
+func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
+	const otherPeerIP = "2001:db8:ffff::2"
+
+	store := newBlockingStore()
+	h := NewPeerHandler(store, logger.New())
+	handleAtMaxBatchSize(t, store, h,
+		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
+		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchMaxSize),
+		&ristypes.RISMessage{Peer: otherPeerIP})
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if !store.peers[otherPeerIP] {
+		t.Errorf("Stop did not write peer %s, handled after the batch reached its maximum size", otherPeerIP)
+	}
+}
+
+// TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull hands each handler, built
+// without its flush goroutine so that nothing takes a wake-up, one update short
+// of a full batch, and a second one exactly a full batch. Only the full batch
+// leaves a wake-up in batchFullCh.
+func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
+	tests := []struct {
+		name      string
+		batchSize int
+		// handle hands a new handler n updates and returns its batchFullCh.
+		handle func(n int) chan struct{}
+	}{
+		{
+			name:      "prefix",
+			batchSize: prefixBatchSize,
+			handle: func(n int) chan struct{} {
+				h := &PrefixHandler{batchFullCh: make(chan struct{}, 1)}
+				h.HandleMessage(prefixAnnouncement(slices.Repeat([]string{"192.0.2.0/24"}, n)))
+
+				return h.batchFullCh
+			},
+		},
+		{
+			name:      "AS",
+			batchSize: asnBatchSize,
+			handle: func(n int) chan struct{} {
+				h := &ASHandler{batchFullCh: make(chan struct{}, 1)}
+				// The origin AS and each AS in the path are one update each.
+				h.HandleMessage(&ristypes.RISMessage{Path: slices.Repeat(ristypes.ASPath{testASNA}, n-1)})
+
+				return h.batchFullCh
+			},
+		},
+		{
+			name:      "peer",
+			batchSize: peerBatchSize,
+			handle: func(n int) chan struct{} {
+				h := &PeerHandler{batchFullCh: make(chan struct{}, 1)}
+				for range n {
+					h.HandleMessage(&ristypes.RISMessage{Peer: testPeerIP})
+				}
+
+				return h.batchFullCh
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := len(tt.handle(tt.batchSize - 1)); got != 0 {
+				t.Errorf("one update short of a full batch left %d wake-ups, want none", got)
+			}
+			if got := len(tt.handle(tt.batchSize)); got != 1 {
+				t.Errorf("a full batch left %d wake-ups, want 1", got)
+			}
+		})
+	}
+}

internal/routewatch/peerhandler.go

@@ -23,6 +23,12 @@ const (
 	// peerBatchSize is the number of peer updates to batch together
 	peerBatchSize = 10000
 
+	// peerBatchMaxSize is the most peer updates 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.
+	peerBatchMaxSize = 100000
+
 	// peerBatchTimeout is the maximum time to wait before flushing a batch
 	peerBatchTimeout = 2 * time.Second
 )
@@ -36,10 +42,11 @@ type PeerHandler struct {
 	logger *logger.Logger
 
 	// Batching. mu guards peerBatch only: the flush goroutine takes the batch
-	// under mu and writes it after releasing mu, so HandleMessage never
+	// under mu and writes it after releasing mu, so HandleMessage waits for
-	// waits for a database write.
+	// a database write only while peerBatch is at peerBatchMaxSize.
 	mu          sync.Mutex
 	peerBatch   []peerUpdate
+	batchTaken  sync.Cond     // signalled when the flush goroutine takes peerBatch; its L is &mu
 	batchFullCh chan struct{} // wakes the flush goroutine when peerBatch is full
 	lastFlush   time.Time     // used only by the flush goroutine
 	stopCh      chan struct{}
@@ -66,6 +73,7 @@ func NewPeerHandler(db database.Store, logger *logger.Logger) *PeerHandler {
 		lastFlush:   time.Now(),
 		stopCh:      make(chan struct{}),
 	}
+	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
 	h.wg.Add(1)
@@ -105,6 +113,13 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
+	// While the batch is at its maximum size, wait until the flush goroutine
+	// takes it. Meanwhile the handler takes nothing off its queue, which
+	// fills, and the streamer drops messages for it.
+	for len(h.peerBatch) >= peerBatchMaxSize {
+		h.batchTaken.Wait()
+	}
+
 	// Add to batch
 	h.peerBatch = append(h.peerBatch, peerUpdate{
 		peerIP:      msg.Peer,
@@ -114,7 +129,8 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 	})
 
 	// A full batch wakes the flush goroutine. If it is writing, the wake-up
-	// waits in batchFullCh and the batch keeps growing until the write ends.
+	// waits in batchFullCh and the batch keeps growing until the write ends
+	// or the batch reaches peerBatchMaxSize.
 	if len(h.peerBatch) >= peerBatchSize {
 		select {
 		case h.batchFullCh <- struct{}{}:
@@ -158,6 +174,7 @@ func (h *PeerHandler) takeBatch() []peerUpdate {
 	}
 	batch := h.peerBatch
 	h.peerBatch = make([]peerUpdate, 0, peerBatchSize)
+	h.batchTaken.Broadcast()
 
 	return batch
 }

internal/routewatch/prefixhandler.go

@@ -24,6 +24,12 @@ const (
 	// prefixBatchSize is the number of prefix updates to batch together
 	prefixBatchSize = 25000
 
+	// 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.
+	prefixBatchMaxSize = 250000
+
 	// prefixBatchTimeout is the maximum time to wait before flushing a batch
 	// DO NOT reduce this timeout - larger batches are more efficient
 	prefixBatchTimeout = 1 * time.Second
@@ -41,10 +47,11 @@ type PrefixHandler struct {
 	metrics *metrics.Tracker
 
 	// Batching. mu guards batch only: the flush goroutine takes the batch
-	// under mu and writes it after releasing mu, so HandleMessage never
+	// under mu and writes it after releasing mu, so HandleMessage waits for
-	// waits for a database write.
+	// a database write only while batch is at prefixBatchMaxSize.
 	mu          sync.Mutex
 	batch       []prefixUpdate
+	batchTaken  sync.Cond     // signalled when the flush goroutine takes batch; its L is &mu
 	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
 	lastFlush   time.Time     // used only by the flush goroutine
 	stopCh      chan struct{}
@@ -70,6 +77,7 @@ func NewPrefixHandler(db database.Store, logger *logger.Logger) *PrefixHandler {
 		lastFlush:   time.Now(),
 		stopCh:      make(chan struct{}),
 	}
+	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
 	h.wg.Add(1)
@@ -109,17 +117,24 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
+	// While the batch is at its maximum size, wait until the flush goroutine
+	// takes it. Meanwhile the handler takes nothing off its queue, which
+	// fills, and the streamer drops messages for it.
+	for len(h.batch) >= prefixBatchMaxSize {
+		h.batchTaken.Wait()
+	}
+
 	// Withdrawals are queued before announcements, because the batch keeps
 	// each peer's last update for a prefix: a message that lists a prefix as
 	// both withdrawn and announced keeps the route (RFC 4271 section 4.3).
+	// A withdrawal names only the prefix and peer: the message's AS path, and
+	// so its origin AS, belongs to the announced prefixes.
 	for _, prefix := range msg.Withdrawals {
 		h.batch = append(h.batch, prefixUpdate{
 			prefix:      canonicalPrefix(prefix),
-			originASN:   originASN, // Use the originASN from path if available
 			peer:        msg.Peer,
 			messageType: "withdrawal",
 			timestamp:   timestamp,
-			path:        msg.Path,
 		})
 		// Record withdrawal in metrics
 		if h.metrics != nil {
@@ -145,7 +160,8 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 
 	// A full batch wakes the flush goroutine. If it is writing, the wake-up
-	// waits in batchFullCh and the batch keeps growing until the write ends.
+	// waits in batchFullCh and the batch keeps growing until the write ends
+	// or the batch reaches prefixBatchMaxSize.
 	if len(h.batch) >= prefixBatchSize {
 		select {
 		case h.batchFullCh <- struct{}{}:
@@ -189,6 +205,7 @@ func (h *PrefixHandler) takeBatch() []prefixUpdate {
 	}
 	batch := h.batch
 	h.batch = make([]prefixUpdate, 0, prefixBatchSize)
+	h.batchTaken.Broadcast()
 
 	return batch
 }
@@ -245,7 +262,6 @@ func (h *PrefixHandler) flushBatch() {
 			// Create deletion record for batch delete
 			routesToDelete = append(routesToDelete, database.LiveRouteDeletion{
 				Prefix:    update.prefix,
-				OriginASN: update.originASN,
 				PeerIP:    update.peer,
 				IPVersion: ipVersion,
 			})
@@ -447,51 +463,24 @@ func (h *PrefixHandler) processAnnouncementDirect(update prefixUpdate) {
 // processWithdrawalDirect handles removing a route directly without prefix table
 // nolint:unused // kept for potential future use
 func (h *PrefixHandler) processWithdrawalDirect(update prefixUpdate) {
-	// For withdrawals, we need to delete the route from live_routes
+	if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil {
-	if update.originASN > 0 {
+		h.logger.Error("Failed to delete live route",
-		if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
+			"prefix", update.prefix,
-			h.logger.Error("Failed to delete live route",
+			"peer", update.peer,
-				"prefix", update.prefix,
+			"error", err,
-				"origin_asn", update.originASN,
+		)
-				"peer", update.peer,
-				"error", err,
-			)
-		}
-	} else {
-		// If no origin ASN, just delete all routes for this prefix from this peer
-		if err := h.db.DeleteLiveRoute(update.prefix, 0, update.peer); err != nil {
-			h.logger.Error("Failed to delete live route (no origin ASN)",
-				"prefix", update.prefix,
-				"peer", update.peer,
-				"error", err,
-			)
-		}
 	}
 }
 
 // processWithdrawal handles removing a route from the live routing table
 // nolint:unused // kept for potential future use
 func (h *PrefixHandler) processWithdrawal(_ *database.Prefix, update prefixUpdate) {
-	// For withdrawals, we need to delete the route from live_routes
+	if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil {
-	// Since we have the origin ASN from the update, we can delete the specific route
+		h.logger.Error("Failed to delete live route",
-	if update.originASN > 0 {
+			"prefix", update.prefix,
-		if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
+			"peer", update.peer,
-			h.logger.Error("Failed to delete live route",
+			"error", err,
-				"prefix", update.prefix,
+		)
-				"origin_asn", update.originASN,
-				"peer", update.peer,
-				"error", err,
-			)
-		}
-	} else {
-		// If no origin ASN, just delete all routes for this prefix from this peer
-		if err := h.db.DeleteLiveRoute(update.prefix, 0, update.peer); err != nil {
-			h.logger.Error("Failed to delete live route (no origin ASN)",
-				"prefix", update.prefix,
-				"peer", update.peer,
-				"error", err,
-			)
-		}
 	}
 }
 

internal/routewatch/prefixhandler_test.go

@@ -3,6 +3,7 @@ package routewatch
 import (
 	"context"
 	"errors"
+	"fmt"
 	"slices"
 	"testing"
 	"time"
@@ -175,32 +176,153 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
 				t.Errorf("peers with a live route for %s = %v, want %v", prefix, gotPeers, tt.wantPeers)
 			}
 
-			stats, err := db.GetStatsContext(ctx)
-			if err != nil {
-				t.Fatalf("GetStatsContext: %v", err)
-			}
-			tableRoutesV4, _, err := db.GetLiveRouteCountsContext(ctx)
-			if err != nil {
-				t.Fatalf("GetLiveRouteCountsContext: %v", err)
-			}
-			if stats.IPv4Routes != len(tt.wantPeers) || tableRoutesV4 != len(tt.wantPeers) {
-				t.Errorf("IPv4 live routes: %d in memory, %d in the table, want %d",
-					stats.IPv4Routes, tableRoutesV4, len(tt.wantPeers))
-			}
-
 			var wantDistribution []database.PrefixDistribution
 			if len(tt.wantPeers) > 0 {
 				wantDistribution = []database.PrefixDistribution{{MaskLength: prefixMaskLength, Count: 1}}
 			}
-			queryDistributionV4, _, err := db.GetPrefixDistributionContext(ctx)
+			assertCounts(t, db, ipv4Version, len(tt.wantPeers), wantDistribution)
-			if err != nil {
-				t.Fatalf("GetPrefixDistributionContext: %v", err)
-			}
-			if !slices.Equal(stats.IPv4PrefixDistribution, wantDistribution) ||
-				!slices.Equal(queryDistributionV4, wantDistribution) {
-				t.Errorf("IPv4 prefix distribution: %v in memory, %v from the table, want %v",
-					stats.IPv4PrefixDistribution, queryDistributionV4, wantDistribution)
-			}
 		})
 	}
 }
+
+// TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer runs updates from one peer
+// through the prefix handler into a real database, flushing each message as its
+// own batch, once with IPv4 prefixes and once with IPv6 prefixes. A peer has at
+// most one route for a prefix: a later announcement replaces it whatever its
+// origin AS, and a withdrawal removes it whatever AS the message's AS path ends
+// in. The in-memory route count and prefix distribution must match the route
+// table.
+func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) {
+	families := []struct {
+		ipVersion           int
+		prefix, otherPrefix string
+		// The mask length of both prefixes.
+		maskLength int
+	}{
+		{ipVersion: ipv4Version, prefix: "198.51.100.0/24", otherPrefix: "203.0.113.0/24", maskLength: 24},
+		{ipVersion: ipv6Version, prefix: "2001:db8:1::/48", otherPrefix: "2001:db8:2::/48", maskLength: 48},
+	}
+
+	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
+	announce := func(path ristypes.ASPath, announced string) *ristypes.RISMessage {
+		return &ristypes.RISMessage{
+			Peer:            testPeerIP,
+			Path:            path,
+			ParsedTimestamp: ts,
+			Announcements:   []ristypes.RISAnnouncement{{NextHop: testPeerIP, Prefixes: []string{announced}}},
+		}
+	}
+
+	for _, family := range families {
+		prefix, otherPrefix := family.prefix, family.otherPrefix
+
+		// One message that announces otherPrefix with origin AS testASNB and
+		// withdraws prefix.
+		announceOtherAndWithdraw := announce(ristypes.ASPath{testASNA, testASNB}, otherPrefix)
+		announceOtherAndWithdraw.Withdrawals = []string{prefix}
+
+		tests := []struct {
+			name     string
+			messages []*ristypes.RISMessage
+			// The origin AS of the peer's live route for each prefix. A prefix
+			// left out must have no live route.
+			wantOrigins map[string]int
+		}{
+			{
+				name: "the peer announces the prefix with a new origin AS",
+				messages: []*ristypes.RISMessage{
+					announce(ristypes.ASPath{testASNA}, prefix),
+					announce(ristypes.ASPath{testASNA, testASNB}, prefix),
+				},
+				wantOrigins: map[string]int{prefix: testASNB},
+			},
+			{
+				name: "one message announces a prefix with another origin AS and withdraws the prefix",
+				messages: []*ristypes.RISMessage{
+					announce(ristypes.ASPath{testASNA}, prefix),
+					announceOtherAndWithdraw,
+				},
+				wantOrigins: map[string]int{otherPrefix: testASNB},
+			},
+		}
+
+		for _, tt := range tests {
+			t.Run(fmt.Sprintf("IPv%d %s", family.ipVersion, tt.name), func(t *testing.T) {
+				db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
+				if err != nil {
+					t.Fatalf("failed to create database: %v", err)
+				}
+				defer func() { _ = db.Close() }()
+
+				// Built without NewPrefixHandler's flush timer, so the test flushes itself.
+				h := &PrefixHandler{db: db, logger: logger.New()}
+				for _, msg := range tt.messages {
+					h.HandleMessage(msg)
+					h.flushBatch()
+				}
+
+				for _, p := range []string{prefix, otherPrefix} {
+					routes, _, err := db.GetPrefixDetailsContext(context.Background(), p)
+					if err != nil && !errors.Is(err, database.ErrNoRoute) {
+						t.Fatalf("GetPrefixDetailsContext(%s): %v", p, err)
+					}
+					var gotOrigins []int
+					for _, route := range routes {
+						gotOrigins = append(gotOrigins, route.OriginASN)
+					}
+					var wantOrigins []int
+					if origin, ok := tt.wantOrigins[p]; ok {
+						wantOrigins = []int{origin}
+					}
+					if !slices.Equal(gotOrigins, wantOrigins) {
+						t.Errorf("origin ASes of the live routes for %s = %v, want %v", p, gotOrigins, wantOrigins)
+					}
+				}
+
+				wantDistribution := []database.PrefixDistribution{
+					{MaskLength: family.maskLength, Count: len(tt.wantOrigins)},
+				}
+				assertCounts(t, db, family.ipVersion, len(tt.wantOrigins), wantDistribution)
+			})
+		}
+	}
+}
+
+// assertCounts checks that the live route count and prefix distribution of one
+// IP version are wantRoutes and wantDistribution, both as held in memory and as
+// read from the route table.
+func assertCounts(t *testing.T, db *database.Database, ipVersion, wantRoutes int,
+	wantDistribution []database.PrefixDistribution) {
+	t.Helper()
+
+	ctx := context.Background()
+	stats, err := db.GetStatsContext(ctx)
+	if err != nil {
+		t.Fatalf("GetStatsContext: %v", err)
+	}
+	tableRoutesV4, tableRoutesV6, err := db.GetLiveRouteCountsContext(ctx)
+	if err != nil {
+		t.Fatalf("GetLiveRouteCountsContext: %v", err)
+	}
+	queryDistributionV4, queryDistributionV6, err := db.GetPrefixDistributionContext(ctx)
+	if err != nil {
+		t.Fatalf("GetPrefixDistributionContext: %v", err)
+	}
+
+	memoryRoutes, tableRoutes := stats.IPv4Routes, tableRoutesV4
+	memoryDistribution, queryDistribution := stats.IPv4PrefixDistribution, queryDistributionV4
+	if ipVersion == ipv6Version {
+		memoryRoutes, tableRoutes = stats.IPv6Routes, tableRoutesV6
+		memoryDistribution, queryDistribution = stats.IPv6PrefixDistribution, queryDistributionV6
+	}
+
+	if memoryRoutes != wantRoutes || tableRoutes != wantRoutes {
+		t.Errorf("IPv%d live routes: %d in memory, %d in the table, want %d",
+			ipVersion, memoryRoutes, tableRoutes, wantRoutes)
+	}
+	if !slices.Equal(memoryDistribution, wantDistribution) ||
+		!slices.Equal(queryDistribution, wantDistribution) {
+		t.Errorf("IPv%d prefix distribution: %v in memory, %v from the table, want %v",
+			ipVersion, memoryDistribution, queryDistribution, wantDistribution)
+	}
+}