Make live route writes cheaper so the prefix handler keeps up (closes #96)

The live route tables lose their random UUID key: the row id, which the prefix-length sample draws, is now declared as the integer key. The prefix tables are keyed by the prefix, without the UUID or the duplicate index. A route update sets only the columns that change, so SQLite no longer rewrites the mask length indexes on every update, nor the origin index while the origin AS stays the same. Each kept index names the queries that use it. BenchmarkPrefixBatchOnFiveMillionRoutes, run by hand, times one 25,000-update batch on 5 million routes. Model: opus-5-5
2026-10-19 11:58:03 +00:00
parent 5806ddaef7
commit 1f612642e3
8 changed files with 439 additions and 149 deletions
@@ -19,7 +19,6 @@ import (
 	"git.eeqj.de/sneak/routewatch/internal/config"
 	"git.eeqj.de/sneak/routewatch/internal/logger"
 	"git.eeqj.de/sneak/routewatch/pkg/asinfo"
-	"github.com/google/uuid"
 	_ "github.com/mattn/go-sqlite3" // CGO SQLite driver
 )

@@ -234,30 +233,42 @@ func (d *Database) Close() error {
 	return d.db.Close()
 }

-// 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.
+// A live-route upsert sets the origin AS of the peer's route for the prefix
+// when the announcement changes it, then updates the rest of the route, and
+// inserts the route only when that update matched no row. 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 statements.
+//
+// SQLite rewrites a row's entry in every index on a column an UPDATE sets, even
+// when the value stays the same, so the update sets only the columns that
+// change: mask_length follows from the prefix, and origin_asn has a statement
+// of its own that changes nothing when the origin AS stays the same, as it
+// mostly does.
 const (
+	updateLiveRouteOriginV4SQL = `UPDATE live_routes_v4 SET origin_asn = ?
+		WHERE prefix = ? AND peer_ip = ? AND origin_asn != ?`
+
 	updateLiveRouteV4SQL = `UPDATE live_routes_v4
-		SET mask_length = ?, origin_asn = ?, as_path = ?, next_hop = ?,
-		last_updated = ? WHERE prefix = ? AND peer_ip = ?`
+		SET as_path = ?, next_hop = ?, last_updated = ?
+		WHERE prefix = ? AND peer_ip = ?`

 	insertLiveRouteV4SQL = `INSERT INTO live_routes_v4
-		(id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
-		last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
+		(prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
+		last_updated) VALUES (?, ?, ?, ?, ?, ?, ?)`
+
+	updateLiveRouteOriginV6SQL = `UPDATE live_routes_v6 SET origin_asn = ?
+		WHERE prefix = ? AND peer_ip = ? AND origin_asn != ?`

 	updateLiveRouteV6SQL = `UPDATE live_routes_v6
-		SET mask_length = ?, origin_asn = ?, as_path = ?, next_hop = ?,
-		last_updated = ? WHERE prefix = ? AND peer_ip = ?`
+		SET as_path = ?, next_hop = ?, last_updated = ?
+		WHERE prefix = ? AND peer_ip = ?`

 	insertLiveRouteV6SQL = `INSERT INTO live_routes_v6
-		(id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
-		last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
+		(prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
+		last_updated) VALUES (?, ?, ?, ?, ?, ?, ?)`
 )

 // Before a new route is inserted and after a route is deleted, the write looks
@@ -278,15 +289,22 @@ const (
 		WHERE prefix = ? AND peer_ip = ?`
 )

-// upsertRouteRow updates a live route with the update statement of its IP
-// version's table, inserting it when no row matched. It reports whether a new
-// row was inserted and whether that row is the first live route for its prefix.
+// upsertRouteRow updates a live route with the origin and update statements of
+// its IP version's table, inserting it when no row matched. It reports whether
+// a new row was inserted and whether that row is the first live route for its
+// prefix.
 func upsertRouteRow(
-	ctx context.Context, upd, ins, has *sql.Stmt, route *LiveRoute,
+	ctx context.Context, setOrigin, upd, ins, has *sql.Stmt, route *LiveRoute,
 	pathJSON string,
 ) (bool, bool, error) {
-	res, err := upd.ExecContext(ctx, route.MaskLength, route.OriginASN, pathJSON,
-		route.NextHop, route.LastUpdated, route.Prefix, route.PeerIP)
+	_, err := setOrigin.ExecContext(ctx, route.OriginASN, route.Prefix,
+		route.PeerIP, route.OriginASN)
+	if err != nil {
+		return false, false, err
+	}
+
+	res, err := upd.ExecContext(ctx, pathJSON, route.NextHop, route.LastUpdated,
+		route.Prefix, route.PeerIP)
 	if err != nil {
 		return false, false, err
 	}
@@ -307,9 +325,9 @@ func upsertRouteRow(
 		return false, false, err
 	}

-	_, err = ins.ExecContext(ctx, route.ID.String(), route.Prefix,
-		route.MaskLength, route.OriginASN, route.PeerIP, pathJSON,
-		route.NextHop, route.LastUpdated)
+	_, err = ins.ExecContext(ctx, route.Prefix, route.MaskLength,
+		route.OriginASN, route.PeerIP, pathJSON, route.NextHop,
+		route.LastUpdated)
 	if err != nil {
 		return false, false, err
 	}
@@ -327,13 +345,23 @@ func upsertRoutesInTable(
 		return 0, nil, nil
 	}

-	family, updateSQL, insertSQL, lookupSQL := ipv4Family, updateLiveRouteV4SQL,
-		insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
+	family, originSQL, updateSQL := ipv4Family, updateLiveRouteOriginV4SQL,
+		updateLiveRouteV4SQL
+	insertSQL, lookupSQL := insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
+
 	if ipVersion == ipVersionV6 {
-		family, updateSQL, insertSQL, lookupSQL = ipv6Family,
-			updateLiveRouteV6SQL, insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
+		family, originSQL, updateSQL = ipv6Family, updateLiveRouteOriginV6SQL,
+			updateLiveRouteV6SQL
+		insertSQL, lookupSQL = insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
 	}

+	setOrigin, err := tx.PrepareContext(ctx, originSQL)
+	if err != nil {
+		return 0, nil, fmt.Errorf("failed to prepare %s origin statement: %w",
+			family, err)
+	}
+	defer func() { _ = setOrigin.Close() }()
+
 	upd, err := tx.PrepareContext(ctx, updateSQL)
 	if err != nil {
 		return 0, nil, fmt.Errorf("failed to prepare %s update statement: %w",
@@ -365,8 +393,8 @@ func upsertRoutesInTable(
 			return 0, nil, fmt.Errorf("failed to encode AS path: %w", err)
 		}

-		inserted, newPrefix, err := upsertRouteRow(ctx, upd, ins, has, route,
-			string(pathJSON))
+		inserted, newPrefix, err := upsertRouteRow(ctx, setOrigin, upd, ins, has,
+			route, string(pathJSON))
 		if err != nil {
 			return 0, nil, fmt.Errorf("failed to upsert route %s: %w",
 				route.Prefix, err)
@@ -684,39 +712,33 @@ func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
 }

 // upsertPrefix updates the last_seen time of prefix with the statements of its
-// IP version's table, inserting the prefix when it is new. It reports whether
-// it inserted it.
+// IP version's table, inserting the prefix when the update matched no row. It
+// reports whether it inserted it.
 func upsertPrefix(
-	ctx context.Context, selectStmt, updateStmt, insertStmt *sql.Stmt,
+	ctx context.Context, updateStmt, insertStmt *sql.Stmt,
 	prefix string, timestamp time.Time,
 ) (bool, error) {
-	var id string
-
-	err := selectStmt.QueryRowContext(ctx, prefix).Scan(&id)
-
-	switch {
-	case err == nil:
-		// Prefix exists, update last_seen
-		_, err = updateStmt.ExecContext(ctx, timestamp, prefix)
-		if err != nil {
-			return false, fmt.Errorf("failed to update prefix %s: %w",
-				prefix, err)
-		}
-
-		return false, nil
-	case errors.Is(err, sql.ErrNoRows):
-		// Prefix doesn't exist, create it
-		_, err = insertStmt.ExecContext(ctx, generateUUID().String(), prefix,
-			timestamp, timestamp)
-		if err != nil {
-			return false, fmt.Errorf("failed to insert prefix %s: %w",
-				prefix, err)
-		}
-
-		return true, nil
-	default:
-		return false, fmt.Errorf("failed to query prefix %s: %w", prefix, err)
+	res, err := updateStmt.ExecContext(ctx, timestamp, prefix)
+	if err != nil {
+		return false, fmt.Errorf("failed to update prefix %s: %w", prefix, err)
 	}
+
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("failed to count updated prefix %s: %w",
+			prefix, err)
+	}
+
+	if affected > 0 {
+		return false, nil
+	}
+
+	_, err = insertStmt.ExecContext(ctx, prefix, timestamp, timestamp)
+	if err != nil {
+		return false, fmt.Errorf("failed to insert prefix %s: %w", prefix, err)
+	}
+
+	return true, nil
 }

 // upsertPrefixesInTable updates the last_seen time of prefixes, all of one IP
@@ -731,26 +753,17 @@ func upsertPrefixesInTable(
 	}

 	family := ipv4Family
-	selectSQL := "SELECT id FROM prefixes_v4 WHERE prefix = ?"
 	updateSQL := "UPDATE prefixes_v4 SET last_seen = ? WHERE prefix = ?"
-	insertSQL := "INSERT INTO prefixes_v4 (id, prefix, first_seen, last_seen) " +
-		"VALUES (?, ?, ?, ?)"
+	insertSQL := "INSERT INTO prefixes_v4 (prefix, first_seen, last_seen) " +
+		"VALUES (?, ?, ?)"

 	if ipVersion == ipVersionV6 {
 		family = ipv6Family
-		selectSQL = "SELECT id FROM prefixes_v6 WHERE prefix = ?"
 		updateSQL = "UPDATE prefixes_v6 SET last_seen = ? WHERE prefix = ?"
-		insertSQL = "INSERT INTO prefixes_v6 (id, prefix, first_seen, last_seen) " +
-			"VALUES (?, ?, ?, ?)"
+		insertSQL = "INSERT INTO prefixes_v6 (prefix, first_seen, last_seen) " +
+			"VALUES (?, ?, ?)"
 	}

-	selectStmt, err := tx.PrepareContext(ctx, selectSQL)
-	if err != nil {
-		return 0, fmt.Errorf("failed to prepare %s select statement: %w",
-			family, err)
-	}
-	defer func() { _ = selectStmt.Close() }()
-
 	updateStmt, err := tx.PrepareContext(ctx, updateSQL)
 	if err != nil {
 		return 0, fmt.Errorf("failed to prepare %s update statement: %w",
@@ -768,8 +781,8 @@ func upsertPrefixesInTable(
 	var newPrefixes int

 	for prefix, timestamp := range prefixes {
-		inserted, err := upsertPrefix(ctx, selectStmt, updateStmt, insertStmt,
-			prefix, timestamp)
+		inserted, err := upsertPrefix(ctx, updateStmt, insertStmt, prefix,
+			timestamp)
 		if err != nil {
 			return 0, err
 		}
@@ -1047,22 +1060,19 @@ func (d *Database) GetOrCreatePrefix(
 		tableName = "prefixes_v6"
 	}

-	var (
-		p     Prefix
-		idStr string
-	)
+	var p Prefix

-	query := fmt.Sprintf("SELECT id, prefix, first_seen, last_seen "+
+	query := fmt.Sprintf("SELECT prefix, first_seen, last_seen "+
 		"FROM %s WHERE prefix = ?", tableName)

-	err = tx.QueryRow(query, prefix).Scan(&idStr, &p.Prefix, &p.FirstSeen, &p.LastSeen)
+	err = tx.QueryRow(query, prefix).Scan(&p.Prefix, &p.FirstSeen, &p.LastSeen)
 	if err == nil {
 		// Prefix exists, update last_seen
-		p.ID, _ = uuid.Parse(idStr)
 		p.IPVersion = ipVersion
-		updateQuery := fmt.Sprintf("UPDATE %s SET last_seen = ? WHERE id = ?", tableName)
+		updateQuery := fmt.Sprintf("UPDATE %s SET last_seen = ? WHERE prefix = ?",
+			tableName)

-		_, err = tx.Exec(updateQuery, timestamp, p.ID.String())
+		_, err = tx.Exec(updateQuery, timestamp, prefix)
 		if err != nil {
 			return nil, err
 		}
@@ -1085,12 +1095,12 @@ func (d *Database) GetOrCreatePrefix(
 	}

 	// Prefix doesn't exist, create it
-	p = Prefix{ID: generateUUID(), Prefix: prefix, IPVersion: ipVersion,
+	p = Prefix{Prefix: prefix, IPVersion: ipVersion,
 		FirstSeen: timestamp, LastSeen: timestamp}
 	insertQuery := fmt.Sprintf("INSERT INTO %s "+
-		"(id, prefix, first_seen, last_seen) VALUES (?, ?, ?, ?)", tableName)
+		"(prefix, first_seen, last_seen) VALUES (?, ?, ?)", tableName)

-	_, err = tx.Exec(insertQuery, p.ID.String(), p.Prefix, p.FirstSeen, p.LastSeen)
+	_, err = tx.Exec(insertQuery, p.Prefix, p.FirstSeen, p.LastSeen)
 	if err != nil {
 		return nil, err
 	}
@@ -1389,17 +1399,24 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
 		return fmt.Errorf("failed to encode AS path: %w", err)
 	}

-	updateSQL, insertSQL, lookupSQL := updateLiveRouteV4SQL,
-		insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
+	originSQL, updateSQL := updateLiveRouteOriginV4SQL, updateLiveRouteV4SQL
+	insertSQL, lookupSQL := insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
+
 	if route.IPVersion == ipVersionV6 {
-		updateSQL, insertSQL, lookupSQL = updateLiveRouteV6SQL,
-			insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
+		originSQL, updateSQL = updateLiveRouteOriginV6SQL, updateLiveRouteV6SQL
+		insertSQL, lookupSQL = insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
 	}

 	ctx := context.Background()

 	// The write lock is held, so no other writer can insert this key between the
 	// update and the insert even though they are separate autocommit statements.
+	setOrigin, err := d.db.PrepareContext(ctx, originSQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare origin statement: %w", err)
+	}
+	defer func() { _ = setOrigin.Close() }()
+
 	upd, err := d.db.PrepareContext(ctx, updateSQL)
 	if err != nil {
 		return fmt.Errorf("failed to prepare update statement: %w", err)
@@ -1418,8 +1435,8 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
 	}
 	defer func() { _ = has.Close() }()

-	inserted, newPrefix, err := upsertRouteRow(ctx, upd, ins, has, route,
-		string(pathJSON))
+	inserted, newPrefix, err := upsertRouteRow(ctx, setOrigin, upd, ins, has,
+		route, string(pathJSON))
 	if err != nil {
 		return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
 	}