diff --git a/internal/database/counts.go b/internal/database/counts.go index 71ce82f..5c8d4a5 100644 --- a/internal/database/counts.go +++ b/internal/database/counts.go @@ -33,8 +33,10 @@ type liveCounts struct { // seed sets every count to the value read from the tables at startup. It runs // before any writer, so it needs no coordination with the adjust methods. -func (c *liveCounts) seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6 int, - distributionV4, distributionV6 []PrefixDistribution) { +func (c *liveCounts) seed( + asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6 int, + distributionV4, distributionV6 []PrefixDistribution, +) { c.mu.Lock() defer c.mu.Unlock() @@ -45,9 +47,11 @@ func (c *liveCounts) seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV c.peers = peers c.routesV4 = routesV4 c.routesV6 = routesV6 + for _, entry := range distributionV4 { addAtMaskLength(c.distributionV4[:], entry.MaskLength, entry.Count) } + for _, entry := range distributionV6 { addAtMaskLength(c.distributionV6[:], entry.MaskLength, entry.Count) } @@ -102,6 +106,7 @@ func (c *liveCounts) addToDistribution(maskLengthsV4, maskLengthsV6 []int, n int for _, maskLength := range maskLengthsV4 { addAtMaskLength(c.distributionV4[:], maskLength, n) } + for _, maskLength := range maskLengthsV6 { addAtMaskLength(c.distributionV6[:], maskLength, n) } @@ -138,9 +143,11 @@ func (c *liveCounts) fill(s *Stats) { // ascending order, the way the distribution query returns them. func distributionList(counts []int) []PrefixDistribution { var list []PrefixDistribution + for maskLength, count := range counts { if count > 0 { - list = append(list, PrefixDistribution{MaskLength: maskLength, Count: count}) + list = append(list, + PrefixDistribution{MaskLength: maskLength, Count: count}) } } @@ -152,7 +159,8 @@ func distributionList(counts []int) []PrefixDistribution { func (d *Database) countRows(ctx context.Context, table string) (int, error) { var n int // table is one of a fixed set of literals below, never external input. - if err := d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+table).Scan(&n); err != nil { + err := d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+table).Scan(&n) + if err != nil { return 0, fmt.Errorf("failed to count %s: %w", table, err) } @@ -167,30 +175,37 @@ func (d *Database) seedCounts(ctx context.Context) error { if err != nil { return err } + prefixesV4, err := d.countRows(ctx, "prefixes_v4") if err != nil { return err } + prefixesV6, err := d.countRows(ctx, "prefixes_v6") if err != nil { return err } + peerings, err := d.countRows(ctx, "peerings") if err != nil { return err } + peers, err := d.countRows(ctx, "bgp_peers") if err != nil { return err } - routesV4, err := d.countRows(ctx, "live_routes_v4") + + routesV4, err := d.countRows(ctx, tableLiveRoutesV4) if err != nil { return err } - routesV6, err := d.countRows(ctx, "live_routes_v6") + + routesV6, err := d.countRows(ctx, tableLiveRoutesV6) if err != nil { return err } + distributionV4, distributionV6, err := d.GetPrefixDistributionContext(ctx) if err != nil { return err diff --git a/internal/database/database.go b/internal/database/database.go index 5393f74..bbe7a74 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -1,4 +1,3 @@ -// Package database provides SQLite storage for BGP routing data including ASNs, prefixes, announcements and peerings. package database import ( @@ -35,6 +34,13 @@ const ( ipVersionV6 = 6 ipv4Bits = 32 ipv6Bits = 128 + + tableLiveRoutesV4 = "live_routes_v4" + tableLiveRoutesV6 = "live_routes_v6" + + // Names of the IP versions in error and log messages. + ipv4Family = "IPv4" + ipv6Family = "IPv6" ) // SQLite memory tuning. cache_size and busy_timeout go in the DSN so every @@ -61,6 +67,9 @@ var ( ErrNoRoute = errors.New("no route found") // ErrNoStaleASN is returned when no ASN needs WHOIS refresh ErrNoStaleASN = errors.New("no stale ASN found") + + errInvalidASN = errors.New("invalid ASN") + errSameASNInPeering = errors.New("cannot create peering with same ASN") ) // Database manages the SQLite database connection and operations. @@ -83,7 +92,9 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { // Ensure directory exists dir := filepath.Dir(dbPath) - if err := os.MkdirAll(dir, dirPermissions); err != nil { + + err := os.MkdirAll(dir, dirPermissions) + if err != nil { return nil, fmt.Errorf("failed to create database directory: %w", err) } @@ -104,12 +115,14 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { sqliteCacheSizeKiB, sqliteBusyTimeoutMs, ) + db, err := sql.Open("sqlite3", dsn) if err != nil { return nil, fmt.Errorf("failed to open database: %w", err) } - if err := db.Ping(); err != nil { + err = db.PingContext(context.Background()) + if err != nil { return nil, fmt.Errorf("failed to ping database: %w", err) } @@ -120,16 +133,20 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { db.SetMaxIdleConns(maxConns) db.SetConnMaxLifetime(0) - database := &Database{db: db, logger: logger, path: dbPath, counts: &liveCounts{}} + database := &Database{ + db: db, logger: logger, path: dbPath, counts: &liveCounts{}, + } - if err := database.Initialize(); err != nil { + err = database.Initialize() + if err != nil { return nil, fmt.Errorf("failed to initialize database: %w", err) } // Seed the in-memory statistics counters from the tables once, before the // streamer starts writing. From here on every write keeps them current, so // the stats endpoints never scan the tables to report counts. - if err := database.seedCounts(context.Background()); err != nil { + err = database.seedCounts(context.Background()) + if err != nil { return nil, fmt.Errorf("failed to seed statistics counters: %w", err) } @@ -151,14 +168,18 @@ func (d *Database) Initialize() error { } for _, pragma := range pragmas { - if err := d.exec(pragma); err != nil { + err := d.exec(pragma) + if err != nil { d.logger.Warn("Failed to set pragma", "pragma", pragma, "error", err) } } // Run WAL checkpoint on startup to consolidate any existing WAL var walPages, checkpointedPages, movedPages int - err := d.db.QueryRow("PRAGMA wal_checkpoint(TRUNCATE)").Scan(&walPages, &checkpointedPages, &movedPages) + + err := d.db.QueryRowContext(context.Background(), + "PRAGMA wal_checkpoint(TRUNCATE)", + ).Scan(&walPages, &checkpointedPages, &movedPages) if err != nil { d.logger.Warn("Failed to checkpoint WAL on startup", "error", err) } else { @@ -182,43 +203,6 @@ func (d *Database) Close() error { return d.db.Close() } -// lock acquires the database mutex and logs debug information -func (d *Database) lock(operation string) { - // Get caller information - _, file, line, _ := runtime.Caller(1) - caller := fmt.Sprintf("%s:%d", filepath.Base(file), line) - - d.logger.Debug("Acquiring database lock", "operation", operation, "caller", caller) - - d.mu.Lock() - d.lockedAt = time.Now() - d.lockedBy = fmt.Sprintf("%s (%s)", operation, caller) - - d.logger.Debug("Database lock acquired", "operation", operation, "caller", caller) -} - -// unlock releases the database mutex and logs debug information including hold duration -func (d *Database) unlock() { - holdDuration := time.Since(d.lockedAt) - lockedBy := d.lockedBy - - d.lockedAt = time.Time{} - d.lockedBy = "" - d.mu.Unlock() - - d.logger.Debug("Database lock released", "held_by", lockedBy, "duration_ms", holdDuration.Milliseconds()) -} - -// beginTx starts a new transaction with logging -func (d *Database) beginTx() (*loggingTx, error) { - tx, err := d.db.Begin() - if err != nil { - return nil, err - } - - return &loggingTx{Tx: tx, logger: d.logger}, nil -} - // A live-route upsert is an UPDATE followed, only when no row matched, by an // INSERT. 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(*). @@ -227,52 +211,80 @@ func (d *Database) beginTx() (*loggingTx, error) { // updated route keeps its original id, exactly as the previous ON CONFLICT // upsert did. const ( - updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?, - last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?` + updateLiveRouteV4SQL = `UPDATE live_routes_v4 + SET mask_length = ?, as_path = ?, next_hop = ?, last_updated = ? + WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?` - insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip, - as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` + insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 + (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, + last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` - updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?, - last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?` + updateLiveRouteV6SQL = `UPDATE live_routes_v6 + SET mask_length = ?, as_path = ?, next_hop = ?, last_updated = ? + WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?` - insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip, - as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` + insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 + (id, 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 // up whether any live route has that prefix, to keep the in-memory prefix // distribution exact. The lookup reads one entry of the prefix index. const ( - prefixHasLiveRouteV4SQL = `SELECT EXISTS (SELECT 1 FROM live_routes_v4 WHERE prefix = ?)` - prefixHasLiveRouteV6SQL = `SELECT EXISTS (SELECT 1 FROM live_routes_v6 WHERE prefix = ?)` + prefixHasLiveRouteV4SQL = `SELECT EXISTS + (SELECT 1 FROM live_routes_v4 WHERE prefix = ?)` + prefixHasLiveRouteV6SQL = `SELECT EXISTS + (SELECT 1 FROM live_routes_v6 WHERE prefix = ?)` ) -// upsertRouteRowV4 updates an IPv4 live route, 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 upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) ( - inserted, newPrefix bool, err error) { - res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated, - route.Prefix, route.OriginASN, route.PeerIP) +// A deletion that names an origin ASN removes at most one row. One without an +// origin ASN removes the prefix's routes from that peer whatever their origin, +// which can be several rows. +const ( + deleteLiveRouteV4SQL = `DELETE FROM live_routes_v4 + WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?` + deleteLiveRouteAnyOriginV4SQL = `DELETE FROM live_routes_v4 + WHERE prefix = ? AND peer_ip = ?` + + deleteLiveRouteV6SQL = `DELETE FROM live_routes_v6 + WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?` + deleteLiveRouteAnyOriginV6SQL = `DELETE FROM live_routes_v6 + 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. +func upsertRouteRow( + ctx context.Context, upd, ins, has *sql.Stmt, route *LiveRoute, + pathJSON string, +) (bool, bool, error) { + res, err := upd.ExecContext(ctx, route.MaskLength, pathJSON, route.NextHop, + route.LastUpdated, route.Prefix, route.OriginASN, route.PeerIP) if err != nil { return false, false, err } + affected, err := res.RowsAffected() if err != nil { return false, false, err } + if affected > 0 { return false, false, nil } var prefixHadRoute bool - if err := has.QueryRow(route.Prefix).Scan(&prefixHadRoute); err != nil { + + err = has.QueryRowContext(ctx, route.Prefix).Scan(&prefixHadRoute) + if err != nil { return false, false, err } - _, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN, - route.PeerIP, pathJSON, route.NextHop, route.LastUpdated) + _, err = ins.ExecContext(ctx, route.ID.String(), route.Prefix, + route.MaskLength, route.OriginASN, route.PeerIP, pathJSON, + route.NextHop, route.LastUpdated) if err != nil { return false, false, err } @@ -280,39 +292,75 @@ func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string return true, !prefixHadRoute, nil } -// upsertRouteRowV6 updates an IPv6 live route, 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 upsertRouteRowV6(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) ( - inserted, newPrefix bool, err error) { - res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated, - route.Prefix, route.OriginASN, route.PeerIP) - if err != nil { - return false, false, err - } - affected, err := res.RowsAffected() - if err != nil { - return false, false, err - } - if affected > 0 { - return false, false, nil +// upsertRoutesInTable upserts routes, all of one IP version, in the live route +// table of that version. It returns how many new rows it inserted and the mask +// lengths of the prefixes that got their first live route. +func upsertRoutesInTable( + ctx context.Context, tx *loggingTx, routes []*LiveRoute, ipVersion int, +) (int, []int, error) { + if len(routes) == 0 { + return 0, nil, nil } - var prefixHadRoute bool - if err := has.QueryRow(route.Prefix).Scan(&prefixHadRoute); err != nil { - return false, false, err + family, updateSQL, insertSQL, lookupSQL := ipv4Family, updateLiveRouteV4SQL, + insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL + if ipVersion == ipVersionV6 { + family, updateSQL, insertSQL, lookupSQL = ipv6Family, + updateLiveRouteV6SQL, insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL } - _, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN, - route.PeerIP, pathJSON, route.NextHop, route.LastUpdated) + upd, err := tx.PrepareContext(ctx, updateSQL) if err != nil { - return false, false, err + return 0, nil, fmt.Errorf("failed to prepare %s update statement: %w", + family, err) + } + defer func() { _ = upd.Close() }() + + ins, err := tx.PrepareContext(ctx, insertSQL) + if err != nil { + return 0, nil, fmt.Errorf("failed to prepare %s insert statement: %w", + family, err) + } + defer func() { _ = ins.Close() }() + + has, err := tx.PrepareContext(ctx, lookupSQL) + if err != nil { + return 0, nil, fmt.Errorf( + "failed to prepare %s prefix lookup statement: %w", family, err) + } + defer func() { _ = has.Close() }() + + var newRows int + // Mask lengths of the prefixes that get their first live route. + var newPrefixMaskLengths []int + + for _, route := range routes { + pathJSON, err := json.Marshal(route.ASPath) + if err != nil { + return 0, nil, fmt.Errorf("failed to encode AS path: %w", err) + } + + inserted, newPrefix, err := upsertRouteRow(ctx, upd, ins, has, route, + string(pathJSON)) + if err != nil { + return 0, nil, fmt.Errorf("failed to upsert route %s: %w", + route.Prefix, err) + } + + if inserted { + newRows++ + } + + if newPrefix { + newPrefixMaskLengths = append(newPrefixMaskLengths, route.MaskLength) + } } - return true, !prefixHadRoute, nil + return newRows, newPrefixMaskLengths, nil } -// UpsertLiveRouteBatch inserts or updates multiple live routes in a single transaction +// UpsertLiveRouteBatch inserts or updates multiple live routes in a single +// transaction, the IPv4 routes first and then the IPv6 routes. func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error { if len(routes) == 0 { return nil @@ -325,86 +373,34 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error { if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } - defer func() { - if err := tx.Rollback(); err != nil && err != sql.ErrTxDone { - d.logger.Error("Failed to rollback transaction", "error", err) - } - }() + defer d.rollback(tx) - // Prepare the update and insert statements for both tables. - updV4, err := tx.Prepare(updateLiveRouteV4SQL) - if err != nil { - return fmt.Errorf("failed to prepare IPv4 update statement: %w", err) - } - defer func() { _ = updV4.Close() }() + var routesV4, routesV6 []*LiveRoute - insV4, err := tx.Prepare(insertLiveRouteV4SQL) - if err != nil { - return fmt.Errorf("failed to prepare IPv4 insert statement: %w", err) - } - defer func() { _ = insV4.Close() }() - - updV6, err := tx.Prepare(updateLiveRouteV6SQL) - if err != nil { - return fmt.Errorf("failed to prepare IPv6 update statement: %w", err) - } - defer func() { _ = updV6.Close() }() - - insV6, err := tx.Prepare(insertLiveRouteV6SQL) - if err != nil { - return fmt.Errorf("failed to prepare IPv6 insert statement: %w", err) - } - defer func() { _ = insV6.Close() }() - - hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL) - if err != nil { - return fmt.Errorf("failed to prepare IPv4 prefix lookup statement: %w", err) - } - defer func() { _ = hasV4.Close() }() - - hasV6, err := tx.Prepare(prefixHasLiveRouteV6SQL) - if err != nil { - return fmt.Errorf("failed to prepare IPv6 prefix lookup statement: %w", err) - } - defer func() { _ = hasV6.Close() }() - - var newV4, newV6 int - // Mask lengths of the prefixes that get their first live route in this batch. - var newPrefixMaskLengthsV4, newPrefixMaskLengthsV6 []int for _, route := range routes { - pathJSON, err := json.Marshal(route.ASPath) - if err != nil { - return fmt.Errorf("failed to encode AS path: %w", err) - } - if route.IPVersion == ipVersionV4 { - inserted, newPrefix, err := upsertRouteRowV4(updV4, insV4, hasV4, route, string(pathJSON)) - if err != nil { - return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) - } - if inserted { - newV4++ - } - if newPrefix { - newPrefixMaskLengthsV4 = append(newPrefixMaskLengthsV4, route.MaskLength) - } - - continue - } - - inserted, newPrefix, err := upsertRouteRowV6(updV6, insV6, hasV6, route, string(pathJSON)) - if err != nil { - return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) - } - if inserted { - newV6++ - } - if newPrefix { - newPrefixMaskLengthsV6 = append(newPrefixMaskLengthsV6, route.MaskLength) + routesV4 = append(routesV4, route) + } else { + routesV6 = append(routesV6, route) } } - if err = tx.Commit(); err != nil { + ctx := context.Background() + + newV4, newPrefixMaskLengthsV4, err := upsertRoutesInTable(ctx, tx, + routesV4, ipVersionV4) + if err != nil { + return err + } + + newV6, newPrefixMaskLengthsV6, err := upsertRoutesInTable(ctx, tx, + routesV6, ipVersionV6) + if err != nil { + return err + } + + err = tx.Commit() + if err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } @@ -414,7 +410,127 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error { return nil } -// DeleteLiveRouteBatch deletes multiple live routes in a single transaction +// deleteRouteRows runs one deletion with the delete statements of its IP +// version's table: withOrigin when it names an origin ASN, otherwise +// withoutOrigin. It returns how many rows it removed and whether that left the +// prefix with no live route. +func deleteRouteRows( + ctx context.Context, withOrigin, withoutOrigin, has *sql.Stmt, + del LiveRouteDeletion, +) (int64, bool, error) { + var ( + res sql.Result + err error + ) + + if del.OriginASN == 0 { + res, err = withoutOrigin.ExecContext(ctx, del.Prefix, del.PeerIP) + } else { + res, err = withOrigin.ExecContext(ctx, del.Prefix, del.OriginASN, + del.PeerIP) + } + + if err != nil { + return 0, false, fmt.Errorf("failed to delete route %s: %w", + del.Prefix, err) + } + + // A deletion with no origin ASN can remove several rows, so use the + // exact rows-affected count to keep the in-memory route counters right. + affected, err := res.RowsAffected() + if err != nil { + return 0, false, fmt.Errorf("failed to count deleted route %s: %w", + del.Prefix, err) + } + + if affected == 0 { + return 0, false, nil + } + + // The prefix leaves the distribution when no live route has it any more. + var prefixHasRoute bool + + err = has.QueryRowContext(ctx, del.Prefix).Scan(&prefixHasRoute) + if err != nil { + return 0, false, fmt.Errorf("failed to look up prefix %s: %w", + del.Prefix, err) + } + + return affected, !prefixHasRoute, nil +} + +// deleteRoutesInTable runs deletions, all of one IP version, on the live route +// table of that version. It returns how many rows it removed and the mask +// lengths of the prefixes it left with no live route. +func deleteRoutesInTable( + ctx context.Context, tx *loggingTx, deletions []LiveRouteDeletion, + ipVersion int, +) (int64, []int, error) { + if len(deletions) == 0 { + return 0, nil, nil + } + + family, withOriginSQL, withoutOriginSQL, lookupSQL := ipv4Family, + deleteLiveRouteV4SQL, deleteLiveRouteAnyOriginV4SQL, + prefixHasLiveRouteV4SQL + if ipVersion == ipVersionV6 { + family, withOriginSQL, withoutOriginSQL, lookupSQL = ipv6Family, + deleteLiveRouteV6SQL, deleteLiveRouteAnyOriginV6SQL, + prefixHasLiveRouteV6SQL + } + + withOrigin, err := tx.PrepareContext(ctx, withOriginSQL) + if err != nil { + return 0, nil, fmt.Errorf( + "failed to prepare %s delete with origin statement: %w", family, err) + } + defer func() { _ = withOrigin.Close() }() + + withoutOrigin, err := tx.PrepareContext(ctx, withoutOriginSQL) + if err != nil { + return 0, nil, fmt.Errorf( + "failed to prepare %s delete without origin statement: %w", + family, err) + } + defer func() { _ = withoutOrigin.Close() }() + + has, err := tx.PrepareContext(ctx, lookupSQL) + if err != nil { + return 0, nil, fmt.Errorf( + "failed to prepare %s prefix lookup statement: %w", family, err) + } + defer func() { _ = has.Close() }() + + var deleted int64 + // Mask lengths of the prefixes left with no live route. + var gonePrefixMaskLengths []int + + for _, del := range deletions { + affected, prefixGone, err := deleteRouteRows(ctx, + withOrigin, withoutOrigin, has, del) + if err != nil { + return 0, nil, err + } + + deleted += affected + + if !prefixGone { + continue + } + + maskLength, err := prefixMaskLength(del.Prefix) + if err != nil { + return 0, nil, err + } + + gonePrefixMaskLengths = append(gonePrefixMaskLengths, maskLength) + } + + return deleted, gonePrefixMaskLengths, nil +} + +// DeleteLiveRouteBatch deletes multiple live routes in a single transaction, +// the IPv4 routes first and then the IPv6 routes. func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { if len(deletions) == 0 { return nil @@ -427,125 +543,34 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } - defer func() { - if err := tx.Rollback(); err != nil && err != sql.ErrTxDone { - d.logger.Error("Failed to rollback transaction", "error", err) - } - }() + defer d.rollback(tx) - // No longer need to separate deletions since we handle them in the loop below + var deletionsV4, deletionsV6 []LiveRouteDeletion - // Prepare statements for both IPv4 and IPv6 tables - stmtV4WithOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`) - if err != nil { - return fmt.Errorf("failed to prepare IPv4 delete with origin statement: %w", err) - } - defer func() { _ = stmtV4WithOrigin.Close() }() - - stmtV4WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`) - if err != nil { - return fmt.Errorf("failed to prepare IPv4 delete without origin statement: %w", err) - } - defer func() { _ = stmtV4WithoutOrigin.Close() }() - - stmtV6WithOrigin, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`) - if err != nil { - return fmt.Errorf("failed to prepare IPv6 delete with origin statement: %w", err) - } - defer func() { _ = stmtV6WithOrigin.Close() }() - - stmtV6WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`) - if err != nil { - return fmt.Errorf("failed to prepare IPv6 delete without origin statement: %w", err) - } - defer func() { _ = stmtV6WithoutOrigin.Close() }() - - hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL) - if err != nil { - return fmt.Errorf("failed to prepare IPv4 prefix lookup statement: %w", err) - } - defer func() { _ = hasV4.Close() }() - - hasV6, err := tx.Prepare(prefixHasLiveRouteV6SQL) - if err != nil { - return fmt.Errorf("failed to prepare IPv6 prefix lookup statement: %w", err) - } - defer func() { _ = hasV6.Close() }() - - // Process deletions - var deletedV4, deletedV6 int64 - // Mask lengths of the prefixes this batch leaves with no live route. - var gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6 []int for _, del := range deletions { - var stmt *sql.Stmt - - // Select appropriate statement based on IP version and whether we have origin ASN - //nolint:nestif // Clear logic for selecting the right statement if del.IPVersion == ipVersionV4 { - if del.OriginASN == 0 { - stmt = stmtV4WithoutOrigin - } else { - stmt = stmtV4WithOrigin - } + deletionsV4 = append(deletionsV4, del) } else { - if del.OriginASN == 0 { - stmt = stmtV6WithoutOrigin - } else { - stmt = stmtV6WithOrigin - } - } - - // Execute deletion - var res sql.Result - if del.OriginASN == 0 { - res, err = stmt.Exec(del.Prefix, del.PeerIP) - } else { - res, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP) - } - - if err != nil { - return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err) - } - - // A deletion with no origin ASN can remove several rows, so use the - // exact rows-affected count to keep the in-memory route counters right. - affected, err := res.RowsAffected() - if err != nil { - return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err) - } - if del.IPVersion == ipVersionV4 { - deletedV4 += affected - } else { - deletedV6 += affected - } - if affected == 0 { - continue - } - - // The prefix leaves the distribution when no live route has it any more. - has := hasV4 - if del.IPVersion != ipVersionV4 { - has = hasV6 - } - var prefixHasRoute bool - if err := has.QueryRow(del.Prefix).Scan(&prefixHasRoute); err != nil { - return fmt.Errorf("failed to look up prefix %s: %w", del.Prefix, err) - } - if prefixHasRoute { - continue - } - maskLength, err := prefixMaskLength(del.Prefix) - if err != nil { - return err - } - if del.IPVersion == ipVersionV4 { - gonePrefixMaskLengthsV4 = append(gonePrefixMaskLengthsV4, maskLength) - } else { - gonePrefixMaskLengthsV6 = append(gonePrefixMaskLengthsV6, maskLength) + deletionsV6 = append(deletionsV6, del) } } - if err = tx.Commit(); err != nil { + ctx := context.Background() + + deletedV4, gonePrefixMaskLengthsV4, err := deleteRoutesInTable(ctx, tx, + deletionsV4, ipVersionV4) + if err != nil { + return err + } + + deletedV6, gonePrefixMaskLengthsV6, err := deleteRoutesInTable(ctx, tx, + deletionsV6, ipVersionV6) + if err != nil { + return err + } + + err = tx.Commit() + if err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } @@ -555,7 +580,107 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { return nil } -// UpdatePrefixesBatch updates the last_seen time for multiple prefixes in a single transaction +// 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. +func upsertPrefix( + ctx context.Context, selectStmt, 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) + } +} + +// upsertPrefixesInTable updates the last_seen time of prefixes, all of one IP +// version, in the prefix table of that version, inserting the ones that are +// new. It returns how many it inserted. +func upsertPrefixesInTable( + ctx context.Context, tx *loggingTx, prefixes map[string]time.Time, + ipVersion int, +) (int, error) { + if len(prefixes) == 0 { + return 0, nil + } + + 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 (?, ?, ?, ?)" + + 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 (?, ?, ?, ?)" + } + + 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", + family, err) + } + defer func() { _ = updateStmt.Close() }() + + insertStmt, err := tx.PrepareContext(ctx, insertSQL) + if err != nil { + return 0, fmt.Errorf("failed to prepare %s insert statement: %w", + family, err) + } + defer func() { _ = insertStmt.Close() }() + + var newPrefixes int + + for prefix, timestamp := range prefixes { + inserted, err := upsertPrefix(ctx, selectStmt, updateStmt, insertStmt, + prefix, timestamp) + if err != nil { + return 0, err + } + + if inserted { + newPrefixes++ + } + } + + return newPrefixes, nil +} + +// UpdatePrefixesBatch updates the last_seen time for multiple prefixes in a +// single transaction, the IPv4 prefixes first and then the IPv6 prefixes. func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { if len(prefixes) == 0 { return nil @@ -568,87 +693,33 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } - defer func() { - if err := tx.Rollback(); err != nil && err != sql.ErrTxDone { - d.logger.Error("Failed to rollback transaction", "error", err) - } - }() + defer d.rollback(tx) - // Prepare statements for both IPv4 and IPv6 tables - selectV4Stmt, err := tx.Prepare("SELECT id FROM prefixes_v4 WHERE prefix = ?") - if err != nil { - return fmt.Errorf("failed to prepare IPv4 select statement: %w", err) - } - defer func() { _ = selectV4Stmt.Close() }() + prefixesV4 := make(map[string]time.Time) + prefixesV6 := make(map[string]time.Time) - updateV4Stmt, err := tx.Prepare("UPDATE prefixes_v4 SET last_seen = ? WHERE prefix = ?") - if err != nil { - return fmt.Errorf("failed to prepare IPv4 update statement: %w", err) - } - defer func() { _ = updateV4Stmt.Close() }() - - insertV4Stmt, err := tx.Prepare("INSERT INTO prefixes_v4 (id, prefix, first_seen, last_seen) VALUES (?, ?, ?, ?)") - if err != nil { - return fmt.Errorf("failed to prepare IPv4 insert statement: %w", err) - } - defer func() { _ = insertV4Stmt.Close() }() - - selectV6Stmt, err := tx.Prepare("SELECT id FROM prefixes_v6 WHERE prefix = ?") - if err != nil { - return fmt.Errorf("failed to prepare IPv6 select statement: %w", err) - } - defer func() { _ = selectV6Stmt.Close() }() - - updateV6Stmt, err := tx.Prepare("UPDATE prefixes_v6 SET last_seen = ? WHERE prefix = ?") - if err != nil { - return fmt.Errorf("failed to prepare IPv6 update statement: %w", err) - } - defer func() { _ = updateV6Stmt.Close() }() - - insertV6Stmt, err := tx.Prepare("INSERT INTO prefixes_v6 (id, prefix, first_seen, last_seen) VALUES (?, ?, ?, ?)") - if err != nil { - return fmt.Errorf("failed to prepare IPv6 insert statement: %w", err) - } - defer func() { _ = insertV6Stmt.Close() }() - - var newV4, newV6 int for prefix, timestamp := range prefixes { - ipVersion := detectIPVersion(prefix) - - var selectStmt, updateStmt, insertStmt *sql.Stmt - if ipVersion == ipVersionV4 { - selectStmt, updateStmt, insertStmt = selectV4Stmt, updateV4Stmt, insertV4Stmt + if detectIPVersion(prefix) == ipVersionV4 { + prefixesV4[prefix] = timestamp } else { - selectStmt, updateStmt, insertStmt = selectV6Stmt, updateV6Stmt, insertV6Stmt - } - - var id string - err = selectStmt.QueryRow(prefix).Scan(&id) - - switch err { - case nil: - // Prefix exists, update last_seen - _, err = updateStmt.Exec(timestamp, prefix) - if err != nil { - return fmt.Errorf("failed to update prefix %s: %w", prefix, err) - } - case sql.ErrNoRows: - // Prefix doesn't exist, create it - _, err = insertStmt.Exec(generateUUID().String(), prefix, timestamp, timestamp) - if err != nil { - return fmt.Errorf("failed to insert prefix %s: %w", prefix, err) - } - if ipVersion == ipVersionV4 { - newV4++ - } else { - newV6++ - } - default: - return fmt.Errorf("failed to query prefix %s: %w", prefix, err) + prefixesV6[prefix] = timestamp } } - if err = tx.Commit(); err != nil { + ctx := context.Background() + + newV4, err := upsertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4) + if err != nil { + return err + } + + newV6, err := upsertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6) + if err != nil { + return err + } + + err = tx.Commit() + if err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } @@ -657,145 +728,10 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { return nil } -// GetOrCreateASNBatch creates or updates multiple ASNs in a single transaction -func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error { - if len(asns) == 0 { - return nil - } - - d.lock("GetOrCreateASNBatch") - defer d.unlock() - - tx, err := d.beginTx() - if err != nil { - return fmt.Errorf("failed to begin transaction: %w", err) - } - defer func() { - if err := tx.Rollback(); err != nil && err != sql.ErrTxDone { - d.logger.Error("Failed to rollback transaction", "error", err) - } - }() - - // Prepare statements - selectStmt, err := tx.Prepare( - "SELECT asn, handle, description, first_seen, last_seen FROM asns WHERE asn = ?") - if err != nil { - return fmt.Errorf("failed to prepare select statement: %w", err) - } - defer func() { _ = selectStmt.Close() }() - - updateStmt, err := tx.Prepare("UPDATE asns SET last_seen = ? WHERE asn = ?") - if err != nil { - return fmt.Errorf("failed to prepare update statement: %w", err) - } - defer func() { _ = updateStmt.Close() }() - - insertStmt, err := tx.Prepare( - "INSERT INTO asns (asn, handle, description, first_seen, last_seen) VALUES (?, ?, ?, ?, ?)") - if err != nil { - return fmt.Errorf("failed to prepare insert statement: %w", err) - } - defer func() { _ = insertStmt.Close() }() - - var newASNs int - for number, timestamp := range asns { - var asn ASN - var handle, description sql.NullString - - err = selectStmt.QueryRow(number).Scan(&asn.ASN, &handle, &description, &asn.FirstSeen, &asn.LastSeen) - - if err == nil { - // ASN exists, update last_seen - _, err = updateStmt.Exec(timestamp, number) - if err != nil { - return fmt.Errorf("failed to update ASN %d: %w", number, err) - } - - continue - } - - if err == sql.ErrNoRows { - // ASN doesn't exist, create it - asn = ASN{ - ASN: number, - FirstSeen: timestamp, - LastSeen: timestamp, - } - - // Look up ASN info - if info, ok := asinfo.Get(number); ok { - asn.Handle = info.Handle - asn.Description = info.Description - } - - _, err = insertStmt.Exec(asn.ASN, asn.Handle, asn.Description, asn.FirstSeen, asn.LastSeen) - if err != nil { - return fmt.Errorf("failed to insert ASN %d: %w", number, err) - } - newASNs++ - - continue - } - - if err != nil { - return fmt.Errorf("failed to query ASN %d: %w", number, err) - } - } - - if err = tx.Commit(); err != nil { - return fmt.Errorf("failed to commit transaction: %w", err) - } - - d.counts.addASNs(newASNs) - - return nil -} - -// GetOrCreateASN retrieves an existing ASN or creates a new one if it doesn't exist. -func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error) { - d.lock("GetOrCreateASN") - defer d.unlock() - - tx, err := d.beginTx() - if err != nil { - return nil, err - } - defer func() { - if err := tx.Rollback(); err != nil && err != sql.ErrTxDone { - d.logger.Error("Failed to rollback transaction", "error", err) - } - }() - - var asn ASN - var handle, description sql.NullString - err = tx.QueryRow("SELECT asn, handle, description, first_seen, last_seen FROM asns WHERE asn = ?", number). - Scan(&asn.ASN, &handle, &description, &asn.FirstSeen, &asn.LastSeen) - - if err == nil { - // ASN exists, update last_seen - asn.Handle = handle.String - asn.Description = description.String - _, err = tx.Exec("UPDATE asns SET last_seen = ? WHERE asn = ?", timestamp, number) - if err != nil { - return nil, err - } - asn.LastSeen = timestamp - - if err = tx.Commit(); err != nil { - d.logger.Error("Failed to commit transaction for ASN update", "asn", number, "error", err) - - return nil, err - } - - return &asn, nil - } - - if err != sql.ErrNoRows { - return nil, err - } - - // ASN doesn't exist, create it with ASN info lookup - asn = ASN{ +// newASN returns a new ASN record first and last seen at timestamp, with the +// handle and description from the AS info lookup when it knows the number. +func newASN(number int, timestamp time.Time) ASN { + asn := ASN{ ASN: number, FirstSeen: timestamp, LastSeen: timestamp, @@ -807,14 +743,177 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error) asn.Description = info.Description } - _, err = tx.Exec("INSERT INTO asns (asn, handle, description, first_seen, last_seen) VALUES (?, ?, ?, ?, ?)", + return asn +} + +// upsertASN updates the last_seen time of ASN number with the given +// statements, inserting the ASN when it is new. It reports whether it inserted +// it. +func upsertASN( + ctx context.Context, selectStmt, updateStmt, insertStmt *sql.Stmt, + number int, timestamp time.Time, +) (bool, error) { + var ( + asn ASN + handle, description sql.NullString + ) + + err := selectStmt.QueryRowContext(ctx, number).Scan(&asn.ASN, &handle, + &description, &asn.FirstSeen, &asn.LastSeen) + if err == nil { + // ASN exists, update last_seen + _, err = updateStmt.ExecContext(ctx, timestamp, number) + if err != nil { + return false, fmt.Errorf("failed to update ASN %d: %w", number, err) + } + + return false, nil + } + + if !errors.Is(err, sql.ErrNoRows) { + return false, fmt.Errorf("failed to query ASN %d: %w", number, err) + } + + // ASN doesn't exist, create it + asn = newASN(number, timestamp) + + _, err = insertStmt.ExecContext(ctx, asn.ASN, asn.Handle, + asn.Description, asn.FirstSeen, asn.LastSeen) + if err != nil { + return false, fmt.Errorf("failed to insert ASN %d: %w", number, err) + } + + return true, nil +} + +// GetOrCreateASNBatch creates or updates multiple ASNs in a single transaction +func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error { + if len(asns) == 0 { + return nil + } + + d.lock("GetOrCreateASNBatch") + defer d.unlock() + + ctx := context.Background() + + tx, err := d.beginTx() + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) + } + defer d.rollback(tx) + + // Prepare statements + selectStmt, err := tx.PrepareContext(ctx, + "SELECT asn, handle, description, first_seen, last_seen "+ + "FROM asns WHERE asn = ?") + if err != nil { + return fmt.Errorf("failed to prepare select statement: %w", err) + } + defer func() { _ = selectStmt.Close() }() + + updateStmt, err := tx.PrepareContext(ctx, + "UPDATE asns SET last_seen = ? WHERE asn = ?") + if err != nil { + return fmt.Errorf("failed to prepare update statement: %w", err) + } + defer func() { _ = updateStmt.Close() }() + + insertStmt, err := tx.PrepareContext(ctx, + "INSERT INTO asns (asn, handle, description, first_seen, last_seen) "+ + "VALUES (?, ?, ?, ?, ?)") + if err != nil { + return fmt.Errorf("failed to prepare insert statement: %w", err) + } + defer func() { _ = insertStmt.Close() }() + + var newASNs int + + for number, timestamp := range asns { + inserted, err := upsertASN(ctx, selectStmt, updateStmt, insertStmt, + number, timestamp) + if err != nil { + return err + } + + if inserted { + newASNs++ + } + } + + err = tx.Commit() + if err != nil { + return fmt.Errorf("failed to commit transaction: %w", err) + } + + d.counts.addASNs(newASNs) + + return nil +} + +// GetOrCreateASN retrieves an existing ASN or creates a new one if it doesn't +// exist. +func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error) { + d.lock("GetOrCreateASN") + defer d.unlock() + + tx, err := d.beginTx() + if err != nil { + return nil, err + } + defer d.rollback(tx) + + var ( + asn ASN + handle, description sql.NullString + ) + + err = tx.QueryRow("SELECT asn, handle, description, first_seen, last_seen "+ + "FROM asns WHERE asn = ?", number). + Scan(&asn.ASN, &handle, &description, &asn.FirstSeen, &asn.LastSeen) + if err == nil { + // ASN exists, update last_seen + asn.Handle = handle.String + asn.Description = description.String + + _, err = tx.Exec("UPDATE asns SET last_seen = ? WHERE asn = ?", + timestamp, number) + if err != nil { + return nil, err + } + + asn.LastSeen = timestamp + + err = tx.Commit() + if err != nil { + d.logger.Error("Failed to commit transaction for ASN update", + "asn", number, "error", err) + + return nil, err + } + + return &asn, nil + } + + if !errors.Is(err, sql.ErrNoRows) { + return nil, err + } + + // ASN doesn't exist, create it with ASN info lookup + asn = newASN(number, timestamp) + + _, err = tx.Exec("INSERT INTO asns "+ + "(asn, handle, description, first_seen, last_seen) "+ + "VALUES (?, ?, ?, ?, ?)", asn.ASN, asn.Handle, asn.Description, asn.FirstSeen, asn.LastSeen) if err != nil { return nil, err } - if err = tx.Commit(); err != nil { - d.logger.Error("Failed to commit transaction for ASN creation", "asn", number, "error", err) + err = tx.Commit() + if err != nil { + d.logger.Error("Failed to commit transaction for ASN creation", + "asn", number, "error", err) return nil, err } @@ -824,8 +923,10 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error) return &asn, nil } -// GetOrCreatePrefix retrieves an existing prefix or creates a new one if it doesn't exist. -func (d *Database) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefix, error) { +// GetOrCreatePrefix retrieves an existing prefix or creates a new one if it +// doesn't exist. +func (d *Database) GetOrCreatePrefix( + prefix string, timestamp time.Time) (*Prefix, error) { d.lock("GetOrCreatePrefix") defer d.unlock() @@ -833,38 +934,42 @@ func (d *Database) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefi if err != nil { return nil, err } - defer func() { - if err := tx.Rollback(); err != nil && err != sql.ErrTxDone { - d.logger.Error("Failed to rollback transaction", "error", err) - } - }() + defer d.rollback(tx) // Determine table based on IP version ipVersion := detectIPVersion(prefix) + tableName := "prefixes_v4" if ipVersion == ipVersionV6 { tableName = "prefixes_v6" } - var p Prefix - var idStr string - query := fmt.Sprintf("SELECT id, prefix, first_seen, last_seen FROM %s WHERE prefix = ?", tableName) - err = tx.QueryRow(query, prefix). - Scan(&idStr, &p.Prefix, &p.FirstSeen, &p.LastSeen) + var ( + p Prefix + idStr string + ) + query := fmt.Sprintf("SELECT id, prefix, first_seen, last_seen "+ + "FROM %s WHERE prefix = ?", tableName) + + err = tx.QueryRow(query, prefix).Scan(&idStr, &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) + _, err = tx.Exec(updateQuery, timestamp, p.ID.String()) if err != nil { return nil, err } + p.LastSeen = timestamp - if err = tx.Commit(); err != nil { - d.logger.Error("Failed to commit transaction for prefix update", "prefix", prefix, "error", err) + err = tx.Commit() + if err != nil { + d.logger.Error("Failed to commit transaction for prefix update", + "prefix", prefix, "error", err) return nil, err } @@ -872,26 +977,25 @@ func (d *Database) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefi return &p, nil } - if err != sql.ErrNoRows { + if !errors.Is(err, sql.ErrNoRows) { return nil, err } // Prefix doesn't exist, create it - p = Prefix{ - ID: generateUUID(), - Prefix: prefix, - IPVersion: ipVersion, - FirstSeen: timestamp, - LastSeen: timestamp, - } - insertQuery := fmt.Sprintf("INSERT INTO %s (id, prefix, first_seen, last_seen) VALUES (?, ?, ?, ?)", tableName) + p = Prefix{ID: generateUUID(), Prefix: prefix, IPVersion: ipVersion, + FirstSeen: timestamp, LastSeen: timestamp} + insertQuery := fmt.Sprintf("INSERT INTO %s "+ + "(id, prefix, first_seen, last_seen) VALUES (?, ?, ?, ?)", tableName) + _, err = tx.Exec(insertQuery, p.ID.String(), p.Prefix, p.FirstSeen, p.LastSeen) if err != nil { return nil, err } - if err = tx.Commit(); err != nil { - d.logger.Error("Failed to commit transaction for prefix creation", "prefix", prefix, "error", err) + err = tx.Commit() + if err != nil { + d.logger.Error("Failed to commit transaction for prefix creation", + "prefix", prefix, "error", err) return nil, err } @@ -905,17 +1009,20 @@ func (d *Database) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefi return &p, nil } -// RecordAnnouncement inserts a new BGP announcement or withdrawal into the database. +// RecordAnnouncement inserts a new BGP announcement or withdrawal into the +// database. func (d *Database) RecordAnnouncement(announcement *Announcement) error { d.lock("RecordAnnouncement") defer d.unlock() err := d.exec(` - INSERT INTO announcements (id, prefix_id, peer_asn, origin_asn, path, next_hop, timestamp, is_withdrawal) + INSERT INTO announcements (id, prefix_id, peer_asn, origin_asn, path, + next_hop, timestamp, is_withdrawal) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, announcement.ID.String(), announcement.PrefixID.String(), announcement.PeerASN, announcement.OriginASN, - announcement.Path, announcement.NextHop, announcement.Timestamp, announcement.IsWithdrawal) + announcement.Path, announcement.NextHop, announcement.Timestamp, + announcement.IsWithdrawal) return err } @@ -924,10 +1031,11 @@ func (d *Database) RecordAnnouncement(announcement *Announcement) error { func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error { // Validate ASNs if asA <= 0 || asB <= 0 { - return fmt.Errorf("invalid ASN: asA=%d, asB=%d", asA, asB) + return fmt.Errorf("%w: asA=%d, asB=%d", errInvalidASN, asA, asB) } + if asA == asB { - return fmt.Errorf("cannot create peering with same ASN: %d", asA) + return fmt.Errorf("%w: %d", errSameASNInPeering, asA) } // Normalize: ensure asA < asB @@ -942,21 +1050,20 @@ func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error { if err != nil { return err } - defer func() { - if err := tx.Rollback(); err != nil && err != sql.ErrTxDone { - d.logger.Error("Failed to rollback transaction", "error", err) - } - }() + defer d.rollback(tx) var exists bool - err = tx.QueryRow("SELECT EXISTS(SELECT 1 FROM peerings WHERE as_a = ? AND as_b = ?)", + + err = tx.QueryRow( + "SELECT EXISTS(SELECT 1 FROM peerings WHERE as_a = ? AND as_b = ?)", asA, asB).Scan(&exists) if err != nil { return err } if exists { - _, err = tx.Exec("UPDATE peerings SET last_seen = ? WHERE as_a = ? AND as_b = ?", + _, err = tx.Exec( + "UPDATE peerings SET last_seen = ? WHERE as_a = ? AND as_b = ?", timestamp, asA, asB) } else { _, err = tx.Exec(` @@ -964,11 +1071,13 @@ func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error { VALUES (?, ?, ?, ?, ?)`, generateUUID().String(), asA, asB, timestamp, timestamp) } + if err != nil { return err } - if err = tx.Commit(); err != nil { + err = tx.Commit() + if err != nil { d.logger.Error("Failed to commit transaction for peering", "as_a", asA, "as_b", asB, @@ -994,61 +1103,68 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error { d.lock("UpdatePeerBatch") defer d.unlock() + ctx := context.Background() + tx, err := d.beginTx() if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } - defer func() { - if err := tx.Rollback(); err != nil && err != sql.ErrTxDone { - d.logger.Error("Failed to rollback transaction", "error", err) - } - }() + defer d.rollback(tx) // Prepare statements - checkStmt, err := tx.Prepare("SELECT EXISTS(SELECT 1 FROM bgp_peers WHERE peer_ip = ?)") + checkStmt, err := tx.PrepareContext(ctx, + "SELECT EXISTS(SELECT 1 FROM bgp_peers WHERE peer_ip = ?)") if err != nil { return fmt.Errorf("failed to prepare check statement: %w", err) } defer func() { _ = checkStmt.Close() }() - updateStmt, err := tx.Prepare( - "UPDATE bgp_peers SET peer_asn = ?, last_seen = ?, last_message_type = ? WHERE peer_ip = ?") + updateStmt, err := tx.PrepareContext(ctx, + "UPDATE bgp_peers SET peer_asn = ?, last_seen = ?, "+ + "last_message_type = ? WHERE peer_ip = ?") if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } defer func() { _ = updateStmt.Close() }() - insertStmt, err := tx.Prepare( - "INSERT INTO bgp_peers (id, peer_ip, peer_asn, first_seen, last_seen, last_message_type) VALUES (?, ?, ?, ?, ?, ?)") + insertStmt, err := tx.PrepareContext(ctx, + "INSERT INTO bgp_peers (id, peer_ip, peer_asn, first_seen, "+ + "last_seen, last_message_type) VALUES (?, ?, ?, ?, ?, ?)") if err != nil { return fmt.Errorf("failed to prepare insert statement: %w", err) } defer func() { _ = insertStmt.Close() }() var newPeers int + for _, update := range peers { var exists bool - err = checkStmt.QueryRow(update.PeerIP).Scan(&exists) + + err = checkStmt.QueryRowContext(ctx, update.PeerIP).Scan(&exists) if err != nil { return fmt.Errorf("failed to check peer %s: %w", update.PeerIP, err) } if exists { - _, err = updateStmt.Exec(update.PeerASN, update.Timestamp, update.MessageType, update.PeerIP) + _, err = updateStmt.ExecContext(ctx, update.PeerASN, + update.Timestamp, update.MessageType, update.PeerIP) } else { - _, err = insertStmt.Exec( + _, err = insertStmt.ExecContext(ctx, generateUUID().String(), update.PeerIP, update.PeerASN, update.Timestamp, update.Timestamp, update.MessageType) } + if err != nil { return fmt.Errorf("failed to update peer %s: %w", update.PeerIP, err) } + if !exists { newPeers++ } } - if err = tx.Commit(); err != nil { + err = tx.Commit() + if err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } @@ -1058,7 +1174,9 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error { } // UpdatePeer updates or creates a BGP peer record -func (d *Database) UpdatePeer(peerIP string, peerASN int, messageType string, timestamp time.Time) error { +func (d *Database) UpdatePeer( + peerIP string, peerASN int, messageType string, timestamp time.Time, +) error { d.lock("UpdatePeer") defer d.unlock() @@ -1066,34 +1184,38 @@ func (d *Database) UpdatePeer(peerIP string, peerASN int, messageType string, ti if err != nil { return err } - defer func() { - if err := tx.Rollback(); err != nil && err != sql.ErrTxDone { - d.logger.Error("Failed to rollback transaction", "error", err) - } - }() + defer d.rollback(tx) var exists bool - err = tx.QueryRow("SELECT EXISTS(SELECT 1 FROM bgp_peers WHERE peer_ip = ?)", peerIP).Scan(&exists) + + err = tx.QueryRow( + "SELECT EXISTS(SELECT 1 FROM bgp_peers WHERE peer_ip = ?)", peerIP, + ).Scan(&exists) if err != nil { return err } if exists { _, err = tx.Exec( - "UPDATE bgp_peers SET peer_asn = ?, last_seen = ?, last_message_type = ? WHERE peer_ip = ?", + "UPDATE bgp_peers SET peer_asn = ?, last_seen = ?, "+ + "last_message_type = ? WHERE peer_ip = ?", peerASN, timestamp, messageType, peerIP, ) } else { _, err = tx.Exec( - "INSERT INTO bgp_peers (id, peer_ip, peer_asn, first_seen, last_seen, last_message_type) VALUES (?, ?, ?, ?, ?, ?)", - generateUUID().String(), peerIP, peerASN, timestamp, timestamp, messageType, + "INSERT INTO bgp_peers (id, peer_ip, peer_asn, first_seen, "+ + "last_seen, last_message_type) VALUES (?, ?, ?, ?, ?, ?)", + generateUUID().String(), peerIP, peerASN, timestamp, timestamp, + messageType, ) } + if err != nil { return err } - if err = tx.Commit(); err != nil { + err = tx.Commit() + if err != nil { d.logger.Error("Failed to commit transaction for peer update", "peer_ip", peerIP, "peer_asn", peerASN, @@ -1130,7 +1252,8 @@ func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) { d.counts.fill(&stats) // Database file size is a cheap stat() on the file. - if fileInfo, err := os.Stat(d.path); err != nil { + fileInfo, err := os.Stat(d.path) + if err != nil { d.logger.Warn("Failed to get database file size", "error", err) } else { stats.FileSizeBytes = fileInfo.Size() @@ -1153,64 +1276,6 @@ func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) { return stats, 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. -func (d *Database) routeTimestampRange(ctx context.Context) (oldest, newest *time.Time, err error) { - oldestV4, ok, err := d.scanRouteTimestamp(ctx, - "SELECT last_updated FROM live_routes_v4 ORDER BY last_updated ASC LIMIT 1") - if err != nil { - return nil, nil, err - } - if ok { - oldest = &oldestV4 - } - - oldestV6, ok, err := d.scanRouteTimestamp(ctx, - "SELECT last_updated FROM live_routes_v6 ORDER BY last_updated ASC LIMIT 1") - if err != nil { - return nil, nil, err - } - if ok && (oldest == nil || oldestV6.Before(*oldest)) { - oldest = &oldestV6 - } - - newestV4, ok, err := d.scanRouteTimestamp(ctx, - "SELECT last_updated FROM live_routes_v4 ORDER BY last_updated DESC LIMIT 1") - if err != nil { - return nil, nil, err - } - if ok { - newest = &newestV4 - } - - newestV6, ok, err := d.scanRouteTimestamp(ctx, - "SELECT last_updated FROM live_routes_v6 ORDER BY last_updated DESC LIMIT 1") - if err != nil { - return nil, nil, err - } - if ok && (newest == nil || newestV6.After(*newest)) { - newest = &newestV6 - } - - return oldest, newest, nil -} - -// scanRouteTimestamp runs a single-row timestamp query. ok is false when the -// table is empty. The query selects the last_updated column directly so the -// driver parses the DATETIME value into a time.Time. -func (d *Database) scanRouteTimestamp(ctx context.Context, query string) (ts time.Time, ok bool, err error) { - err = d.db.QueryRowContext(ctx, query).Scan(&ts) - switch { - case errors.Is(err, sql.ErrNoRows): - return time.Time{}, false, nil - case err != nil: - return time.Time{}, false, err - default: - return ts, true, nil - } -} - // UpsertLiveRoute inserts or updates a live route func (d *Database) UpsertLiveRoute(route *LiveRoute) error { d.lock("UpsertLiveRoute") @@ -1221,37 +1286,37 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error { return fmt.Errorf("failed to encode AS path: %w", err) } - updateSQL, insertSQL, lookupSQL := updateLiveRouteV4SQL, insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL + updateSQL, insertSQL, lookupSQL := updateLiveRouteV4SQL, + insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL if route.IPVersion == ipVersionV6 { - updateSQL, insertSQL, lookupSQL = updateLiveRouteV6SQL, insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL + updateSQL, insertSQL, lookupSQL = updateLiveRouteV6SQL, + 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. - upd, err := d.db.Prepare(updateSQL) + upd, err := d.db.PrepareContext(ctx, updateSQL) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } defer func() { _ = upd.Close() }() - ins, err := d.db.Prepare(insertSQL) + ins, err := d.db.PrepareContext(ctx, insertSQL) if err != nil { return fmt.Errorf("failed to prepare insert statement: %w", err) } defer func() { _ = ins.Close() }() - has, err := d.db.Prepare(lookupSQL) + has, err := d.db.PrepareContext(ctx, lookupSQL) if err != nil { return fmt.Errorf("failed to prepare prefix lookup statement: %w", err) } defer func() { _ = has.Close() }() - var inserted, newPrefix bool - if route.IPVersion == ipVersionV4 { - inserted, newPrefix, err = upsertRouteRowV4(upd, ins, has, route, string(pathJSON)) - } else { - inserted, newPrefix, err = upsertRouteRowV6(upd, ins, has, route, string(pathJSON)) - } + inserted, newPrefix, err := upsertRouteRow(ctx, upd, ins, has, route, + string(pathJSON)) if err != nil { return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) } @@ -1263,6 +1328,7 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error { d.counts.addRoutes(0, 1) } } + if newPrefix { if route.IPVersion == ipVersionV4 { d.counts.addToDistribution([]int{route.MaskLength}, nil, 1) @@ -1274,9 +1340,30 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error { return nil } +// deleteLiveRouteRows deletes the live routes for prefix and peerIP from the +// table of the prefix's IP version, only those from originASN unless it is 0. +// Literal per-table queries (rather than one formatted with the table name) +// keep the delete free of dynamically built SQL. +func deleteLiveRouteRows( + tx *loggingTx, isV4 bool, prefix string, originASN int, peerIP string, +) (sql.Result, error) { + switch { + case isV4 && originASN == 0: + return tx.Exec(deleteLiveRouteAnyOriginV4SQL, prefix, peerIP) + case isV4: + return tx.Exec(deleteLiveRouteV4SQL, prefix, originASN, peerIP) + case originASN == 0: + return tx.Exec(deleteLiveRouteAnyOriginV6SQL, prefix, peerIP) + default: + return tx.Exec(deleteLiveRouteV6SQL, prefix, originASN, peerIP) + } +} + // DeleteLiveRoute deletes a live route // If originASN is 0, deletes all routes for the prefix/peer combination -func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) error { +func (d *Database) DeleteLiveRoute( + prefix string, originASN int, peerIP string, +) error { d.lock("DeleteLiveRoute") defer d.unlock() @@ -1294,30 +1381,9 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } - defer func() { - if err := tx.Rollback(); err != nil && err != sql.ErrTxDone { - d.logger.Error("Failed to rollback transaction", "error", err) - } - }() + defer d.rollback(tx) - // Literal per-table queries (rather than one formatted with the table name) - // so the delete carries no dynamically built SQL. A delete with no origin - // ASN can remove several rows. - var res sql.Result - switch { - case isV4 && originASN == 0: - res, err = tx.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP) - case isV4: - res, err = tx.Exec( - `DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, - prefix, originASN, peerIP) - case originASN == 0: - res, err = tx.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP) - default: - res, err = tx.Exec( - `DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, - prefix, originASN, peerIP) - } + res, err := deleteLiveRouteRows(tx, isV4, prefix, originASN, peerIP) if err != nil { return err } @@ -1326,6 +1392,7 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) if err != nil { return err } + if affected == 0 { return nil } @@ -1335,12 +1402,16 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) if isV4 { lookupSQL = prefixHasLiveRouteV4SQL } + var prefixHasRoute bool - if err := tx.QueryRow(lookupSQL, prefix).Scan(&prefixHasRoute); err != nil { + + err = tx.QueryRow(lookupSQL, prefix).Scan(&prefixHasRoute) + if err != nil { return err } - if err := tx.Commit(); err != nil { + err = tx.Commit() + if err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } @@ -1349,6 +1420,7 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) } else { d.counts.addRoutes(0, -int(affected)) } + if !prefixHasRoute { maskLength, _ := ipnet.Mask.Size() if isV4 { @@ -1361,16 +1433,22 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) return nil } -// GetPrefixDistribution returns the distribution of unique prefixes by mask length -func (d *Database) GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) { +// GetPrefixDistribution returns the distribution of unique prefixes by mask +// length +func (d *Database) GetPrefixDistribution() ( + []PrefixDistribution, []PrefixDistribution, error, +) { return d.GetPrefixDistributionContext(context.Background()) } -// GetPrefixDistributionContext returns the distribution of unique prefixes by mask length with context support. -// It reads every live route, so the stats read does not call it; it seeds the -// in-memory distribution once at startup. +// GetPrefixDistributionContext returns the distribution of unique prefixes by +// mask length with context support. It reads every live route, so the stats +// read does not call it; it seeds the in-memory distribution once at startup. func (d *Database) GetPrefixDistributionContext(ctx context.Context) ( - ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) { + []PrefixDistribution, []PrefixDistribution, error, +) { + var ipv4, ipv6 []PrefixDistribution + // IPv4 distribution - count unique prefixes from v4 table query := ` SELECT mask_length, COUNT(DISTINCT prefix) as count @@ -1378,6 +1456,7 @@ func (d *Database) GetPrefixDistributionContext(ctx context.Context) ( GROUP BY mask_length ORDER BY mask_length ` + rows4, err := d.db.QueryContext(ctx, query) if err != nil { return nil, nil, fmt.Errorf("failed to query IPv4 distribution: %w", err) @@ -1390,13 +1469,19 @@ func (d *Database) GetPrefixDistributionContext(ctx context.Context) ( for rows4.Next() { var dist PrefixDistribution - if err := rows4.Scan(&dist.MaskLength, &dist.Count); err != nil { - return nil, nil, fmt.Errorf("failed to scan IPv4 distribution: %w", err) + + err := rows4.Scan(&dist.MaskLength, &dist.Count) + if err != nil { + return nil, nil, fmt.Errorf("failed to scan IPv4 distribution: %w", + err) } + ipv4 = append(ipv4, dist) } + // A read that stops partway ends the loop without an error of its own. - if err := rows4.Err(); err != nil { + err = rows4.Err() + if err != nil { return nil, nil, fmt.Errorf("failed to read IPv4 distribution: %w", err) } @@ -1407,6 +1492,7 @@ func (d *Database) GetPrefixDistributionContext(ctx context.Context) ( GROUP BY mask_length ORDER BY mask_length ` + rows6, err := d.db.QueryContext(ctx, query) if err != nil { return nil, nil, fmt.Errorf("failed to query IPv6 distribution: %w", err) @@ -1419,12 +1505,18 @@ func (d *Database) GetPrefixDistributionContext(ctx context.Context) ( for rows6.Next() { var dist PrefixDistribution - if err := rows6.Scan(&dist.MaskLength, &dist.Count); err != nil { - return nil, nil, fmt.Errorf("failed to scan IPv6 distribution: %w", err) + + err := rows6.Scan(&dist.MaskLength, &dist.Count) + if err != nil { + return nil, nil, fmt.Errorf("failed to scan IPv6 distribution: %w", + err) } + ipv6 = append(ipv6, dist) } - if err := rows6.Err(); err != nil { + + err = rows6.Err() + if err != nil { return nil, nil, fmt.Errorf("failed to read IPv6 distribution: %w", err) } @@ -1432,20 +1524,27 @@ func (d *Database) GetPrefixDistributionContext(ctx context.Context) ( } // GetLiveRouteCounts returns the count of IPv4 and IPv6 routes -func (d *Database) GetLiveRouteCounts() (ipv4Count, ipv6Count int, err error) { +func (d *Database) GetLiveRouteCounts() (int, int, error) { return d.GetLiveRouteCountsContext(context.Background()) } -// GetLiveRouteCountsContext returns the count of IPv4 and IPv6 routes with context support -func (d *Database) GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error) { +// GetLiveRouteCountsContext returns the count of IPv4 and IPv6 routes with +// context support +func (d *Database) GetLiveRouteCountsContext( + ctx context.Context, +) (int, int, error) { + var ipv4Count, ipv6Count int + // Get IPv4 count from dedicated table - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v4").Scan(&ipv4Count) + err := d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v4"). + Scan(&ipv4Count) if err != nil { return 0, 0, fmt.Errorf("failed to count IPv4 routes: %w", err) } // Get IPv6 count from dedicated table - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v6").Scan(&ipv6Count) + err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v6"). + Scan(&ipv6Count) if err != nil { return 0, 0, fmt.Errorf("failed to count IPv6 routes: %w", err) } @@ -1458,18 +1557,25 @@ func (d *Database) GetASDetails(asn int) (*ASN, []LiveRoute, error) { return d.GetASDetailsContext(context.Background(), asn) } -// GetASDetailsContext returns detailed information about an AS including prefixes with context support -func (d *Database) GetASDetailsContext(ctx context.Context, asn int) (*ASN, []LiveRoute, error) { +// GetASDetailsContext returns detailed information about an AS including +// prefixes with context support +func (d *Database) GetASDetailsContext( + ctx context.Context, asn int, +) (*ASN, []LiveRoute, error) { // Get AS information - var asnInfo ASN - var handle, description sql.NullString - err := d.db.QueryRowContext(ctx, - "SELECT asn, handle, description, first_seen, last_seen FROM asns WHERE asn = ?", - asn, - ).Scan(&asnInfo.ASN, &handle, &description, &asnInfo.FirstSeen, &asnInfo.LastSeen) + var ( + asnInfo ASN + handle, description sql.NullString + ) + err := d.db.QueryRowContext(ctx, + "SELECT asn, handle, description, first_seen, last_seen "+ + "FROM asns WHERE asn = ?", + asn, + ).Scan(&asnInfo.ASN, &handle, &description, &asnInfo.FirstSeen, + &asnInfo.LastSeen) if err != nil { - if err == sql.ErrNoRows { + if errors.Is(err, sql.ErrNoRows) { return nil, nil, fmt.Errorf("%w: AS%d", ErrNoRoute, asn) } @@ -1480,9 +1586,6 @@ func (d *Database) GetASDetailsContext(ctx context.Context, asn int) (*ASN, []Li asnInfo.Description = description.String // Get prefixes announced by this AS from both tables - var allPrefixes []LiveRoute - - // Query IPv4 prefixes queryV4 := ` SELECT prefix, mask_length, MAX(last_updated) as last_updated FROM live_routes_v4 @@ -1490,38 +1593,11 @@ func (d *Database) GetASDetailsContext(ctx context.Context, asn int) (*ASN, []Li GROUP BY prefix, mask_length ` - rows4, err := d.db.QueryContext(ctx, queryV4, asn) + allPrefixes, err := d.originPrefixes(ctx, queryV4, asn, ipVersionV4) if err != nil { - return &asnInfo, nil, fmt.Errorf("failed to query IPv4 prefixes: %w", err) - } - defer func() { _ = rows4.Close() }() - - for rows4.Next() { - var route LiveRoute - var lastUpdatedStr string - err := rows4.Scan(&route.Prefix, &route.MaskLength, &lastUpdatedStr) - if err != nil { - d.logger.Error("Failed to scan IPv4 prefix row", "error", err, "asn", asn) - - continue - } - // Parse the timestamp string - route.LastUpdated, err = time.Parse("2006-01-02 15:04:05-07:00", lastUpdatedStr) - if err != nil { - // Try without timezone - route.LastUpdated, err = time.Parse("2006-01-02 15:04:05", lastUpdatedStr) - if err != nil { - d.logger.Error("Failed to parse timestamp", "error", err, "timestamp", lastUpdatedStr) - - continue - } - } - route.OriginASN = asn - route.IPVersion = ipVersionV4 - allPrefixes = append(allPrefixes, route) + return &asnInfo, nil, err } - // Query IPv6 prefixes queryV6 := ` SELECT prefix, mask_length, MAX(last_updated) as last_updated FROM live_routes_v6 @@ -1529,41 +1605,18 @@ func (d *Database) GetASDetailsContext(ctx context.Context, asn int) (*ASN, []Li GROUP BY prefix, mask_length ` - rows6, err := d.db.QueryContext(ctx, queryV6, asn) + prefixesV6, err := d.originPrefixes(ctx, queryV6, asn, ipVersionV6) if err != nil { - return &asnInfo, allPrefixes, fmt.Errorf("failed to query IPv6 prefixes: %w", err) - } - defer func() { _ = rows6.Close() }() - - for rows6.Next() { - var route LiveRoute - var lastUpdatedStr string - err := rows6.Scan(&route.Prefix, &route.MaskLength, &lastUpdatedStr) - if err != nil { - d.logger.Error("Failed to scan IPv6 prefix row", "error", err, "asn", asn) - - continue - } - // Parse the timestamp string - route.LastUpdated, err = time.Parse("2006-01-02 15:04:05-07:00", lastUpdatedStr) - if err != nil { - // Try without timezone - route.LastUpdated, err = time.Parse("2006-01-02 15:04:05", lastUpdatedStr) - if err != nil { - d.logger.Error("Failed to parse timestamp", "error", err, "timestamp", lastUpdatedStr) - - continue - } - } - route.OriginASN = asn - route.IPVersion = ipVersionV6 - allPrefixes = append(allPrefixes, route) + return &asnInfo, allPrefixes, err } - return &asnInfo, allPrefixes, nil + return &asnInfo, append(allPrefixes, prefixesV6...), nil } -// ASPeer represents a peering relationship with another AS including handle, description, and timestamps. +// ASPeer represents a peering relationship with another AS including handle, +// description, and timestamps. +// +//nolint:tagliatelle // field names are the existing JSON API type ASPeer struct { ASN int `json:"asn"` Handle string `json:"handle"` @@ -1577,8 +1630,11 @@ func (d *Database) GetASPeers(asn int) ([]ASPeer, error) { return d.GetASPeersContext(context.Background(), asn) } -// GetASPeersContext returns all ASes that peer with the given AS with context support -func (d *Database) GetASPeersContext(ctx context.Context, asn int) ([]ASPeer, error) { +// GetASPeersContext returns all ASes that peer with the given AS with context +// support +func (d *Database) GetASPeersContext( + ctx context.Context, asn int, +) ([]ASPeer, error) { query := ` SELECT CASE @@ -1605,18 +1661,23 @@ func (d *Database) GetASPeersContext(ctx context.Context, asn int) ([]ASPeer, er defer func() { _ = rows.Close() }() var peers []ASPeer + for rows.Next() { var peer ASPeer - err := rows.Scan(&peer.ASN, &peer.Handle, &peer.Description, &peer.FirstSeen, &peer.LastSeen) + + err := rows.Scan(&peer.ASN, &peer.Handle, &peer.Description, + &peer.FirstSeen, &peer.LastSeen) if err != nil { d.logger.Error("Failed to scan peer row", "error", err, "asn", asn) continue } + peers = append(peers, peer) } - if err := rows.Err(); err != nil { + err = rows.Err() + if err != nil { return nil, fmt.Errorf("error iterating AS peers: %w", err) } @@ -1628,25 +1689,28 @@ func (d *Database) GetPrefixDetails(prefix string) ([]LiveRoute, error) { return d.GetPrefixDetailsContext(context.Background(), prefix) } -// GetPrefixDetailsContext returns detailed information about a prefix with context support -func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error) { +// GetPrefixDetailsContext returns detailed information about a prefix with +// context support +func (d *Database) GetPrefixDetailsContext( + ctx context.Context, prefix string, +) ([]LiveRoute, error) { // Determine if it's IPv4 or IPv6 by parsing the prefix _, ipnet, err := net.ParseCIDR(prefix) if err != nil { return nil, fmt.Errorf("invalid prefix format: %w", err) } - tableName := "live_routes_v4" + tableName := tableLiveRoutesV4 ipVersion := ipVersionV4 + if ipnet.IP.To4() == nil { - tableName = "live_routes_v6" + tableName = tableLiveRoutesV6 ipVersion = ipVersionV6 } - //nolint:gosec // Table name is hardcoded based on IP version query := fmt.Sprintf(` - SELECT lr.origin_asn, lr.peer_ip, lr.as_path, lr.next_hop, lr.last_updated, - a.handle, a.description + SELECT lr.origin_asn, lr.peer_ip, lr.as_path, lr.next_hop, + lr.last_updated, a.handle, a.description FROM %s lr LEFT JOIN asns a ON a.asn = lr.origin_asn WHERE lr.prefix = ? @@ -1660,10 +1724,13 @@ func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ( defer func() { _ = rows.Close() }() var routes []LiveRoute + for rows.Next() { - var route LiveRoute - var pathJSON string - var handle, description sql.NullString + var ( + route LiveRoute + pathJSON string + handle, description sql.NullString + ) err := rows.Scan( &route.OriginASN, &route.PeerIP, &pathJSON, &route.NextHop, @@ -1674,7 +1741,8 @@ func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ( } // Decode AS path - if err := json.Unmarshal([]byte(pathJSON), &route.ASPath); err != nil { + err = json.Unmarshal([]byte(pathJSON), &route.ASPath) + if err != nil { route.ASPath = []int{} } @@ -1684,6 +1752,11 @@ func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ( routes = append(routes, route) } + err = rows.Err() + if err != nil { + return nil, fmt.Errorf("failed to read prefix details: %w", err) + } + if len(routes) == 0 { return nil, fmt.Errorf("%w: %s", ErrNoRoute, prefix) } @@ -1691,21 +1764,26 @@ func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ( return routes, nil } -// GetRandomPrefixesByLength returns a random sample of prefixes with the specified mask length -func (d *Database) GetRandomPrefixesByLength(maskLength, ipVersion, limit int) ([]LiveRoute, error) { - return d.GetRandomPrefixesByLengthContext(context.Background(), maskLength, ipVersion, limit) +// GetRandomPrefixesByLength returns a random sample of prefixes with the +// specified mask length +func (d *Database) GetRandomPrefixesByLength( + maskLength, ipVersion, limit int, +) ([]LiveRoute, error) { + return d.GetRandomPrefixesByLengthContext(context.Background(), + maskLength, ipVersion, limit) } -// GetRandomPrefixesByLengthContext returns a random sample of prefixes with context support +// GetRandomPrefixesByLengthContext returns a random sample of prefixes with +// context support func (d *Database) GetRandomPrefixesByLengthContext( - ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error) { + ctx context.Context, maskLength, ipVersion, limit int, +) ([]LiveRoute, error) { // Select unique prefixes with their most recent route information - tableName := "live_routes_v4" + tableName := tableLiveRoutesV4 if ipVersion == ipVersionV6 { - tableName = "live_routes_v6" + tableName = tableLiveRoutesV6 } - //nolint:gosec // Table name is hardcoded based on IP version query := fmt.Sprintf(` WITH unique_prefixes AS ( SELECT prefix, MAX(last_updated) as max_updated @@ -1715,10 +1793,11 @@ func (d *Database) GetRandomPrefixesByLengthContext( ORDER BY RANDOM() LIMIT ? ) - SELECT lr.prefix, lr.mask_length, lr.origin_asn, lr.as_path, + SELECT lr.prefix, lr.mask_length, lr.origin_asn, lr.as_path, lr.peer_ip, lr.last_updated FROM %s lr - INNER JOIN unique_prefixes up ON lr.prefix = up.prefix AND lr.last_updated = up.max_updated + INNER JOIN unique_prefixes up + ON lr.prefix = up.prefix AND lr.last_updated = up.max_updated WHERE lr.mask_length = ? `, tableName, tableName) @@ -1731,9 +1810,13 @@ func (d *Database) GetRandomPrefixesByLengthContext( }() var routes []LiveRoute + for rows.Next() { - var route LiveRoute - var pathJSON string + var ( + route LiveRoute + pathJSON string + ) + err := rows.Scan( &route.Prefix, &route.MaskLength, @@ -1744,23 +1827,32 @@ func (d *Database) GetRandomPrefixesByLengthContext( ) // Set IP version based on which table we queried route.IPVersion = ipVersion + if err != nil { continue } // Decode AS path - if err := json.Unmarshal([]byte(pathJSON), &route.ASPath); err != nil { + err = json.Unmarshal([]byte(pathJSON), &route.ASPath) + if err != nil { route.ASPath = []int{} } routes = append(routes, route) } + err = rows.Err() + if err != nil { + return nil, fmt.Errorf("failed to read random prefixes: %w", err) + } + return routes, nil } // GetNextStaleASN returns a random ASN that needs WHOIS data refresh. -func (d *Database) GetNextStaleASN(ctx context.Context, staleThreshold time.Duration) (int, error) { +func (d *Database) GetNextStaleASN( + ctx context.Context, staleThreshold time.Duration, +) (int, error) { cutoff := time.Now().Add(-staleThreshold) // Select a random stale ASN using ORDER BY RANDOM() @@ -1773,9 +1865,10 @@ func (d *Database) GetNextStaleASN(ctx context.Context, staleThreshold time.Dura ` var asn int + err := d.db.QueryRowContext(ctx, query, cutoff).Scan(&asn) if err != nil { - if err == sql.ErrNoRows { + if errors.Is(err, sql.ErrNoRows) { return 0, ErrNoStaleASN } @@ -1786,6 +1879,8 @@ func (d *Database) GetNextStaleASN(ctx context.Context, staleThreshold time.Dura } // WHOISStats contains statistics about WHOIS data freshness. +// +//nolint:tagliatelle // field names are the existing JSON API type WHOISStats struct { TotalASNs int `json:"total_asns"` StaleASNs int `json:"stale_asns"` @@ -1794,19 +1889,25 @@ type WHOISStats struct { } // GetWHOISStats returns statistics about WHOIS data freshness. -func (d *Database) GetWHOISStats(ctx context.Context, staleThreshold time.Duration) (*WHOISStats, error) { +func (d *Database) GetWHOISStats( + ctx context.Context, staleThreshold time.Duration, +) (*WHOISStats, error) { cutoff := time.Now().Add(-staleThreshold) query := ` SELECT COUNT(*) as total, - COALESCE(SUM(CASE WHEN whois_updated_at IS NULL THEN 1 ELSE 0 END), 0) as never_fetched, - COALESCE(SUM(CASE WHEN whois_updated_at IS NOT NULL AND whois_updated_at < ? THEN 1 ELSE 0 END), 0) as stale, - COALESCE(SUM(CASE WHEN whois_updated_at IS NOT NULL AND whois_updated_at >= ? THEN 1 ELSE 0 END), 0) as fresh + COALESCE(SUM(CASE WHEN whois_updated_at IS NULL + THEN 1 ELSE 0 END), 0) as never_fetched, + COALESCE(SUM(CASE WHEN whois_updated_at IS NOT NULL + AND whois_updated_at < ? THEN 1 ELSE 0 END), 0) as stale, + COALESCE(SUM(CASE WHEN whois_updated_at IS NOT NULL + AND whois_updated_at >= ? THEN 1 ELSE 0 END), 0) as fresh FROM asns ` var stats WHOISStats + err := d.db.QueryRowContext(ctx, query, cutoff, cutoff).Scan( &stats.TotalASNs, &stats.NeverFetched, @@ -1821,7 +1922,9 @@ func (d *Database) GetWHOISStats(ctx context.Context, staleThreshold time.Durati } // UpdateASNWHOIS updates an ASN record with WHOIS data. -func (d *Database) UpdateASNWHOIS(ctx context.Context, update *ASNWHOISUpdate) error { +func (d *Database) UpdateASNWHOIS( + ctx context.Context, update *ASNWHOISUpdate, +) error { d.lock("UpdateASNWHOIS") defer d.unlock() @@ -1873,38 +1976,17 @@ func (d *Database) GetIPInfo(ip string) (*IPInfo, error) { return d.GetIPInfoContext(context.Background(), ip) } -// GetIPInfoContext returns comprehensive IP information with context support. -// -// It finds the most specific live route for the address with point lookups on -// the prefix index: for each mask length from the longest (32 or 128) down to 0 -// it looks up the address's network prefix at that length, and the first length -// with a live route wins. That is at most 33 or 129 indexed lookups, however -// many routes are live. The prefix is looked up in the text form net/netip -// prints, which is the form the prefix handler stores. -func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error) { - addr, err := netip.ParseAddr(ip) - if err != nil { - return nil, fmt.Errorf("%w: %s", ErrInvalidIP, ip) - } - // An IPv4 address written in IPv6 form (::ffff:192.0.2.1) is an IPv4 lookup. - addr = addr.Unmap() - - info := &IPInfo{IP: ip, IPVersion: ipVersionV6} - routesTable, prefixesTable := "live_routes_v6", "prefixes_v6" - if addr.Is4() { - info.IPVersion = ipVersionV4 - routesTable, prefixesTable = "live_routes_v4", "prefixes_v4" - } - - // Route info with peer count and prefix first_seen - //nolint:gosec // Table names are hardcoded based on IP version - query := fmt.Sprintf(` +// ipInfoQuery reads the live route for one prefix with its peer count, the +// prefix's first_seen and the origin AS's details. %[1]s is the live route +// table and %[2]s the prefix table of the address's IP version. +const ipInfoQuery = ` SELECT lr.prefix, lr.mask_length, lr.origin_asn, lr.last_updated, - (SELECT COUNT(DISTINCT peer_ip) FROM %[1]s WHERE prefix = lr.prefix) as num_peers, + (SELECT COUNT(DISTINCT peer_ip) FROM %[1]s + WHERE prefix = lr.prefix) as num_peers, p.first_seen, a.handle, a.description, @@ -1921,7 +2003,36 @@ func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, er LEFT JOIN asns a ON a.asn = lr.origin_asn WHERE lr.prefix = ? LIMIT 1 - `, routesTable, prefixesTable) + ` + +// GetIPInfoContext returns comprehensive IP information with context support. +// +// It finds the most specific live route for the address with point lookups on +// the prefix index: for each mask length from the longest (32 or 128) down to 0 +// it looks up the address's network prefix at that length, and the first length +// with a live route wins. That is at most 33 or 129 indexed lookups, however +// many routes are live. The prefix is looked up in the text form net/netip +// prints, which is the form the prefix handler stores. +func (d *Database) GetIPInfoContext( + ctx context.Context, ip string, +) (*IPInfo, error) { + addr, err := netip.ParseAddr(ip) + if err != nil { + return nil, fmt.Errorf("%w: %s", ErrInvalidIP, ip) + } + // An IPv4 address written in IPv6 form (::ffff:192.0.2.1) is an IPv4 lookup. + addr = addr.Unmap() + + info := &IPInfo{IP: ip, IPVersion: ipVersionV6} + + routesTable, prefixesTable := tableLiveRoutesV6, "prefixes_v6" + if addr.Is4() { + routesTable, prefixesTable = tableLiveRoutesV4, "prefixes_v4" + info.IPVersion = ipVersionV4 + } + + // Route info with peer count and prefix first_seen + query := fmt.Sprintf(ipInfoQuery, routesTable, prefixesTable) stmt, err := d.db.PrepareContext(ctx, query) if err != nil { @@ -1929,41 +2040,36 @@ func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, er } defer func() { _ = stmt.Close() }() - var handle, description, asName, orgName, orgID, address, countryCode, abuseEmail, rir sql.NullString - var prefixFirstSeen sql.NullTime - var whoisUpdatedAt sql.NullTime + var ( + handle, description, asName, orgName, orgID sql.NullString + address, countryCode, abuseEmail, rir sql.NullString + prefixFirstSeen, whoisUpdatedAt sql.NullTime + ) found := false + for bits := addr.BitLen(); bits >= 0; bits-- { prefix := netip.PrefixFrom(addr, bits).Masked().String() + err := stmt.QueryRowContext(ctx, prefix).Scan( - &info.Netblock, - &info.MaskLength, - &info.ASN, - &info.LastSeen, - &info.NumPeers, - &prefixFirstSeen, - &handle, - &description, - &asName, - &orgName, - &orgID, - &address, - &countryCode, - &abuseEmail, - &rir, - &whoisUpdatedAt, + &info.Netblock, &info.MaskLength, &info.ASN, &info.LastSeen, + &info.NumPeers, &prefixFirstSeen, &handle, &description, + &asName, &orgName, &orgID, &address, &countryCode, &abuseEmail, + &rir, &whoisUpdatedAt, ) if errors.Is(err, sql.ErrNoRows) { continue } + if err != nil { return nil, fmt.Errorf("failed to query routes: %w", err) } + found = true break } + if !found { return nil, fmt.Errorf("%w for IP %s", ErrNoRoute, ip) } @@ -1984,18 +2090,23 @@ func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, er // Check if WHOIS data needs refresh (never fetched or older than 30 days) const staleThreshold = 30 * 24 * time.Hour - info.NeedsWHOISRefresh = !whoisUpdatedAt.Valid || time.Since(whoisUpdatedAt.Time) > staleThreshold + + info.NeedsWHOISRefresh = !whoisUpdatedAt.Valid || + time.Since(whoisUpdatedAt.Time) > staleThreshold return info, nil } -// Vacuum runs incremental vacuum to reclaim unused pages without blocking writes. -// It frees up to the specified number of pages per call (0 = all freeable pages). +// Vacuum runs incremental vacuum to reclaim unused pages without blocking +// writes. It frees up to the specified number of pages per call (0 = all +// freeable pages). func (d *Database) Vacuum(ctx context.Context) error { // Free up to 1000 pages per call (~4MB with default 4KB page size) // This keeps each vacuum operation quick and non-blocking const pagesToFree = 1000 - rows, err := d.db.QueryContext(ctx, fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree)) + + rows, err := d.db.QueryContext(ctx, + fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree)) if err != nil { return fmt.Errorf("failed to run incremental vacuum: %w", err) } @@ -2006,7 +2117,9 @@ func (d *Database) Vacuum(ctx context.Context) error { // than one page. for rows.Next() { } - if err := rows.Err(); err != nil { + + err = rows.Err() + if err != nil { return fmt.Errorf("failed to run incremental vacuum: %w", err) } @@ -2038,6 +2151,7 @@ func (d *Database) Checkpoint(ctx context.Context) error { // Ping checks if the database connection is alive with a lightweight query. func (d *Database) Ping(ctx context.Context) error { var result int + err := d.db.QueryRowContext(ctx, "SELECT 1").Scan(&result) if err != nil { return fmt.Errorf("database ping failed: %w", err) @@ -2045,3 +2159,185 @@ func (d *Database) Ping(ctx context.Context) error { return nil } + +// lock acquires the database mutex and logs debug information +func (d *Database) lock(operation string) { + // Get caller information + _, file, line, _ := runtime.Caller(1) + caller := fmt.Sprintf("%s:%d", filepath.Base(file), line) + + d.logger.Debug("Acquiring database lock", + "operation", operation, "caller", caller) + + d.mu.Lock() + d.lockedAt = time.Now() + d.lockedBy = fmt.Sprintf("%s (%s)", operation, caller) + + d.logger.Debug("Database lock acquired", + "operation", operation, "caller", caller) +} + +// unlock releases the database mutex and logs debug information including +// hold duration +func (d *Database) unlock() { + holdDuration := time.Since(d.lockedAt) + lockedBy := d.lockedBy + + d.lockedAt = time.Time{} + d.lockedBy = "" + d.mu.Unlock() + + d.logger.Debug("Database lock released", + "held_by", lockedBy, "duration_ms", holdDuration.Milliseconds()) +} + +// beginTx starts a new transaction with logging +func (d *Database) beginTx() (*loggingTx, error) { + tx, err := d.db.BeginTx(context.Background(), nil) + if err != nil { + return nil, err + } + + return &loggingTx{Tx: tx, logger: d.logger}, nil +} + +// rollback rolls back tx unless it has already been committed, and logs a +// failure. Every write defers it right after beginning its transaction. +func (d *Database) rollback(tx *loggingTx) { + err := tx.Rollback() + if err != nil && !errors.Is(err, sql.ErrTxDone) { + d.logger.Error("Failed to rollback transaction", "error", err) + } +} + +// 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. +func (d *Database) routeTimestampRange( + ctx context.Context, +) (*time.Time, *time.Time, error) { + var oldest, newest *time.Time + + oldestV4, ok, err := d.scanRouteTimestamp(ctx, "SELECT last_updated "+ + "FROM live_routes_v4 ORDER BY last_updated ASC LIMIT 1") + if err != nil { + return nil, nil, err + } + + if ok { + oldest = &oldestV4 + } + + oldestV6, ok, err := d.scanRouteTimestamp(ctx, "SELECT last_updated "+ + "FROM live_routes_v6 ORDER BY last_updated ASC LIMIT 1") + if err != nil { + return nil, nil, err + } + + if ok && (oldest == nil || oldestV6.Before(*oldest)) { + oldest = &oldestV6 + } + + newestV4, ok, err := d.scanRouteTimestamp(ctx, "SELECT last_updated "+ + "FROM live_routes_v4 ORDER BY last_updated DESC LIMIT 1") + if err != nil { + return nil, nil, err + } + + if ok { + newest = &newestV4 + } + + newestV6, ok, err := d.scanRouteTimestamp(ctx, "SELECT last_updated "+ + "FROM live_routes_v6 ORDER BY last_updated DESC LIMIT 1") + if err != nil { + return nil, nil, err + } + + if ok && (newest == nil || newestV6.After(*newest)) { + newest = &newestV6 + } + + return oldest, newest, nil +} + +// scanRouteTimestamp runs a single-row timestamp query. The bool is false when +// the table is empty. The query selects the last_updated column directly so the +// driver parses the DATETIME value into a time.Time. +func (d *Database) scanRouteTimestamp( + ctx context.Context, query string, +) (time.Time, bool, error) { + var ts time.Time + + err := d.db.QueryRowContext(ctx, query).Scan(&ts) + switch { + case errors.Is(err, sql.ErrNoRows): + return time.Time{}, false, nil + case err != nil: + return time.Time{}, false, err + default: + return ts, true, nil + } +} + +// originPrefixes runs query, which selects the prefix, mask length and latest +// last_updated of every prefix asn originates in the live route table of +// ipVersion. Rows that fail to scan or whose timestamp does not parse are +// logged and skipped. +func (d *Database) originPrefixes( + ctx context.Context, query string, asn, ipVersion int, +) ([]LiveRoute, error) { + family := ipv4Family + if ipVersion == ipVersionV6 { + family = ipv6Family + } + + rows, err := d.db.QueryContext(ctx, query, asn) + if err != nil { + return nil, fmt.Errorf("failed to query %s prefixes: %w", family, err) + } + defer func() { _ = rows.Close() }() + + var routes []LiveRoute + + for rows.Next() { + var ( + route LiveRoute + lastUpdatedStr string + ) + + err := rows.Scan(&route.Prefix, &route.MaskLength, &lastUpdatedStr) + if err != nil { + d.logger.Error("Failed to scan "+family+" prefix row", + "error", err, "asn", asn) + + continue + } + + // Parse the timestamp string + route.LastUpdated, err = time.Parse("2006-01-02 15:04:05-07:00", + lastUpdatedStr) + if err != nil { + // Try without timezone + route.LastUpdated, err = time.Parse("2006-01-02 15:04:05", + lastUpdatedStr) + if err != nil { + d.logger.Error("Failed to parse timestamp", + "error", err, "timestamp", lastUpdatedStr) + + continue + } + } + + route.OriginASN = asn + route.IPVersion = ipVersion + routes = append(routes, route) + } + + err = rows.Err() + if err != nil { + return nil, fmt.Errorf("failed to read %s prefixes: %w", family, err) + } + + return routes, nil +} diff --git a/internal/database/interface.go b/internal/database/interface.go index 89cd4f0..727aef1 100644 --- a/internal/database/interface.go +++ b/internal/database/interface.go @@ -29,6 +29,8 @@ type Stats struct { // Store defines the interface for database operations. It provides methods for // managing ASNs, prefixes, announcements, peerings, BGP peers, and live routes. // Implementations must be safe for concurrent use. +// +//nolint:interfacebloat // the one seam handlers and tests depend on type Store interface { // ASN operations GetOrCreateASN(number int, timestamp time.Time) (*ASN, error) @@ -57,8 +59,10 @@ type Store interface { UpsertLiveRouteBatch(routes []*LiveRoute) error DeleteLiveRoute(prefix string, originASN int, peerIP string) error DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error - GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) - GetPrefixDistributionContext(ctx context.Context) (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) + GetPrefixDistribution() ( + ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) + GetPrefixDistributionContext(ctx context.Context) ( + ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) GetLiveRouteCounts() (ipv4Count, ipv6Count int, err error) GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error) @@ -79,7 +83,9 @@ type Store interface { GetPrefixDetails(prefix string) ([]LiveRoute, error) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error) GetRandomPrefixesByLength(maskLength, ipVersion, limit int) ([]LiveRoute, error) - GetRandomPrefixesByLengthContext(ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error) + GetRandomPrefixesByLengthContext( + ctx context.Context, maskLength, ipVersion, limit int, + ) ([]LiveRoute, error) // Lifecycle Close() error diff --git a/internal/database/slowquery.go b/internal/database/slowquery.go index 1ac302a..c082a5e 100644 --- a/internal/database/slowquery.go +++ b/internal/database/slowquery.go @@ -19,29 +19,21 @@ func logSlowQuery(logger *logger.Logger, query string, start time.Time) { } // queryRow wraps QueryRow with slow query logging -// nolint:unused // kept for consistency with other query wrappers -func (d *Database) queryRow(query string, args ...interface{}) *sql.Row { +// +//nolint:unused // kept for consistency with other query wrappers +func (d *Database) queryRow(query string, args ...any) *sql.Row { start := time.Now() defer logSlowQuery(d.logger, query, start) - return d.db.QueryRow(query, args...) -} - -// query wraps Query with slow query logging -// nolint:unused // kept for future use to ensure all queries go through slow query logging -func (d *Database) query(query string, args ...interface{}) (*sql.Rows, error) { - start := time.Now() - defer logSlowQuery(d.logger, query, start) - - return d.db.Query(query, args...) + return d.db.QueryRowContext(context.Background(), query, args...) } // exec wraps Exec with slow query logging -func (d *Database) exec(query string, args ...interface{}) error { +func (d *Database) exec(query string, args ...any) error { start := time.Now() defer logSlowQuery(d.logger, query, start) - _, err := d.db.Exec(query, args...) + _, err := d.db.ExecContext(context.Background(), query, args...) return err } @@ -49,35 +41,22 @@ func (d *Database) exec(query string, args ...interface{}) error { // loggingTx wraps sql.Tx to log slow queries type loggingTx struct { *sql.Tx + logger *logger.Logger } // QueryRow wraps sql.Tx.QueryRow to log slow queries -func (tx *loggingTx) QueryRow(query string, args ...interface{}) *sql.Row { +func (tx *loggingTx) QueryRow(query string, args ...any) *sql.Row { start := time.Now() defer logSlowQuery(tx.logger, query, start) - return tx.Tx.QueryRow(query, args...) -} - -// Query wraps sql.Tx.Query to log slow queries -func (tx *loggingTx) Query(query string, args ...interface{}) (*sql.Rows, error) { - start := time.Now() - defer logSlowQuery(tx.logger, query, start) - - return tx.Tx.Query(query, args...) -} - -// QueryContext wraps sql.Tx.QueryContext to log slow queries -func (tx *loggingTx) QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) { - start := time.Now() - defer logSlowQuery(tx.logger, query, start) - - return tx.Tx.QueryContext(ctx, query, args...) + return tx.Tx.QueryRowContext(context.Background(), query, args...) } // QueryRowContext wraps sql.Tx.QueryRowContext to log slow queries -func (tx *loggingTx) QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row { +func (tx *loggingTx) QueryRowContext( + ctx context.Context, query string, args ...any, +) *sql.Row { start := time.Now() defer logSlowQuery(tx.logger, query, start) @@ -85,15 +64,17 @@ func (tx *loggingTx) QueryRowContext(ctx context.Context, query string, args ... } // Exec wraps sql.Tx.Exec to log slow queries -func (tx *loggingTx) Exec(query string, args ...interface{}) (sql.Result, error) { +func (tx *loggingTx) Exec(query string, args ...any) (sql.Result, error) { start := time.Now() defer logSlowQuery(tx.logger, query, start) - return tx.Tx.Exec(query, args...) + return tx.Tx.ExecContext(context.Background(), query, args...) } // ExecContext wraps sql.Tx.ExecContext to log slow queries -func (tx *loggingTx) ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) { +func (tx *loggingTx) ExecContext( + ctx context.Context, query string, args ...any, +) (sql.Result, error) { start := time.Now() defer logSlowQuery(tx.logger, query, start) diff --git a/internal/database/utils.go b/internal/database/utils.go index 8094c78..4ba0b0d 100644 --- a/internal/database/utils.go +++ b/internal/database/utils.go @@ -27,6 +27,7 @@ func prefixMaskLength(prefix string) (int, error) { if err != nil { return 0, fmt.Errorf("invalid prefix %s: %w", prefix, err) } + maskLength, _ := network.Mask.Size() return maskLength, nil