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/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.