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'
-  batches take at most about 0.25 GiB of the Go heap's 1.5 GiB.
+  drops messages as above. When the delete of a peer's live routes after its
+  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

@@ -658,48 +658,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
-// mask length of each of the peer's routes whose prefix no other peer has a
-// live route for: the delete takes those prefixes out of the prefix
-// distribution.
-const (
-	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 = ?`
+// peerRouteChunkSize is the most routes of a peer that one transaction of
+// DeleteLiveRoutesOfPeer deletes from each route table. A peer can have a
+// million routes; deleted in one transaction, they would hold the database lock
+// for most of a minute.
+const peerRouteChunkSize = 10000
 
-	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
-		WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)`
-	deletePeerLiveRoutesV6SQL = `DELETE FROM live_routes_v6 WHERE peer_ip = ?`
+		ORDER BY r.prefix LIMIT ?`
+	deletePeerRouteChunkV6SQL = `DELETE FROM live_routes_v6
+		WHERE peer_ip = ? AND prefix <= ?`
 )
 
-// deletePeerRoutes deletes all live routes of peerIP from one table, given that
-// table's two statements above. It returns how many routes it deleted and the
-// mask length of each prefix it left with no live route.
+// deletePeerRoutes deletes the first peerRouteChunkSize live routes of peerIP,
+// or as many as it has, from one table, given that table's two statements
+// 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,
-	onlyRouteMaskLengthsSQL, deleteSQL, peerIP string,
+	ctx context.Context, tx *loggingTx, chunkSQL, 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 +730,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 +748,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
-// many routes it deleted.
+// as when the peer's BGP session goes down. It deletes them a chunk at a time
+// 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")
-	defer d.unlock()
+	total := 0
 
-	tx, err := d.beginTx()
-	if err != nil {
-		return 0, fmt.Errorf("failed to begin transaction: %w", err)
+	for {
+		deletedV4, deletedV6, err := d.deletePeerRouteChunk(peerIP)
+		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
@@ -2332,6 +2342,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/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
-	// both under mu and writes them after releasing mu, so HandleMessage
-	// waits for a database write only while batch is at prefixBatchMaxSize.
+	// Batching. mu guards batch, held and peersDown only: the flush goroutine
+	// takes them under mu and writes them after releasing mu, so
+	// HandleMessage waits for a database write only while batch is at
+	// prefixBatchMaxSize.
 	mu    sync.Mutex
 	batch []prefixUpdate
-	// peersDown holds the peers whose BGP session went down since the batch
-	// was taken.
+	// held holds the updates that the peers whose route delete failed in the
+	// 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,
-// so they are dropped; updates the peer sends later stay in the batch and are
-// written after the delete.
+// peer's updates still in the batch or held came before the down, which
+// replaces them, so they are dropped; updates the peer sends later stay in the
+// 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 {
-		return update.peer == peer
-	})
-	h.peersDown = append(h.peersDown, peer)
+	isPeers := func(update prefixUpdate) bool { return update.peer == peer }
+	h.batch = slices.DeleteFunc(h.batch, isPeers)
+	h.held = slices.DeleteFunc(h.held, isPeers)
+
+	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
-// since the last flush, and leaves an empty batch and no peers in their place
-// for HandleMessage to fill.
+// takeBatch returns the pending updates, the held ones first, and the peers in
+// peersDown, and leaves an empty batch, no held updates and no peers in their
+// 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.
-	h.deleteLiveRoutesOfPeers(peersDown)
+	// batch holds only the updates the peer sent after the down. A peer whose
+	// 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,
-// so only a delete that removed routes is logged.
-func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) {
+// whose BGP session went down, and returns those whose delete failed. Most
+// downs come from peers that have no routes, so only a delete that removed
+// 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",
-				"peer", peer, "error", err)
+				"Failed to delete the live routes of a peer whose session "+
+					"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 {