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 10,000 routes taken in prefix order through the peer's key, each in its own transaction, so the database lock is released between chunks and a failure undoes at most one chunk. Model: opus-5-5
2026-10-19 20:42:40 +00:00
parent ef32082339
commit d64ab03d7c
4 changed files with 389 additions and 79 deletions
@@ -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)
+	}
+
+	deleted, err := db.deletePeerRouteChunk(peerIP, ipVersionV4)
+	if err != nil {
+		t.Fatalf("deletePeerRouteChunk: %v", err)
+	}
+
+	if deleted != peerRouteChunkSize {
+		t.Errorf("deletePeerRouteChunk deleted %d routes, want %d",
+			deleted, 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.
@@ -658,72 +658,44 @@ 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 DeleteLiveRoutesOfPeer
+// deletes in one transaction. 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.
-func deletePeerRoutes(
-	ctx context.Context, tx *loggingTx,
-	onlyRouteMaskLengthsSQL, deleteSQL, peerIP string,
-) (int64, []int, error) {
-	rows, err := tx.QueryContext(ctx, onlyRouteMaskLengthsSQL, peerIP)
-	if err != nil {
-		return 0, nil, err
-	}
-	defer func() { _ = rows.Close() }()
-
-	var goneMaskLengths []int
-
-	for rows.Next() {
-		var maskLength int
-
-		err := rows.Scan(&maskLength)
-		if err != nil {
-			return 0, nil, err
-		}
-
-		goneMaskLengths = append(goneMaskLengths, maskLength)
+// deletePeerRouteChunk deletes, in a transaction of its own, the first
+// peerRouteChunkSize live routes of peerIP, or as many as it has, from the
+// route table of one IP version. It returns how many routes it deleted.
+func (d *Database) deletePeerRouteChunk(peerIP string, ipVersion int) (int, error) {
+	chunkSQL, deleteSQL := peerRouteChunkV4SQL, deletePeerRouteChunkV4SQL
+	if ipVersion == ipVersionV6 {
+		chunkSQL, deleteSQL = peerRouteChunkV6SQL, deletePeerRouteChunkV6SQL
 	}

-	err = rows.Err()
-	if err != nil {
-		return 0, nil, err
-	}
-
-	res, err := tx.ExecContext(ctx, deleteSQL, peerIP)
-	if err != nil {
-		return 0, nil, err
-	}
-
-	deleted, err := res.RowsAffected()
-	if err != nil {
-		return 0, nil, err
-	}
-
-	return deleted, goneMaskLengths, nil
-}
-
-// 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.
-func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
 	d.lock("DeleteLiveRoutesOfPeer")
 	defer d.unlock()

@@ -735,18 +707,50 @@ func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {

 	ctx := context.Background()

-	deletedV4, goneMaskLengthsV4, err := deletePeerRoutes(ctx, tx,
-		peerOnlyRouteMaskLengthsV4SQL, deletePeerLiveRoutesV4SQL, peerIP)
+	rows, err := tx.QueryContext(ctx, chunkSQL, peerIP, peerRouteChunkSize)
 	if err != nil {
-		return 0, fmt.Errorf("failed to delete the IPv4 routes of peer %s: %w",
-			peerIP, err)
+		return 0, err
+	}
+	defer func() { _ = rows.Close() }()
+
+	var (
+		lastPrefix      string
+		goneMaskLengths []int
+	)
+
+	for rows.Next() {
+		var (
+			maskLength      int
+			otherPeerHasOne bool
+		)
+
+		err := rows.Scan(&lastPrefix, &maskLength, &otherPeerHasOne)
+		if err != nil {
+			return 0, err
+		}
+
+		if !otherPeerHasOne {
+			goneMaskLengths = append(goneMaskLengths, maskLength)
+		}
 	}

-	deletedV6, goneMaskLengthsV6, err := deletePeerRoutes(ctx, tx,
-		peerOnlyRouteMaskLengthsV6SQL, deletePeerLiveRoutesV6SQL, peerIP)
+	err = rows.Err()
 	if err != nil {
-		return 0, fmt.Errorf("failed to delete the IPv6 routes of peer %s: %w",
-			peerIP, err)
+		return 0, err
+	}
+
+	if lastPrefix == "" {
+		return 0, nil // the peer has no routes left in this table
+	}
+
+	res, err := tx.ExecContext(ctx, deleteSQL, peerIP, lastPrefix)
+	if err != nil {
+		return 0, err
+	}
+
+	deleted, err := res.RowsAffected()
+	if err != nil {
+		return 0, err
 	}

 	err = tx.Commit()
@@ -754,10 +758,43 @@ func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
 		return 0, fmt.Errorf("failed to commit transaction: %w", err)
 	}

-	d.counts.addRoutes(-int(deletedV4), -int(deletedV6))
-	d.counts.addToDistribution(goneMaskLengthsV4, goneMaskLengthsV6, -1)
+	if ipVersion == ipVersionV6 {
+		d.counts.addRoutes(0, -int(deleted))
+		d.counts.addToDistribution(nil, goneMaskLengths, -1)
+	} else {
+		d.counts.addRoutes(-int(deleted), 0)
+		d.counts.addToDistribution(goneMaskLengths, nil, -1)
+	}

-	return int(deletedV4 + deletedV6), nil
+	return int(deleted), 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 of
+// peerRouteChunkSize 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 _, ipVersion := range []int{ipVersionV4, ipVersionV6} {
+		for {
+			deleted, err := d.deletePeerRouteChunk(peerIP, ipVersion)
+			if err != nil {
+				return total, fmt.Errorf(
+					"failed to delete the IPv%d routes of peer %s: %w",
+					ipVersion, peerIP, err)
+			}
+
+			total += deleted
+
+			if deleted < peerRouteChunkSize {
+				break
+			}
+		}
+	}
+
+	return total, nil
 }

 // insertPrefixesInTable inserts the prefixes, all of one IP version, that the