Keep a peer-down until its route delete succeeds (closes #101)

When the delete of a peer's live routes after its session went down
failed, the down was dropped and the routes stayed. The prefix handler
now keeps such a peer, with the updates it sent after the down, for the
next flush, which retries the delete before it writes them; each
failure is logged. The delete runs in chunks of at most 10,000 routes
per table, taken in prefix order through the peer's key, each chunk in
its own transaction, so the database lock is released between chunks
and a failure undoes at most one chunk.

Model: opus-5-5

internal/database/counts_internal_test.go

@@ -2,6 +2,7 @@ package database
 
 import (
 	"context"
+	"fmt"
 	"slices"
 	"sync"
 	"testing"
@@ -494,6 +495,116 @@ 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)
+
+	var peerRoutes, otherPeerRoutes []*LiveRoute
+
+	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)
+		}
+	}
+
+	ctx := context.Background()
+	checkRoutes := func(when string, wantRoutes int) {
+		t.Helper()
+
+		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)
+		}
+
+		// The shared prefixes went with the first chunk, so each route left
+		// has a prefix of its own.
+		want := []PrefixDistribution{{MaskLength: 24, Count: wantRoutes}}
+		assertDistribution(t, when+": IPv4 distribution",
+			stats.IPv4PrefixDistribution, want)
+		assertDistribution(t, when+": IPv4 distribution query",
+			queryDistribution, want)
+	}
+
+	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)
+	}
+
+	checkRoutes("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))
+	}
+
+	checkRoutes("after DeleteLiveRoutesOfPeer", sharedPrefixes)
+}
+
 // 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,56 +658,83 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 	return nil
 }
 
-// Deleting all live routes of a peer. The first query of each table selects the
+// peerRouteChunkSize is the most routes of a peer that one transaction of
-// mask length of each of the peer's routes whose prefix no other peer has a
+// DeleteLiveRoutesOfPeer deletes from each route table. A peer can have a
-// live route for: the delete takes those prefixes out of the prefix
+// million routes; deleted in one transaction, they would hold the database lock
-// distribution.
+// for most of a minute.
-const (
+const peerRouteChunkSize = 10000
-	peerOnlyRouteMaskLengthsV4SQL = `SELECT r.mask_length
-		FROM live_routes_v4 AS r WHERE r.peer_ip = ?
-		AND NOT EXISTS (SELECT 1 FROM live_routes_v4 AS o
-		WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)`
-	deletePeerLiveRoutesV4SQL = `DELETE FROM live_routes_v4 WHERE peer_ip = ?`
 
-	peerOnlyRouteMaskLengthsV6SQL = `SELECT r.mask_length
+// Deleting a chunk of a peer's live routes. The first query of each table
+// selects the peer's first routes in prefix order, through the
+// UNIQUE(peer_ip, prefix) key, each with its mask length and whether another
+// peer has a live route for its prefix: the delete takes the prefixes no other
+// peer has out of the prefix distribution. The second query deletes the peer's
+// routes up to the last prefix the first one selected.
+const (
+	peerRouteChunkV4SQL = `SELECT r.prefix, r.mask_length,
+		EXISTS (SELECT 1 FROM live_routes_v4 AS o
+		WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)
+		FROM live_routes_v4 AS r WHERE r.peer_ip = ?
+		ORDER BY r.prefix LIMIT ?`
+	deletePeerRouteChunkV4SQL = `DELETE FROM live_routes_v4
+		WHERE peer_ip = ? AND prefix <= ?`
+
+	peerRouteChunkV6SQL = `SELECT r.prefix, r.mask_length,
+		EXISTS (SELECT 1 FROM live_routes_v6 AS o
+		WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)
 		FROM live_routes_v6 AS r WHERE r.peer_ip = ?
-		AND NOT EXISTS (SELECT 1 FROM live_routes_v6 AS o
+		ORDER BY r.prefix LIMIT ?`
-		WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)`
+	deletePeerRouteChunkV6SQL = `DELETE FROM live_routes_v6
-	deletePeerLiveRoutesV6SQL = `DELETE FROM live_routes_v6 WHERE peer_ip = ?`
+		WHERE peer_ip = ? AND prefix <= ?`
 )
 
-// deletePeerRoutes deletes all live routes of peerIP from one table, given that
+// deletePeerRoutes deletes the first peerRouteChunkSize live routes of peerIP,
-// table's two statements above. It returns how many routes it deleted and the
+// or as many as it has, from one table, given that table's two statements
-// mask length of each prefix it left with no live route.
+// above. It returns how many routes it deleted and the mask length of each
+// prefix it left with no live route.
 func deletePeerRoutes(
-	ctx context.Context, tx *loggingTx,
+	ctx context.Context, tx *loggingTx, chunkSQL, deleteSQL, peerIP string,
-	onlyRouteMaskLengthsSQL, deleteSQL, peerIP string,
 ) (int64, []int, error) {
-	rows, err := tx.QueryContext(ctx, onlyRouteMaskLengthsSQL, peerIP)
+	rows, err := tx.QueryContext(ctx, chunkSQL, peerIP, peerRouteChunkSize)
 	if err != nil {
 		return 0, nil, err
 	}
 	defer func() { _ = rows.Close() }()
 
-	var goneMaskLengths []int
+	var (
+		selected        int
+		lastPrefix      string
+		goneMaskLengths []int
+	)
 
 	for rows.Next() {
-		var maskLength int
+		var (
+			maskLength         int
+			otherPeerHasARoute bool
+		)
 
-		err := rows.Scan(&maskLength)
+		err := rows.Scan(&lastPrefix, &maskLength, &otherPeerHasARoute)
 		if err != nil {
 			return 0, nil, err
 		}
 
+		selected++
+
+		if !otherPeerHasARoute {
 			goneMaskLengths = append(goneMaskLengths, maskLength)
 		}
+	}
 
 	err = rows.Err()
 	if err != nil {
 		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
 	}
@@ -720,44 +747,66 @@ func deletePeerRoutes(
 	return deleted, goneMaskLengths, nil
 }
 
-// DeleteLiveRoutesOfPeer deletes every live route of the peer, IPv4 and IPv6,
+// deletePeerRouteChunk deletes, in one transaction, the first
-// in one transaction, as when the peer's BGP session goes down. It returns how
+// peerRouteChunkSize IPv4 and the first peerRouteChunkSize IPv6 live routes of
-// many routes it deleted.
+// the peer, or as many as it has. It returns how many of each it deleted.
-func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
+func (d *Database) deletePeerRouteChunk(peerIP string) (int, int, error) {
 	d.lock("DeleteLiveRoutesOfPeer")
 	defer d.unlock()
 
 	tx, err := d.beginTx()
 	if err != nil {
-		return 0, fmt.Errorf("failed to begin transaction: %w", err)
+		return 0, 0, fmt.Errorf("failed to begin transaction: %w", err)
 	}
 	defer d.rollback(tx)
 
 	ctx := context.Background()
 
 	deletedV4, goneMaskLengthsV4, err := deletePeerRoutes(ctx, tx,
-		peerOnlyRouteMaskLengthsV4SQL, deletePeerLiveRoutesV4SQL, peerIP)
+		peerRouteChunkV4SQL, deletePeerRouteChunkV4SQL, peerIP)
 	if err != nil {
-		return 0, fmt.Errorf("failed to delete the IPv4 routes of peer %s: %w",
+		return 0, 0, fmt.Errorf("failed to delete the IPv4 routes of peer %s: %w",
 			peerIP, err)
 	}
 
 	deletedV6, goneMaskLengthsV6, err := deletePeerRoutes(ctx, tx,
-		peerOnlyRouteMaskLengthsV6SQL, deletePeerLiveRoutesV6SQL, peerIP)
+		peerRouteChunkV6SQL, deletePeerRouteChunkV6SQL, peerIP)
 	if err != nil {
-		return 0, fmt.Errorf("failed to delete the IPv6 routes of peer %s: %w",
+		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, fmt.Errorf("failed to commit transaction: %w", err)
+		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 + deletedV6), nil
+	return int(deletedV4), int(deletedV6), nil
+}
+
+// DeleteLiveRoutesOfPeer deletes every live route of the peer, IPv4 and IPv6,
+// as when the peer's BGP session goes down. It deletes them a chunk at a time,
+// each 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) {
+	total := 0
+
+	for {
+		deletedV4, deletedV6, err := d.deletePeerRouteChunk(peerIP)
+		if err != nil {
+			return total, err
+		}
+
+		total += deletedV4 + deletedV6
+
+		if deletedV4 < peerRouteChunkSize && deletedV6 < peerRouteChunkSize {
+			return total, nil
+		}
+	}
 }
 
 // insertPrefixesInTable inserts the prefixes, all of one IP version, that the

internal/routewatch/prefixhandler.go

@@ -54,7 +54,7 @@ type PrefixHandler struct {
 	mu    sync.Mutex
 	batch []prefixUpdate
 	// peersDown holds the peers whose BGP session went down since the batch
-	// was taken.
+	// 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.
@@ -269,8 +269,12 @@ func (h *PrefixHandler) flushBatch() {
 	batchSize := len(batch)
 
 	// The routes of a peer whose session went down are deleted first: the
-	// batch holds only the updates the peer sent after the down.
+	// batch holds only the updates the peer sent after the down. A peer whose
-	h.deleteLiveRoutesOfPeers(peersDown)
+	// delete failed is kept, with those updates, for the next flush to retry.
+	failedPeers := h.deleteLiveRoutesOfPeers(peersDown)
+	if len(failedPeers) > 0 {
+		batch = h.keepForNextFlush(batch, failedPeers)
+	}
 
 	// For each prefix and peer, keep only the last update the feed delivered.
 	// Updates from different peers for the same prefix are all kept. Arrival
@@ -344,15 +348,21 @@ func (h *PrefixHandler) flushBatch() {
 }
 
 // deleteLiveRoutesOfPeers deletes all live routes of each of peers, the peers
-// whose BGP session went down. Most downs come from peers that have no routes,
+// whose BGP session went down, and returns those whose delete failed. Most
-// so only a delete that removed routes is logged.
+// downs come from peers that have no routes, so only a delete that removed
-func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) {
+// routes is logged.
+func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) []string {
+	var failed []string
+
 	for _, peer := range peers {
 		deleted, err := h.db.DeleteLiveRoutesOfPeer(peer)
 		if err != nil {
 			h.logger.Error(
-				"Failed to delete the live routes of a peer whose session went down",
+				"Failed to delete the live routes of a peer whose session "+
-				"peer", peer, "error", err)
+					"went down; the next flush retries",
+				"peer", peer, "routes_deleted", deleted, "error", err)
+
+			failed = append(failed, peer)
 
 			continue
 		}
@@ -363,6 +373,40 @@ func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) {
 				"peer", peer, "routes", deleted)
 		}
 	}
+
+	return failed
+}
+
+// keepForNextFlush puts the peers whose route delete failed back in front of
+// peersDown, and their updates from batch back in front of the batch, 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. It returns the updates of the
+// other peers, for this flush to write.
+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)
+	})
+	h.batch = slices.Concat(kept, h.batch)
+	h.peersDown = slices.Concat(failedPeers, h.peersDown)
+
+	return toWrite
 }
 
 // writeBatch writes the collected route upserts, route deletions and prefix

internal/routewatch/prefixhandler_internal_test.go

@@ -560,6 +560,124 @@ func TestPrefixHandlerHandlesADownWhileAFlushWrites(t *testing.T) {
 	})
 }
 
+// errTestPeerDelete is the error firstPeerDeleteFailsStore's first delete
+// 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 second case the peer's session
+// goes down again, and the peer announces another prefix, while the delete
+// fails: the announcement sent between the two downs must not be written.
+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
+	)
+
+	tests := []struct {
+		name string
+		// Handled while the delete fails, if not nil.
+		whileFailing []*ristypes.RISMessage
+		wantDeletes  int
+		// The peers left with a live route for each prefix.
+		wantPeers map[string][]string
+	}{
+		{
+			name:        "the delete fails once",
+			wantDeletes: 2,
+			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),
+			},
+			// The retry and the delete for the second down.
+			wantDeletes: 3,
+			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
+			h.flushBatch()
+
+			if store.deletes != tt.wantDeletes {
+				t.Errorf("the store was asked for %d deletes of the peer's "+
+					"routes, want %d", store.deletes, tt.wantDeletes)
+			}
+
+			assertPeers(t, db, tt.wantPeers)
+			assertCounts(t, db, ipv4Version, 1, []database.PrefixDistribution{
+				{MaskLength: maskLength, Count: 1},
+			})
+		})
+	}
+}
+
 // newTestDatabase opens a database in a new temporary directory. It is closed
 // when the test ends.
 func newTestDatabase(t *testing.T) *database.Database {