diff --git a/internal/database/counts_internal_test.go b/internal/database/counts_internal_test.go index c5decf7..bbe5577 100644 --- a/internal/database/counts_internal_test.go +++ b/internal/database/counts_internal_test.go @@ -29,7 +29,6 @@ func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute { } return &LiveRoute{ - ID: uuid.New(), Prefix: prefix, MaskLength: maskLength, IPVersion: ipVersionV4, @@ -44,7 +43,6 @@ func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute { // mkV6Route builds an IPv6 live route. func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute { return &LiveRoute{ - ID: uuid.New(), Prefix: prefix, MaskLength: 32, IPVersion: ipVersionV6, diff --git a/internal/database/database.go b/internal/database/database.go index a536e77..08ac7d1 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -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) } diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index d93b188..6650a7b 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "log/slog" + "math/rand/v2" "net/netip" "slices" "sync" @@ -14,7 +15,6 @@ import ( "git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/logger" - "github.com/google/uuid" "github.com/mattn/go-sqlite3" ) @@ -293,7 +293,7 @@ func TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer(t *testing.T) { t.Fatalf("UpsertLiveRoute(%s): %v", tt.route.Prefix, err) } - _, err = db.db.ExecContext(t.Context(), tt.insertSQL, uuid.NewString(), + _, err = db.db.ExecContext(t.Context(), tt.insertSQL, tt.route.Prefix, tt.route.MaskLength, otherOriginASN, tt.route.PeerIP, "[]", tt.route.NextHop, ts) @@ -317,7 +317,6 @@ func writeOriginRoutes(t *testing.T, db *Database, origins map[string]int) { for prefix, asn := range origins { routes = append(routes, &LiveRoute{ - ID: uuid.New(), Prefix: prefix, MaskLength: netip.MustParsePrefix(prefix).Bits(), IPVersion: detectIPVersion(prefix), @@ -1090,3 +1089,279 @@ func TestGetRandomPrefixesByLengthSkipsWithdrawnRoutes(t *testing.T) { routes, kept.Prefix) } } + +// TestLiveRouteUpdateSetsOriginWhenItChanges announces a peer's route for a +// prefix, then again with the same origin AS and a new path, then with a new +// origin AS. Each time the route must carry the announced origin AS, path and +// time, and the AS details must list the prefix under that origin AS only. The +// update sets the origin AS with a statement of its own, which changes nothing +// while the origin AS stays the same. +func TestLiveRouteUpdateSetsOriginWhenItChanges(t *testing.T) { + t.Parallel() + + 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() }() + + const firstOrigin, secondOrigin = 64500, 64501 + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + for _, asn := range []int{firstOrigin, secondOrigin} { + _, err := db.GetOrCreateASN(asn, ts) + if err != nil { + t.Fatalf("GetOrCreateASN(%d): %v", asn, err) + } + } + + route := mkV4Route(t, "198.51.100.0/24", firstOrigin, ts) + + for _, announced := range []struct { + origin int + path []int + }{ + {firstOrigin, []int{firstOrigin}}, + {firstOrigin, []int{secondOrigin, firstOrigin}}, + {secondOrigin, []int{firstOrigin, secondOrigin}}, + } { + ts = ts.Add(time.Minute) + route.OriginASN, route.ASPath, route.LastUpdated = + announced.origin, announced.path, ts + + err := db.UpsertLiveRouteBatch([]*LiveRoute{route}) + if err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + routes, _, err := db.GetPrefixDetails(route.Prefix) + if err != nil { + t.Fatalf("GetPrefixDetails: %v", err) + } + + if len(routes) != 1 || routes[0].OriginASN != announced.origin || + !slices.Equal(routes[0].ASPath, announced.path) || + !routes[0].LastUpdated.Equal(ts) { + t.Errorf("after announcing origin AS%d, path %v at %s: routes %+v", + announced.origin, announced.path, ts, routes) + } + + for _, asn := range []int{firstOrigin, secondOrigin} { + _, prefixes, err := db.GetASDetails(asn) + if err != nil { + t.Fatalf("GetASDetails(%d): %v", asn, err) + } + + listed := len(prefixes) == 1 && prefixes[0].Prefix == route.Prefix + if listed != (asn == announced.origin) { + t.Errorf("after announcing origin AS%d, AS%d lists %+v", + announced.origin, asn, prefixes) + } + } + } +} + +// TestUpdatePrefixesBatchKeepsFirstSeen writes a prefix, then a batch with it +// and a new prefix. Only the new prefix is counted, and the known one keeps its +// first_seen time while its last_seen time moves on. +func TestUpdatePrefixesBatchKeepsFirstSeen(t *testing.T) { + t.Parallel() + + 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() }() + + const known = "198.51.100.0/24" + + first := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + later := first.Add(time.Hour) + + err = db.UpdatePrefixesBatch(map[string]time.Time{known: first}) + if err != nil { + t.Fatalf("UpdatePrefixesBatch: %v", err) + } + + err = db.UpdatePrefixesBatch(map[string]time.Time{ + known: later, testV6Prefix: later, + }) + if err != nil { + t.Fatalf("UpdatePrefixesBatch: %v", err) + } + + stats, err := db.GetStats() + if err != nil { + t.Fatalf("GetStats: %v", err) + } + + if stats.IPv4Prefixes != 1 || stats.IPv6Prefixes != 1 { + t.Errorf("prefix counts = (v4 %d, v6 %d), want (1, 1)", + stats.IPv4Prefixes, stats.IPv6Prefixes) + } + + for _, tt := range []struct { + query, prefix string + firstSeen, lastSeen time.Time + }{ + { + query: "SELECT first_seen, last_seen FROM prefixes_v4 WHERE prefix = ?", + prefix: known, firstSeen: first, lastSeen: later, + }, + { + query: "SELECT first_seen, last_seen FROM prefixes_v6 WHERE prefix = ?", + prefix: testV6Prefix, firstSeen: later, lastSeen: later, + }, + } { + var firstSeen, lastSeen time.Time + + err := db.db.QueryRowContext(t.Context(), tt.query, tt.prefix). + Scan(&firstSeen, &lastSeen) + if err != nil { + t.Fatalf("reading %s: %v", tt.prefix, err) + } + + if !firstSeen.Equal(tt.firstSeen) || !lastSeen.Equal(tt.lastSeen) { + t.Errorf("%s: first seen %s, last seen %s; want %s, %s", tt.prefix, + firstSeen, lastSeen, tt.firstSeen, tt.lastSeen) + } + } +} + +// Parameters for BenchmarkPrefixBatchOnFiveMillionRoutes. +const ( + // benchPeers and benchPrefixes size the table the batches are written to: + // every peer has a route for every prefix, 5 million routes in all. + benchPeers = 10 + benchPrefixes = 500000 + // benchBatchSize is how many updates each batch holds, as many as a full + // prefix handler batch (prefixBatchSize in internal/routewatch). + benchBatchSize = 25000 + // benchWithdrawalEvery makes every fifth update of a timed batch a + // withdrawal; the others are announcements. + benchWithdrawalEvery = 5 + // benchOrigins is how many origin ASes the prefixes are spread over. + benchOrigins = 50000 + // benchFirstASN is the lowest AS number in the benchmark's AS paths. + benchFirstASN = 64500 +) + +// BenchmarkPrefixBatchOnFiveMillionRoutes times the writes the prefix handler +// makes for one full batch, 25,000 announcements and withdrawals of routes +// picked at random, to a database that already holds 5 million live routes: +// 500,000 IPv4 prefixes, each from 10 peers. It builds that database first, +// one batch at a time, which takes some minutes and a few GB of disk under +// TMPDIR. It is run by hand, not by make test, for example in the test image: +// +// go test -run '^$' -bench FiveMillion -benchtime 20x ./internal/database/ +func BenchmarkPrefixBatchOnFiveMillionRoutes(b *testing.B) { + db, err := New(&config.Config{StateDir: b.TempDir()}, logger.New()) + if err != nil { + b.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + rng := rand.New(rand.NewPCG(1, 2)) //nolint:gosec // repeatable, not secret + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + // Each peer sends its whole table, as when its session comes up. + for peer := range benchPeers { + for first := 0; first < benchPrefixes; first += benchBatchSize { + routes := make([]*LiveRoute, 0, benchBatchSize) + for prefix := first; prefix < first+benchBatchSize; prefix++ { + routes = append(routes, benchRoute(rng, prefix, peer, ts)) + } + + writeBenchBatch(b, db, routes, nil, ts) + } + } + + for b.Loop() { + ts = ts.Add(time.Second) + + var ( + routes []*LiveRoute + withdrawals []LiveRouteDeletion + ) + + for i := range benchBatchSize { + route := benchRoute(rng, rng.IntN(benchPrefixes), + rng.IntN(benchPeers), ts) + if i%benchWithdrawalEvery == 0 { + withdrawals = append(withdrawals, LiveRouteDeletion{ + Prefix: route.Prefix, PeerIP: route.PeerIP, + IPVersion: ipVersionV4, + }) + } else { + routes = append(routes, route) + } + } + + writeBenchBatch(b, db, routes, withdrawals, ts) + } +} + +// benchRoute returns the route peer has for the prefix numbered prefix, a /24, +// with a new AS path of three to six ASes that ends in the prefix's origin. +func benchRoute(rng *rand.Rand, prefix, peer int, ts time.Time) *LiveRoute { + const minPathLength, extraPathLength = 3, 4 + + network := netip.AddrFrom4([4]byte{ + byte(1 + prefix>>16), byte(prefix >> 8), byte(prefix), 0, + }) + origin := benchFirstASN + prefix%benchOrigins + + path := []int{benchFirstASN + peer} + for range minPathLength - 2 + rng.IntN(extraPathLength) { + path = append(path, benchFirstASN+rng.IntN(benchOrigins)) + } + + path = append(path, origin) + peerIP := netip.AddrFrom4([4]byte{198, 51, 100, byte(1 + peer)}).String() + + return &LiveRoute{ + Prefix: netip.PrefixFrom(network, ipv4Bits-8).String(), + MaskLength: ipv4Bits - 8, + IPVersion: ipVersionV4, + OriginASN: origin, + PeerIP: peerIP, + ASPath: path, + NextHop: peerIP, + LastUpdated: ts, + } +} + +// writeBenchBatch writes what the prefix handler writes for a batch: the +// announced routes, the withdrawals, and the last-seen time of every prefix +// the batch names. +func writeBenchBatch( + b *testing.B, db *Database, routes []*LiveRoute, + withdrawals []LiveRouteDeletion, ts time.Time, +) { + b.Helper() + + prefixes := make(map[string]time.Time, len(routes)+len(withdrawals)) + for _, route := range routes { + prefixes[route.Prefix] = ts + } + + for _, withdrawal := range withdrawals { + prefixes[withdrawal.Prefix] = ts + } + + err := db.UpsertLiveRouteBatch(routes) + if err != nil { + b.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + err = db.DeleteLiveRouteBatch(withdrawals) + if err != nil { + b.Fatalf("DeleteLiveRouteBatch: %v", err) + } + + err = db.UpdatePrefixesBatch(prefixes) + if err != nil { + b.Fatalf("UpdatePrefixesBatch: %v", err) + } +} diff --git a/internal/database/models.go b/internal/database/models.go index 3872f95..0a1d80a 100644 --- a/internal/database/models.go +++ b/internal/database/models.go @@ -37,7 +37,6 @@ type ASN struct { // Prefix represents an IP prefix (CIDR block) with its IP version (4 or 6) // and first/last seen timestamps. type Prefix struct { - ID uuid.UUID Prefix string IPVersion int // 4 or 6 FirstSeen time.Time @@ -71,7 +70,6 @@ type ASNPeering struct { // //nolint:tagliatelle // field names are the existing JSON API type LiveRoute struct { - ID uuid.UUID `json:"id"` Prefix string `json:"prefix"` MaskLength int `json:"mask_length"` IPVersion int `json:"ip_version"` diff --git a/internal/database/schema.sql b/internal/database/schema.sql index ef2f5dd..c5d85a1 100644 --- a/internal/database/schema.sql +++ b/internal/database/schema.sql @@ -27,21 +27,20 @@ CREATE TABLE IF NOT EXISTS asns ( whois_updated_at DATETIME -- when we last fetched WHOIS data ); --- IPv4 prefixes table +-- IPv4 prefixes table. Every read and write finds a row by its prefix, so the +-- rows are stored by prefix (WITHOUT ROWID) and need no index of their own. CREATE TABLE IF NOT EXISTS prefixes_v4 ( - id TEXT PRIMARY KEY, - prefix TEXT UNIQUE NOT NULL, + prefix TEXT PRIMARY KEY, first_seen DATETIME NOT NULL, last_seen DATETIME NOT NULL -); +) WITHOUT ROWID; --- IPv6 prefixes table +-- IPv6 prefixes table, stored as prefixes_v4 is. CREATE TABLE IF NOT EXISTS prefixes_v6 ( - id TEXT PRIMARY KEY, - prefix TEXT UNIQUE NOT NULL, + prefix TEXT PRIMARY KEY, first_seen DATETIME NOT NULL, last_seen DATETIME NOT NULL -); +) WITHOUT ROWID; CREATE TABLE IF NOT EXISTS announcements ( id TEXT PRIMARY KEY, @@ -75,11 +74,6 @@ CREATE TABLE IF NOT EXISTS bgp_peers ( last_message_type TEXT ); --- Indexes for prefixes_v4 table -CREATE INDEX IF NOT EXISTS idx_prefixes_v4_prefix ON prefixes_v4(prefix); - --- Indexes for prefixes_v6 table -CREATE INDEX IF NOT EXISTS idx_prefixes_v6_prefix ON prefixes_v6(prefix); CREATE INDEX IF NOT EXISTS idx_announcements_timestamp ON announcements(timestamp); CREATE INDEX IF NOT EXISTS idx_announcements_prefix_id ON announcements(prefix_id); CREATE INDEX IF NOT EXISTS idx_announcements_peer_asn ON announcements(peer_asn); @@ -97,9 +91,14 @@ CREATE INDEX IF NOT EXISTS idx_bgp_peers_asn ON bgp_peers(peer_asn); CREATE INDEX IF NOT EXISTS idx_bgp_peers_last_seen ON bgp_peers(last_seen); CREATE INDEX IF NOT EXISTS idx_bgp_peers_ip ON bgp_peers(peer_ip); --- IPv4 routing table maintained by PrefixHandler +-- IPv4 routing table maintained by PrefixHandler. A route write changes the +-- table and its entry in every index on a column the write sets, so the table +-- has only the indexes that the queries named with them use. CREATE TABLE IF NOT EXISTS live_routes_v4 ( - id TEXT PRIMARY KEY, + -- The row id, which the prefix-length pages' sample draws through the + -- mask_length index. Declared as a column, it stays the same for as long + -- as the route is live, also through a VACUUM. + id INTEGER PRIMARY KEY, prefix TEXT NOT NULL, mask_length INTEGER NOT NULL, -- CIDR mask length (0-32) origin_asn INTEGER NOT NULL, @@ -107,15 +106,15 @@ CREATE TABLE IF NOT EXISTS live_routes_v4 ( as_path TEXT NOT NULL, -- JSON array next_hop TEXT NOT NULL, last_updated DATETIME NOT NULL, - -- A peer has at most one route for a prefix. peer_ip comes first so that - -- this key also finds all of a peer's routes, when its BGP session goes - -- down; lookups by prefix use the prefix index below. + -- A peer has at most one route for a prefix. This key finds the route a + -- write updates or deletes and, peer_ip coming first, all of a peer's + -- routes when its BGP session goes down. UNIQUE(peer_ip, prefix) ); --- IPv6 routing table maintained by PrefixHandler +-- IPv6 routing table maintained by PrefixHandler, as live_routes_v4. CREATE TABLE IF NOT EXISTS live_routes_v6 ( - id TEXT PRIMARY KEY, + id INTEGER PRIMARY KEY, prefix TEXT NOT NULL, mask_length INTEGER NOT NULL, -- CIDR mask length (0-128) origin_asn INTEGER NOT NULL, @@ -123,24 +122,29 @@ CREATE TABLE IF NOT EXISTS live_routes_v6 ( as_path TEXT NOT NULL, -- JSON array next_hop TEXT NOT NULL, last_updated DATETIME NOT NULL, - -- As in live_routes_v4: peer_ip first, to find all of a peer's routes. UNIQUE(peer_ip, prefix) ); --- Indexes for live_routes_v4 table --- Finds a prefix's newest route in one search, whatever the number of peers. +-- Indexes for both live route tables, and the queries that use each. +-- +-- prefix, last_updated: a prefix's routes, for the prefix page and its API +-- answer, the IP lookup and its peer count, and the check whether a prefix +-- still has a live route; and a prefix's newest route for the prefix-length +-- pages, in one search whatever the number of peers. CREATE INDEX IF NOT EXISTS idx_live_routes_v4_prefix_last_updated ON live_routes_v4(prefix, last_updated); -CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_length ON live_routes_v4(mask_length); -CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_asn ON live_routes_v4(origin_asn); -CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated); --- Index to optimize prefix distribution queries -CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_prefix ON live_routes_v4(mask_length, prefix); - --- Indexes for live_routes_v6 table --- Finds a prefix's newest route in one search, whatever the number of peers. CREATE INDEX IF NOT EXISTS idx_live_routes_v6_prefix_last_updated ON live_routes_v6(prefix, last_updated); +-- mask_length: the prefix-length pages' random sample, the row id range of a +-- mask length and the first route of it at or after a row id. +CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_length ON live_routes_v4(mask_length); CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_length ON live_routes_v6(mask_length); +-- mask_length, prefix: the prefix-length pages' list of a mask length's +-- prefixes in text order, and the prefix distribution read at startup. +CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_prefix ON live_routes_v4(mask_length, prefix); +CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_prefix ON live_routes_v6(mask_length, prefix); +-- origin_asn: the prefixes an AS originates, for the AS page and its API answer. +CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_asn ON live_routes_v4(origin_asn); CREATE INDEX IF NOT EXISTS idx_live_routes_v6_origin_asn ON live_routes_v6(origin_asn); -CREATE INDEX IF NOT EXISTS idx_live_routes_v6_last_updated ON live_routes_v6(last_updated); --- Index to optimize prefix distribution queries -CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_prefix ON live_routes_v6(mask_length, prefix); \ No newline at end of file +-- last_updated: the oldest and newest route times in /api/v1/stats and +-- /status.json. +CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated); +CREATE INDEX IF NOT EXISTS idx_live_routes_v6_last_updated ON live_routes_v6(last_updated); \ No newline at end of file diff --git a/internal/routewatch/app_integration_internal_test.go b/internal/routewatch/app_integration_internal_test.go index bdab0a4..07d4060 100644 --- a/internal/routewatch/app_integration_internal_test.go +++ b/internal/routewatch/app_integration_internal_test.go @@ -18,7 +18,6 @@ import ( "git.eeqj.de/sneak/routewatch/internal/ristypes" "git.eeqj.de/sneak/routewatch/internal/server" "git.eeqj.de/sneak/routewatch/internal/streamer" - "github.com/google/uuid" ) // mockStore is a mock implementation of database.Store for testing @@ -97,7 +96,6 @@ func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error { } m.Prefixes[prefix] = &database.Prefix{ - ID: uuid.New(), Prefix: prefix, IPVersion: ipVersion, FirstSeen: timestamp, @@ -133,7 +131,6 @@ func (m *mockStore) GetOrCreatePrefix( } p := &database.Prefix{ - ID: uuid.New(), Prefix: prefix, IPVersion: ipVersion, FirstSeen: timestamp, @@ -370,7 +367,6 @@ func (m *mockStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error { // Track prefix if _, exists := m.Prefixes[route.Prefix]; !exists { m.Prefixes[route.Prefix] = &database.Prefix{ - ID: uuid.New(), Prefix: route.Prefix, IPVersion: route.IPVersion, FirstSeen: route.LastUpdated, diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 4858e76..c3ae80e 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -12,7 +12,6 @@ import ( "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/metrics" "git.eeqj.de/sneak/routewatch/internal/ristypes" - "github.com/google/uuid" ) const ( @@ -461,7 +460,6 @@ func (h *PrefixHandler) createLiveRoute(update prefixUpdate) *database.LiveRoute // Create live route record liveRoute := &database.LiveRoute{ - ID: uuid.New(), Prefix: update.prefix, MaskLength: maskLength, IPVersion: ipVersion, diff --git a/internal/server/handlers_internal_test.go b/internal/server/handlers_internal_test.go index ff68f08..8ea16a3 100644 --- a/internal/server/handlers_internal_test.go +++ b/internal/server/handlers_internal_test.go @@ -19,7 +19,6 @@ import ( "git.eeqj.de/sneak/routewatch/internal/metrics" "git.eeqj.de/sneak/routewatch/internal/streamer" "git.eeqj.de/sneak/routewatch/pkg/asinfo" - "github.com/google/uuid" ) // The prefix and peer addresses the prefix page tests store routes for. @@ -114,12 +113,12 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) { err = db.UpsertLiveRouteBatch([]*database.LiveRoute{ { - ID: uuid.New(), Prefix: "198.51.100.0/24", MaskLength: 24, IPVersion: 4, + Prefix: "198.51.100.0/24", MaskLength: 24, IPVersion: 4, OriginASN: 64500, PeerIP: "192.0.2.1", ASPath: []int{64500}, NextHop: "192.0.2.1", LastUpdated: ts, }, { - ID: uuid.New(), Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6, + Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6, OriginASN: 64501, PeerIP: testPeerV6, ASPath: []int{64501}, NextHop: testPeerV6, LastUpdated: ts, }, @@ -204,12 +203,12 @@ func TestPrefixDetailHandlersParsePrefix(t *testing.T) { err = db.UpsertLiveRouteBatch([]*database.LiveRoute{ { - ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, + Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, OriginASN: 64500, PeerIP: testPeerV4, ASPath: []int{64500}, NextHop: testPeerV4, LastUpdated: ts, }, { - ID: uuid.New(), Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6, + Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6, OriginASN: 64501, PeerIP: testPeerV6, ASPath: []int{64501}, NextHop: testPeerV6, LastUpdated: ts, }, @@ -287,12 +286,12 @@ func TestPrefixDetailShowsOriginNames(t *testing.T) { err = db.UpsertLiveRouteBatch([]*database.LiveRoute{ { - ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, + Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, OriginASN: origins[0], PeerIP: testPeerV4, ASPath: []int{origins[0]}, NextHop: testPeerV4, LastUpdated: ts, }, { - ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, + Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, OriginASN: origins[1], PeerIP: "198.51.100.2", ASPath: []int{origins[1]}, NextHop: "198.51.100.2", LastUpdated: ts, }, @@ -347,7 +346,7 @@ func TestPagesSayOriginZeroIsNoKnownOrigin(t *testing.T) { t.Cleanup(func() { _ = db.Close() }) err = db.UpsertLiveRouteBatch([]*database.LiveRoute{{ - ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, + Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, OriginASN: 0, PeerIP: testPeerV4, ASPath: []int{64500, 64501, 64502}, NextHop: testPeerV4, LastUpdated: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC), }})