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.
Stop still writes what the handler holds.

While a write runs, the next batch has no size limit; the README memory
section says so.

Model: opus-5-5

README.md

@@ -233,6 +233,10 @@ What happens at each limit:
   keeps running.
 - 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
+  limit. A database that stays slower than the feed grows memory instead of
+  dropping updates.
 
 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", PeerIP: "192.0.2.1", IPVersion: ipVersionV4},
+		{Prefix: "203.0.113.0/24", OriginASN: 64501, PeerIP: "192.0.2.1", IPVersion: ipVersionV4},
 	}); err != nil {
 		t.Fatalf("DeleteLiveRouteBatch: %v", err)
 	}
@@ -327,11 +327,13 @@ func announceRoutes(db *Database, batch bool, routes []*LiveRoute) error {
 }
 
 // withdrawRoutes removes routes in one DeleteLiveRouteBatch, or with one
-// DeleteLiveRoute each.
+// DeleteLiveRoute each. It names each route by prefix and peer only, with no
+// 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, route.PeerIP); err != nil {
+			if err := db.DeleteLiveRoute(route.Prefix, 0, route.PeerIP); err != nil {
 				return err
 			}
 		}

internal/database/database.go

@@ -228,22 +228,21 @@ func (d *Database) beginTx() (*loggingTx, error) {
 }
 
 // A live-route upsert is an UPDATE followed, only when no row matched, by an
-// INSERT. A live route is keyed by prefix and peer, so an announcement replaces
-// the peer's route for the prefix whatever its origin AS. The UPDATE's
-// rows-affected count (1 for an existing key, 0 for a new one) is what lets the
-// in-memory route counters stay exact without a COUNT(*). Callers hold the
-// database write lock, so no other writer can insert the same key between the
-// 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.
+// INSERT. The UPDATE's rows-affected count (1 for an existing key, 0 for a new
+// one) is what lets the in-memory route counters stay exact without a COUNT(*).
+// Callers hold the database write lock, so no other writer can insert the same
+// key between the 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 = ?, origin_asn = ?, as_path = ?,
-		next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
+	updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?,
+		last_updated = ? WHERE prefix = ? AND origin_asn = ? 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 = ?, origin_asn = ?, as_path = ?,
-		next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
+	updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?,
+		last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
 
 	insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip,
 		as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
@@ -262,8 +261,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, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
-		route.Prefix, route.PeerIP)
+	res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
+		route.Prefix, route.OriginASN, route.PeerIP)
 	if err != nil {
 		return false, false, err
 	}
@@ -294,8 +293,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, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
-		route.Prefix, route.PeerIP)
+	res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
+		route.Prefix, route.OriginASN, route.PeerIP)
 	if err != nil {
 		return false, false, err
 	}
@@ -442,18 +441,32 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 		}
 	}()
 
-	// Prepare statements for both IPv4 and IPv6 tables
-	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 statement: %w", err)
-	}
-	defer func() { _ = deleteV4.Close() }()
+	// No longer need to separate deletions since we handle them in the loop below
 
-	deleteV6, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`)
+	// 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 = ?`)
 	if err != nil {
-		return fmt.Errorf("failed to prepare IPv6 delete statement: %w", err)
+		return fmt.Errorf("failed to prepare IPv4 delete with origin statement: %w", err)
 	}
-	defer func() { _ = deleteV6.Close() }()
+	defer func() { _ = stmtV4WithOrigin.Close() }()
+
+	stmtV4WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`)
+	if err != nil {
+		return fmt.Errorf("failed to prepare IPv4 delete without origin statement: %w", err)
+	}
+	defer func() { _ = stmtV4WithoutOrigin.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 {
@@ -472,18 +485,38 @@ 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 {
-		stmt := deleteV4
-		if del.IPVersion != ipVersionV4 {
-			stmt = deleteV6
+		var stmt *sql.Stmt
+
+		// Select appropriate statement based on IP version and whether we have origin ASN
+		//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)
 		}
 
-		// The feed often withdraws a route that is not live, which deletes
-		// nothing, so the in-memory route counters use the rows-affected count.
+		// A deletion with no origin ASN can remove several rows, so use the
+		// exact rows-affected count to keep the in-memory route counters right.
 		affected, err := res.RowsAffected()
 		if err != nil {
 			return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err)
@@ -1249,8 +1282,9 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
 	return nil
 }
 
-// DeleteLiveRoute deletes the live route the peer has for the prefix
-func (d *Database) DeleteLiveRoute(prefix, peerIP string) error {
+// DeleteLiveRoute deletes a live route
+// If originASN is 0, deletes all routes for the prefix/peer combination
+func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) error {
 	d.lock("DeleteLiveRoute")
 	defer d.unlock()
 
@@ -1275,12 +1309,22 @@ func (d *Database) DeleteLiveRoute(prefix, peerIP string) error {
 	}()
 
 	// Literal per-table queries (rather than one formatted with the table name)
-	// so the delete carries no dynamically built SQL.
+	// so the delete carries no dynamically built SQL. A delete with no origin
+	// ASN can remove several rows.
 	var res sql.Result
-	if isV4 {
+	switch {
+	case isV4 && originASN == 0:
 		res, err = tx.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
-	} else {
+	case isV4:
+		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,7 +13,6 @@ 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
@@ -123,8 +122,8 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) {
 	lookup("2001:db9::1", "")
 
 	err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
-		{Prefix: "10.1.2.0/24", PeerIP: testPeerIP, IPVersion: ipVersionV4},
-		{Prefix: "2001:db8:1:2::/64", PeerIP: testPeerIP, IPVersion: ipVersionV6},
+		{Prefix: "10.1.2.0/24", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV4},
+		{Prefix: "2001:db8:1:2::/64", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV6},
 	})
 	if err != nil {
 		t.Fatalf("DeleteLiveRouteBatch: %v", err)
@@ -206,40 +205,6 @@ 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
@@ -341,6 +306,7 @@ 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, peerIP string) error
+	DeleteLiveRoute(prefix string, originASN int, 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,6 +114,7 @@ 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, peer_ip) -- a peer has at most one route for a prefix
+	UNIQUE(prefix, origin_asn, peer_ip)
 );
 
 -- 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, peer_ip) -- a peer has at most one route for a prefix
+	UNIQUE(prefix, origin_asn, peer_ip)
 );
 
 -- 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, peerIP string) error {
+func (m *mockStore) DeleteLiveRoute(prefix string, originASN int, peerIP string) error {
 	// Simple mock - just return nil
 	return nil
 }

internal/routewatch/ashandler.go

@@ -32,12 +32,15 @@ type ASHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// Batching
-	mu        sync.Mutex
-	batch     []asnOp
-	lastFlush time.Time
-	stopCh    chan struct{}
-	wg        sync.WaitGroup
+	// Batching. mu guards batch only: the flush goroutine takes the batch
+	// under mu and writes it after releasing mu, so HandleMessage never
+	// waits for a database write.
+	mu          sync.Mutex
+	batch       []asnOp
+	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
+	lastFlush   time.Time     // used only by the flush goroutine
+	stopCh      chan struct{}
+	wg          sync.WaitGroup
 }
 
 type asnOp struct {
@@ -52,11 +55,12 @@ type asnOp struct {
 // background goroutine is terminated.
 func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
 	h := &ASHandler{
-		db:        db,
-		logger:    logger,
-		batch:     make([]asnOp, 0, asnBatchSize),
-		lastFlush: time.Now(),
-		stopCh:    make(chan struct{}),
+		db:          db,
+		logger:      logger,
+		batch:       make([]asnOp, 0, asnBatchSize),
+		batchFullCh: make(chan struct{}, 1),
+		lastFlush:   time.Now(),
+		stopCh:      make(chan struct{}),
 	}
 
 	// Start the flush timer goroutine
@@ -85,8 +89,8 @@ func (h *ASHandler) QueueCapacity() int {
 // HandleMessage processes a RIS Live BGP message by extracting all ASNs from
 // the AS path and queuing them for batch insertion into the database. The
 // origin ASN (last element in the path) and all transit ASNs are recorded
-// with their associated timestamps. The batch is automatically flushed when
-// it reaches the configured size threshold.
+// with their associated timestamps. When the batch reaches the configured size
+// threshold, the flush goroutine is woken to write it.
 func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// Use the pre-parsed timestamp
 	timestamp := msg.ParsedTimestamp
@@ -118,13 +122,18 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 		}
 	}
 
-	// Check if we need to flush
+	// 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.
 	if len(h.batch) >= asnBatchSize {
-		h.flushBatchLocked()
+		select {
+		case h.batchFullCh <- struct{}{}:
+		default: // already woken
+		}
 	}
 }
 
-// flushLoop runs in a goroutine and periodically flushes batches
+// flushLoop runs in a goroutine, the only one that writes the batch: it
+// flushes when the batch is full, periodically, and once more on Stop.
 func (h *ASHandler) flushLoop() {
 	defer h.wg.Done()
 	ticker := time.NewTicker(asnBatchTimeout)
@@ -133,31 +142,46 @@ func (h *ASHandler) flushLoop() {
 	for {
 		select {
 		case <-ticker.C:
-			h.mu.Lock()
 			if time.Since(h.lastFlush) >= asnBatchTimeout {
-				h.flushBatchLocked()
+				h.flushBatch()
 			}
-			h.mu.Unlock()
+		case <-h.batchFullCh:
+			h.flushBatch()
 		case <-h.stopCh:
 			// Final flush
-			h.mu.Lock()
-			h.flushBatchLocked()
-			h.mu.Unlock()
+			h.flushBatch()
 
 			return
 		}
 	}
 }
 
-// flushBatchLocked flushes the ASN batch to the database (must be called with mutex held)
-func (h *ASHandler) flushBatchLocked() {
+// takeBatch returns the pending operations and leaves an empty batch in their
+// place for HandleMessage to fill.
+func (h *ASHandler) takeBatch() []asnOp {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
 	if len(h.batch) == 0 {
+		return nil
+	}
+	batch := h.batch
+	h.batch = make([]asnOp, 0, asnBatchSize)
+
+	return batch
+}
+
+// flushBatch takes the pending batch and writes it to the database, without
+// holding the mutex during the write.
+func (h *ASHandler) flushBatch() {
+	batch := h.takeBatch()
+	if len(batch) == 0 {
 		return
 	}
 
 	// Process ASNs first (deduped)
 	asnMap := make(map[int]time.Time)
-	for _, op := range h.batch {
+	for _, op := range batch {
 		if existing, ok := asnMap[op.number]; !ok || op.timestamp.After(existing) {
 			asnMap[op.number] = op.timestamp
 		}
@@ -168,8 +192,6 @@ func (h *ASHandler) flushBatchLocked() {
 		h.logger.Error("Failed to process ASN batch", "error", err, "count", len(asnMap))
 	}
 
-	// Clear batch
-	h.batch = h.batch[:0]
 	h.lastFlush = time.Now()
 }
 

internal/routewatch/batchwrite_test.go

@@ -0,0 +1,192 @@
+package routewatch
+
+import (
+	"slices"
+	"sync"
+	"testing"
+	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/database"
+	"git.eeqj.de/sneak/routewatch/internal/logger"
+	"git.eeqj.de/sneak/routewatch/internal/ristypes"
+)
+
+// handleMessageTimeout is how long a test waits for HandleMessage to return,
+// or for a write to start, before it fails.
+const handleMessageTimeout = 2 * time.Second
+
+// 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
+// first of them starts. It records what each write was given.
+type blockingStore struct {
+	*mockStore
+
+	release      chan struct{}
+	writeStarted chan struct{}
+	startOnce    sync.Once
+
+	mu       sync.Mutex
+	prefixes map[string]bool
+	asns     map[int]bool
+	peers    map[string]bool
+}
+
+func newBlockingStore() *blockingStore {
+	return &blockingStore{
+		mockStore:    newMockStore(),
+		release:      make(chan struct{}),
+		writeStarted: make(chan struct{}),
+		prefixes:     make(map[string]bool),
+		asns:         make(map[int]bool),
+		peers:        make(map[string]bool),
+	}
+}
+
+func (s *blockingStore) waitForRelease() {
+	s.startOnce.Do(func() { close(s.writeStarted) })
+	<-s.release
+}
+
+func (s *blockingStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
+	s.waitForRelease()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, route := range routes {
+		s.prefixes[route.Prefix] = true
+	}
+
+	return nil
+}
+
+func (s *blockingStore) GetOrCreateASNBatch(asns map[int]time.Time) error {
+	s.waitForRelease()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for asn := range asns {
+		s.asns[asn] = true
+	}
+
+	return nil
+}
+
+func (s *blockingStore) UpdatePeerBatch(peers map[string]database.PeerUpdate) error {
+	s.waitForRelease()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for peerIP := range peers {
+		s.peers[peerIP] = true
+	}
+
+	return nil
+}
+
+// returnsInTime reports whether f returns within handleMessageTimeout.
+func returnsInTime(f func()) bool {
+	done := make(chan struct{})
+	go func() {
+		f()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		return true
+	case <-time.After(handleMessageTimeout):
+		return false
+	}
+}
+
+// handleWhileAWriteIsBlocked hands h the messages in fill, which fill its
+// batch so that the handler writes it to store, waits until the write starts,
+// 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 close(store.release)
+
+	if !returnsInTime(func() {
+		for _, msg := range fill {
+			h.HandleMessage(msg)
+		}
+	}) {
+		t.Fatal("HandleMessage did not return after filling the batch: it waits for the database write")
+	}
+
+	select {
+	case <-store.writeStarted:
+	case <-time.After(handleMessageTimeout):
+		t.Fatal("the handler did not write its full batch")
+	}
+
+	if !returnsInTime(func() { h.HandleMessage(extra) }) {
+		t.Fatal("HandleMessage did not return while a database write was blocked")
+	}
+}
+
+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))},
+		announce([]string{extraPrefix}))
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if !store.prefixes[extraPrefix] {
+		t.Errorf("Stop did not write %s, handled while a write was blocked", extraPrefix)
+	}
+}
+
+func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
+	store := newBlockingStore()
+	h := NewASHandler(store, logger.New())
+	handleWhileAWriteIsBlocked(t, store, h,
+		[]*ristypes.RISMessage{{Path: slices.Repeat(ristypes.ASPath{testASNA}, asnBatchSize)}},
+		&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 while a write was blocked", testASNB)
+	}
+}
+
+func TestPeerHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
+	const otherPeerIP = "2001:db8:ffff::2"
+
+	store := newBlockingStore()
+	h := NewPeerHandler(store, logger.New())
+	handleWhileAWriteIsBlocked(t, store, h,
+		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
+		&ristypes.RISMessage{Peer: otherPeerIP})
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if !store.peers[otherPeerIP] {
+		t.Errorf("Stop did not write peer %s, handled while a write was blocked", otherPeerIP)
+	}
+}

internal/routewatch/peerhandler.go

@@ -35,12 +35,15 @@ type PeerHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// Batching
-	mu        sync.Mutex
-	peerBatch []peerUpdate
-	lastFlush time.Time
-	stopCh    chan struct{}
-	wg        sync.WaitGroup
+	// Batching. mu guards peerBatch only: the flush goroutine takes the batch
+	// under mu and writes it after releasing mu, so HandleMessage never
+	// waits for a database write.
+	mu          sync.Mutex
+	peerBatch   []peerUpdate
+	batchFullCh chan struct{} // wakes the flush goroutine when peerBatch is full
+	lastFlush   time.Time     // used only by the flush goroutine
+	stopCh      chan struct{}
+	wg          sync.WaitGroup
 }
 
 type peerUpdate struct {
@@ -56,11 +59,12 @@ type peerUpdate struct {
 // should be stopped by calling Stop when it is no longer needed.
 func NewPeerHandler(db database.Store, logger *logger.Logger) *PeerHandler {
 	h := &PeerHandler{
-		db:        db,
-		logger:    logger,
-		peerBatch: make([]peerUpdate, 0, peerBatchSize),
-		lastFlush: time.Now(),
-		stopCh:    make(chan struct{}),
+		db:          db,
+		logger:      logger,
+		peerBatch:   make([]peerUpdate, 0, peerBatchSize),
+		batchFullCh: make(chan struct{}, 1),
+		lastFlush:   time.Now(),
+		stopCh:      make(chan struct{}),
 	}
 
 	// Start the flush timer goroutine
@@ -87,8 +91,8 @@ func (h *PeerHandler) QueueCapacity() int {
 
 // HandleMessage processes a RIS message to track peer information.
 // It extracts the peer IP address and ASN from the message and adds
-// the update to an internal batch. When the batch reaches peerBatchSize
-// or the batch timeout expires, the batch is flushed to the database.
+// the update to an internal batch. The flush goroutine writes the batch to the
+// database when it reaches peerBatchSize or the batch timeout expires.
 func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// Parse peer ASN from string
 	peerASN := 0
@@ -109,13 +113,18 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 		timestamp:   msg.ParsedTimestamp,
 	})
 
-	// Check if we need to flush
+	// 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.
 	if len(h.peerBatch) >= peerBatchSize {
-		h.flushBatchLocked()
+		select {
+		case h.batchFullCh <- struct{}{}:
+		default: // already woken
+		}
 	}
 }
 
-// flushLoop runs in a goroutine and periodically flushes batches
+// flushLoop runs in a goroutine, the only one that writes the batch: it
+// flushes when the batch is full, periodically, and once more on Stop.
 func (h *PeerHandler) flushLoop() {
 	defer h.wg.Done()
 	ticker := time.NewTicker(peerBatchTimeout)
@@ -124,31 +133,46 @@ func (h *PeerHandler) flushLoop() {
 	for {
 		select {
 		case <-ticker.C:
-			h.mu.Lock()
 			if time.Since(h.lastFlush) >= peerBatchTimeout {
-				h.flushBatchLocked()
+				h.flushBatch()
 			}
-			h.mu.Unlock()
+		case <-h.batchFullCh:
+			h.flushBatch()
 		case <-h.stopCh:
 			// Final flush
-			h.mu.Lock()
-			h.flushBatchLocked()
-			h.mu.Unlock()
+			h.flushBatch()
 
 			return
 		}
 	}
 }
 
-// flushBatchLocked flushes the peer batch to the database (must be called with mutex held)
-func (h *PeerHandler) flushBatchLocked() {
+// takeBatch returns the pending updates and leaves an empty batch in their
+// place for HandleMessage to fill.
+func (h *PeerHandler) takeBatch() []peerUpdate {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
 	if len(h.peerBatch) == 0 {
+		return nil
+	}
+	batch := h.peerBatch
+	h.peerBatch = make([]peerUpdate, 0, peerBatchSize)
+
+	return batch
+}
+
+// flushBatch takes the pending batch and writes it to the database, without
+// holding the mutex during the write.
+func (h *PeerHandler) flushBatch() {
+	batch := h.takeBatch()
+	if len(batch) == 0 {
 		return
 	}
 
 	// Deduplicate by peer IP, keeping the latest update for each peer
 	peerMap := make(map[string]peerUpdate)
-	for _, update := range h.peerBatch {
+	for _, update := range batch {
 		if existing, ok := peerMap[update.peerIP]; !ok || update.timestamp.After(existing.timestamp) {
 			peerMap[update.peerIP] = update
 		}
@@ -170,8 +194,6 @@ func (h *PeerHandler) flushBatchLocked() {
 		h.logger.Error("Failed to process peer batch", "error", err, "count", len(dbPeerMap))
 	}
 
-	// Clear batch
-	h.peerBatch = h.peerBatch[:0]
 	h.lastFlush = time.Now()
 }
 

internal/routewatch/prefixhandler.go

@@ -40,12 +40,15 @@ type PrefixHandler struct {
 	logger  *logger.Logger
 	metrics *metrics.Tracker
 
-	// Batching
-	mu        sync.Mutex
-	batch     []prefixUpdate
-	lastFlush time.Time
-	stopCh    chan struct{}
-	wg        sync.WaitGroup
+	// Batching. mu guards batch only: the flush goroutine takes the batch
+	// under mu and writes it after releasing mu, so HandleMessage never
+	// waits for a database write.
+	mu          sync.Mutex
+	batch       []prefixUpdate
+	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
+	lastFlush   time.Time     // used only by the flush goroutine
+	stopCh      chan struct{}
+	wg          sync.WaitGroup
 }
 
 type prefixUpdate struct {
@@ -60,11 +63,12 @@ type prefixUpdate struct {
 // NewPrefixHandler creates a new batched prefix tracking handler
 func NewPrefixHandler(db database.Store, logger *logger.Logger) *PrefixHandler {
 	h := &PrefixHandler{
-		db:        db,
-		logger:    logger,
-		batch:     make([]prefixUpdate, 0, prefixBatchSize),
-		lastFlush: time.Now(),
-		stopCh:    make(chan struct{}),
+		db:          db,
+		logger:      logger,
+		batch:       make([]prefixUpdate, 0, prefixBatchSize),
+		batchFullCh: make(chan struct{}, 1),
+		lastFlush:   time.Now(),
+		stopCh:      make(chan struct{}),
 	}
 
 	// Start the flush timer goroutine
@@ -108,14 +112,14 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// 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 {
@@ -140,13 +144,18 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 		}
 	}
 
-	// Check if we need to flush
+	// 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.
 	if len(h.batch) >= prefixBatchSize {
-		h.flushBatchLocked()
+		select {
+		case h.batchFullCh <- struct{}{}:
+		default: // already woken
+		}
 	}
 }
 
-// flushLoop runs in a goroutine and periodically flushes batches
+// flushLoop runs in a goroutine, the only one that writes the batch: it
+// flushes when the batch is full, periodically, and once more on Stop.
 func (h *PrefixHandler) flushLoop() {
 	defer h.wg.Done()
 	ticker := time.NewTicker(prefixBatchTimeout)
@@ -155,30 +164,45 @@ func (h *PrefixHandler) flushLoop() {
 	for {
 		select {
 		case <-ticker.C:
-			h.mu.Lock()
 			if time.Since(h.lastFlush) >= prefixBatchTimeout {
-				h.flushBatchLocked()
+				h.flushBatch()
 			}
-			h.mu.Unlock()
+		case <-h.batchFullCh:
+			h.flushBatch()
 		case <-h.stopCh:
 			// Final flush
-			h.mu.Lock()
-			h.flushBatchLocked()
-			h.mu.Unlock()
+			h.flushBatch()
 
 			return
 		}
 	}
 }
 
-// flushBatchLocked flushes the prefix batch to the database (must be called with mutex held)
-func (h *PrefixHandler) flushBatchLocked() {
+// takeBatch returns the pending updates and leaves an empty batch in their
+// place for HandleMessage to fill.
+func (h *PrefixHandler) takeBatch() []prefixUpdate {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
 	if len(h.batch) == 0 {
+		return nil
+	}
+	batch := h.batch
+	h.batch = make([]prefixUpdate, 0, prefixBatchSize)
+
+	return batch
+}
+
+// flushBatch takes the pending batch and writes it to the database, without
+// holding the mutex during the write.
+func (h *PrefixHandler) flushBatch() {
+	batch := h.takeBatch()
+	if len(batch) == 0 {
 		return
 	}
 
 	startTime := time.Now()
-	batchSize := len(h.batch)
+	batchSize := len(batch)
 
 	// 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
@@ -186,7 +210,7 @@ func (h *PrefixHandler) flushBatchLocked() {
 	// updates within one second carry the same one.
 	type prefixAndPeer struct{ prefix, peer string }
 	lastUpdates := make(map[prefixAndPeer]prefixUpdate)
-	for _, update := range h.batch {
+	for _, update := range batch {
 		lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update
 	}
 
@@ -221,6 +245,7 @@ func (h *PrefixHandler) flushBatchLocked() {
 			// Create deletion record for batch delete
 			routesToDelete = append(routesToDelete, database.LiveRouteDeletion{
 				Prefix:    update.prefix,
+				OriginASN: update.originASN,
 				PeerIP:    update.peer,
 				IPVersion: ipVersion,
 			})
@@ -260,8 +285,6 @@ func (h *PrefixHandler) flushBatchLocked() {
 		"duration_ms", elapsed.Milliseconds(),
 	)
 
-	// Clear batch
-	h.batch = h.batch[:0]
 	h.lastFlush = time.Now()
 }
 
@@ -424,24 +447,51 @@ 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) {
-	if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil {
-		h.logger.Error("Failed to delete live route",
-			"prefix", update.prefix,
-			"peer", update.peer,
-			"error", err,
-		)
+	// For withdrawals, we need to delete the route from live_routes
+	if update.originASN > 0 {
+		if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
+			h.logger.Error("Failed to delete live route",
+				"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,
+			)
+		}
 	}
 }
 
 // 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) {
-	if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil {
-		h.logger.Error("Failed to delete live route",
-			"prefix", update.prefix,
-			"peer", update.peer,
-			"error", err,
-		)
+	// For withdrawals, we need to delete the route from live_routes
+	// Since we have the origin ASN from the update, we can delete the specific route
+	if update.originASN > 0 {
+		if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
+			h.logger.Error("Failed to delete live route",
+				"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,7 +3,6 @@ package routewatch
 import (
 	"context"
 	"errors"
-	"fmt"
 	"slices"
 	"testing"
 	"time"
@@ -31,11 +30,6 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
 
 	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
 	h := &PrefixHandler{db: db, logger: logger.New()}
-	flush := func() {
-		h.mu.Lock()
-		defer h.mu.Unlock()
-		h.flushBatchLocked()
-	}
 
 	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
 	h.HandleMessage(&ristypes.RISMessage{
@@ -47,7 +41,7 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
 			Prefixes: []string{"2001:db8:1::/48", "192.0.2.0/24"},
 		}},
 	})
-	flush()
+	h.flushBatch()
 
 	for ip, want := range map[string]string{
 		"2001:db8:1::1": "2001:db8:1::/48",
@@ -67,7 +61,7 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
 		ParsedTimestamp: ts.Add(time.Minute),
 		Withdrawals:     []string{"2001:db8:1:0:0:0:0:0/48", "192.0.2.0/24"},
 	})
-	flush()
+	h.flushBatch()
 
 	for _, ip := range []string{"2001:db8:1::1", "192.0.2.1"} {
 		if info, err := db.GetIPInfo(ip); !errors.Is(err, database.ErrNoRoute) {
@@ -165,9 +159,7 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
 				for _, msg := range batch {
 					h.HandleMessage(msg)
 				}
-				h.mu.Lock()
-				h.flushBatchLocked()
-				h.mu.Unlock()
+				h.flushBatch()
 			}
 
 			ctx := context.Background()
@@ -183,155 +175,32 @@ 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}}
 			}
-			assertCounts(t, db, ipv4Version, len(tt.wantPeers), wantDistribution)
+			queryDistributionV4, _, err := db.GetPrefixDistributionContext(ctx)
+			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.mu.Lock()
-					h.flushBatchLocked()
-					h.mu.Unlock()
-				}
-
-				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)
-	}
-}