From 698390f66cc375038f62830366e87554a15358b8 Mon Sep 17 00:00:00 2001 From: sneak Date: Tue, 22 Sep 2026 07:23:43 +0000 Subject: [PATCH] Serve /api/v1/stats counts from realtime in-memory counters (closes #27) The stats endpoints recomputed every count with a COUNT(*) over each table plus a MIN/MAX union scan on every request, which took the whole timeout and returned 500 once the database passed a few GiB. The row counts (ASNs, prefixes, peerings, peers, live routes) now live in memory: seeded once at startup and adjusted on each write as routes and records are inserted, updated and removed, so a read serves them with no scan and no timer. A live-route upsert is an UPDATE, then an INSERT only when no row matched, so its rows-affected count tells a new route from a re-announced one. Oldest/newest timestamps read one row from each end of the last_updated index, which also fixes the per-call scan warning. The distinct-prefix histogram stays a single indexed query. JSON shape of /api/v1/stats and the status page is unchanged. Model: opus-4-8 --- TODO.md | 4 + internal/database/counts.go | 146 +++++++++ internal/database/counts_test.go | 335 ++++++++++++++++++++ internal/database/database.go | 525 +++++++++++++++++++------------ internal/database/interface.go | 2 + internal/server/handlers.go | 22 +- 6 files changed, 812 insertions(+), 222 deletions(-) create mode 100644 internal/database/counts.go create mode 100644 internal/database/counts_test.go diff --git a/TODO.md b/TODO.md index e4df8b9..9e25bd4 100644 --- a/TODO.md +++ b/TODO.md @@ -23,6 +23,10 @@ runs make check on main. # Completed Steps +- 2026-09-22: realtime in-memory database statistics: counts seeded at + startup and adjusted on every write, oldest/newest route timestamps via + index-end lookups; `/api/v1/stats` no longer scans the tables (closes + #27) - 2026-07-07 Adopted scripts-to-rule-them-all: `script/` entrypoints, Makefile shims, README Entrypoints section - 2026-02-22: repo policy compliance: required policy files, .gitignore diff --git a/internal/database/counts.go b/internal/database/counts.go new file mode 100644 index 0000000..a305ba2 --- /dev/null +++ b/internal/database/counts.go @@ -0,0 +1,146 @@ +package database + +import ( + "context" + "fmt" + "sync" +) + +// liveCounts holds the running row counts that the stats endpoints report. They +// are seeded once at startup from the tables and then adjusted on every write, +// so a stats read serves them from memory instead of running a COUNT(*) over +// each table. Those scans, once the database passed a few GiB, took the whole +// request timeout and made /api/v1/stats return 500 (issue 27). +// +// A single mutex guards all fields so the stats reader takes a consistent +// snapshot at one instant and writers, which already run under the database +// write lock, adjust the counts after their transaction commits. +type liveCounts struct { + mu sync.RWMutex + asns int + prefixesV4 int + prefixesV6 int + peerings int + peers int + routesV4 int + routesV6 int +} + +// 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) { + c.mu.Lock() + defer c.mu.Unlock() + + c.asns = asns + c.prefixesV4 = prefixesV4 + c.prefixesV6 = prefixesV6 + c.peerings = peerings + c.peers = peers + c.routesV4 = routesV4 + c.routesV6 = routesV6 +} + +// addASNs adds n to the ASN count. +func (c *liveCounts) addASNs(n int) { + c.mu.Lock() + c.asns += n + c.mu.Unlock() +} + +// addPrefixes adds to the IPv4 and IPv6 prefix counts. +func (c *liveCounts) addPrefixes(v4, v6 int) { + c.mu.Lock() + c.prefixesV4 += v4 + c.prefixesV6 += v6 + c.mu.Unlock() +} + +// addPeerings adds n to the peering count. +func (c *liveCounts) addPeerings(n int) { + c.mu.Lock() + c.peerings += n + c.mu.Unlock() +} + +// addPeers adds n to the BGP peer count. +func (c *liveCounts) addPeers(n int) { + c.mu.Lock() + c.peers += n + c.mu.Unlock() +} + +// addRoutes adds to the IPv4 and IPv6 live-route counts. Deletions pass +// negative values. +func (c *liveCounts) addRoutes(v4, v6 int) { + c.mu.Lock() + c.routesV4 += v4 + c.routesV6 += v6 + c.mu.Unlock() +} + +// fill copies the counts into a Stats, including the derived totals, under a +// single read lock so the reader sees one consistent snapshot. +func (c *liveCounts) fill(s *Stats) { + c.mu.RLock() + defer c.mu.RUnlock() + + s.ASNs = c.asns + s.IPv4Prefixes = c.prefixesV4 + s.IPv6Prefixes = c.prefixesV6 + s.Prefixes = c.prefixesV4 + c.prefixesV6 + s.Peerings = c.peerings + s.Peers = c.peers + s.IPv4Routes = c.routesV4 + s.IPv6Routes = c.routesV6 + s.LiveRoutes = c.routesV4 + c.routesV6 +} + +// countRows returns the number of rows in the named table. It is used only at +// startup to seed the in-memory counters, so a full COUNT(*) is acceptable. +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 { + return 0, fmt.Errorf("failed to count %s: %w", table, err) + } + + return n, nil +} + +// seedCounts reads the current row counts from the tables into the in-memory +// counters. It runs once at startup, before the streamer begins writing. +func (d *Database) seedCounts(ctx context.Context) error { + asns, err := d.countRows(ctx, "asns") + 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") + if err != nil { + return err + } + routesV6, err := d.countRows(ctx, "live_routes_v6") + if err != nil { + return err + } + + d.counts.seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6) + + return nil +} diff --git a/internal/database/counts_test.go b/internal/database/counts_test.go new file mode 100644 index 0000000..6db2859 --- /dev/null +++ b/internal/database/counts_test.go @@ -0,0 +1,335 @@ +package database + +import ( + "context" + "sync" + "testing" + "time" + + "git.eeqj.de/sneak/routewatch/internal/config" + "git.eeqj.de/sneak/routewatch/internal/logger" + "github.com/google/uuid" +) + +// mkV4Route builds an IPv4 live route with its range columns populated. +func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute { + t.Helper() + + start, end, err := CalculateIPv4Range(prefix) + if err != nil { + t.Fatalf("CalculateIPv4Range(%s): %v", prefix, err) + } + + return &LiveRoute{ + ID: uuid.New(), + Prefix: prefix, + MaskLength: 24, + IPVersion: ipVersionV4, + OriginASN: asn, + PeerIP: "192.0.2.1", + ASPath: []int{asn}, + NextHop: "192.0.2.254", + LastUpdated: ts, + V4IPStart: &start, + V4IPEnd: &end, + } +} + +// mkV6Route builds an IPv6 live route. +func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute { + return &LiveRoute{ + ID: uuid.New(), + Prefix: prefix, + MaskLength: 32, + IPVersion: ipVersionV6, + OriginASN: asn, + PeerIP: "2001:db8::1", + ASPath: []int{asn}, + NextHop: "2001:db8::ffff", + LastUpdated: ts, + } +} + +// TestLiveCountsTrackWritesInRealtime checks that the stats counts start at +// zero, reflect each write the moment it commits (no recompute, no timer), do +// not move when a route is merely re-announced, and drop when a route is +// deleted. These counts are what /api/v1/stats reports; before this change the +// endpoint recomputed them with a COUNT(*) over each table on every request. +func TestLiveCountsTrackWritesInRealtime(t *testing.T) { + cfg := &config.Config{StateDir: t.TempDir()} + + db, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + ctx := context.Background() + + empty, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext on empty database: %v", err) + } + if empty.ASNs != 0 || empty.Prefixes != 0 || empty.Peerings != 0 || + empty.Peers != 0 || empty.LiveRoutes != 0 { + t.Fatalf("empty database counts nonzero: %+v", empty) + } + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + if err := db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts}); err != nil { + t.Fatalf("GetOrCreateASNBatch: %v", err) + } + if err := db.UpdatePrefixesBatch(map[string]time.Time{ + "198.51.100.0/24": ts, + "2001:db8::/32": ts, + }); err != nil { + t.Fatalf("UpdatePrefixesBatch: %v", err) + } + if err := db.UpdatePeerBatch(map[string]PeerUpdate{ + "192.0.2.1": {PeerIP: "192.0.2.1", PeerASN: 64500, MessageType: "UPDATE", Timestamp: ts}, + }); err != nil { + t.Fatalf("UpdatePeerBatch: %v", err) + } + if err := db.RecordPeering(64500, 64501, ts); err != nil { + t.Fatalf("RecordPeering: %v", err) + } + + routes := []*LiveRoute{ + mkV4Route(t, "198.51.100.0/24", 64500, ts), + mkV4Route(t, "203.0.113.0/24", 64501, ts.Add(time.Minute)), + mkV6Route("2001:db8::/32", 64502, ts.Add(2*time.Minute)), + } + if err := db.UpsertLiveRouteBatch(routes); err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + stats, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + assertCounts(t, "after inserts", stats, wantCounts{ + asns: 2, prefixes: 2, peerings: 1, peers: 1, + ipv4Routes: 2, ipv6Routes: 1, liveRoutes: 3, + }) + + // Re-announcing the same routes is an update, not an insert: counts hold. + if err := db.UpsertLiveRouteBatch(routes); err != nil { + t.Fatalf("UpsertLiveRouteBatch (re-announce): %v", err) + } + stats, err = db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + assertCounts(t, "after re-announce", stats, wantCounts{ + asns: 2, prefixes: 2, peerings: 1, peers: 1, + ipv4Routes: 2, ipv6Routes: 1, liveRoutes: 3, + }) + + // A withdrawal removes one route. + if err := db.DeleteLiveRouteBatch([]LiveRouteDeletion{ + {Prefix: "203.0.113.0/24", OriginASN: 64501, PeerIP: "192.0.2.1", IPVersion: ipVersionV4}, + }); err != nil { + t.Fatalf("DeleteLiveRouteBatch: %v", err) + } + stats, err = db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + assertCounts(t, "after delete", stats, wantCounts{ + asns: 2, prefixes: 2, peerings: 1, peers: 1, + ipv4Routes: 1, ipv6Routes: 1, liveRoutes: 2, + }) +} + +// TestLiveCountsSeededFromDatabaseAtStartup writes rows, reopens the same +// database file, and checks the counts come back from the seed scan rather than +// starting at zero. +func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) { + cfg := &config.Config{StateDir: t.TempDir()} + + db, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + if err := db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts, 64502: ts}); err != nil { + t.Fatalf("GetOrCreateASNBatch: %v", err) + } + if err := db.UpsertLiveRouteBatch([]*LiveRoute{ + mkV4Route(t, "198.51.100.0/24", 64500, ts), + mkV6Route("2001:db8::/32", 64502, ts), + }); err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + if err := db.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + reopened, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to reopen database: %v", err) + } + defer func() { _ = reopened.Close() }() + + stats, err := reopened.GetStatsContext(context.Background()) + if err != nil { + t.Fatalf("GetStatsContext after reopen: %v", err) + } + if stats.ASNs != 3 { + t.Errorf("seeded ASNs = %d, want 3", stats.ASNs) + } + if stats.IPv4Routes != 1 || stats.IPv6Routes != 1 || stats.LiveRoutes != 2 { + t.Errorf("seeded routes = (v4 %d, v6 %d, total %d), want (1, 1, 2)", + stats.IPv4Routes, stats.IPv6Routes, stats.LiveRoutes) + } +} + +// TestStatsRouteTimestamps checks the oldest/newest route timestamps are read +// from the right rows across both tables and parse into time.Time. The old +// MIN/MAX union query read its result into *time.Time, which the driver could +// not parse, so it logged a warning every call and left both timestamps nil. +func TestStatsRouteTimestamps(t *testing.T) { + cfg := &config.Config{StateDir: t.TempDir()} + + db, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + ctx := context.Background() + + empty, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext on empty database: %v", err) + } + if empty.OldestRoute != nil || empty.NewestRoute != nil { + t.Fatalf("empty database timestamps = (%v, %v), want (nil, nil)", + empty.OldestRoute, empty.NewestRoute) + } + + base := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + oldest := base + newest := base.Add(2 * time.Minute) + if err := db.UpsertLiveRouteBatch([]*LiveRoute{ + mkV4Route(t, "198.51.100.0/24", 64500, base.Add(time.Minute)), + mkV4Route(t, "203.0.113.0/24", 64501, oldest), + mkV6Route("2001:db8::/32", 64502, newest), + }); err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + stats, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + if stats.OldestRoute == nil || !stats.OldestRoute.Equal(oldest) { + t.Errorf("OldestRoute = %v, want %v", stats.OldestRoute, oldest) + } + if stats.NewestRoute == nil || !stats.NewestRoute.Equal(newest) { + t.Errorf("NewestRoute = %v, want %v", stats.NewestRoute, newest) + } +} + +// TestLiveCountsConcurrentReadWrite runs writers and stats readers at once so +// the race detector proves the counters are safe under concurrent use. +func TestLiveCountsConcurrentReadWrite(t *testing.T) { + cfg := &config.Config{StateDir: t.TempDir()} + + db, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + const writers = 4 + var wg sync.WaitGroup + + wg.Add(writers) + for w := range writers { + go func(base int) { + defer wg.Done() + for i := range 25 { + asn := 65000 + base*100 + i + route := mkV6Route("2001:db8::/32", asn, ts) + route.PeerIP = "2001:db8::" + uuid.NewString() + if err := db.UpsertLiveRoute(route); err != nil { + t.Errorf("UpsertLiveRoute: %v", err) + + return + } + } + }(w) + } + + var readerWG sync.WaitGroup + readerWG.Add(1) + stop := make(chan struct{}) + go func() { + defer readerWG.Done() + for { + select { + case <-stop: + return + default: + if _, err := db.GetStatsContext(context.Background()); err != nil { + t.Errorf("GetStatsContext: %v", err) + + return + } + } + } + }() + + wg.Wait() + close(stop) + readerWG.Wait() + + stats, err := db.GetStatsContext(context.Background()) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + if want := writers * 25; stats.IPv6Routes != want { + t.Errorf("IPv6Routes = %d, want %d", stats.IPv6Routes, want) + } +} + +type wantCounts struct { + asns int + prefixes int + peerings int + peers int + ipv4Routes int + ipv6Routes int + liveRoutes int +} + +func assertCounts(t *testing.T, when string, got Stats, want wantCounts) { + t.Helper() + + if got.ASNs != want.asns { + t.Errorf("%s: ASNs = %d, want %d", when, got.ASNs, want.asns) + } + if got.Prefixes != want.prefixes { + t.Errorf("%s: Prefixes = %d, want %d", when, got.Prefixes, want.prefixes) + } + if got.Peerings != want.peerings { + t.Errorf("%s: Peerings = %d, want %d", when, got.Peerings, want.peerings) + } + if got.Peers != want.peers { + t.Errorf("%s: Peers = %d, want %d", when, got.Peers, want.peers) + } + if got.IPv4Routes != want.ipv4Routes { + t.Errorf("%s: IPv4Routes = %d, want %d", when, got.IPv4Routes, want.ipv4Routes) + } + if got.IPv6Routes != want.ipv6Routes { + t.Errorf("%s: IPv6Routes = %d, want %d", when, got.IPv6Routes, want.ipv6Routes) + } + if got.LiveRoutes != want.liveRoutes { + t.Errorf("%s: LiveRoutes = %d, want %d", when, got.LiveRoutes, want.liveRoutes) + } +} diff --git a/internal/database/database.go b/internal/database/database.go index d7a7be4..4007b48 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -72,6 +72,7 @@ type Database struct { mu sync.Mutex lockedAt time.Time lockedBy string + counts *liveCounts } // New creates a new database connection and initializes the schema. @@ -115,12 +116,19 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { db.SetMaxIdleConns(maxConns) db.SetConnMaxLifetime(0) - database := &Database{db: db, logger: logger, path: dbPath} + database := &Database{db: db, logger: logger, path: dbPath, counts: &liveCounts{}} if err := database.Initialize(); 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 { + return nil, fmt.Errorf("failed to seed statistics counters: %w", err) + } + return database, nil } @@ -208,6 +216,81 @@ func (d *Database) beginTx() (*loggingTx, error) { 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(*). +// Callers hold the database write lock, so no other writer can insert the same +// key between the two statements. The id column is set only on INSERT, so an +// updated route keeps its original id, exactly as the previous ON CONFLICT +// upsert did. +const ( + updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?, + last_updated = ?, ip_start = ?, ip_end = ? 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, ip_start, ip_end) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + + 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 (?, ?, ?, ?, ?, ?, ?, ?)` +) + +// upsertRouteRowV4 updates an IPv4 live route, inserting it when no row matched, +// and reports whether a new row was inserted. +func upsertRouteRowV4(upd, ins *sql.Stmt, route *LiveRoute, pathJSON string) (inserted bool, err error) { + if route.V4IPStart == nil || route.V4IPEnd == nil { + return false, fmt.Errorf("IPv4 route %s missing range values", route.Prefix) + } + + res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated, + *route.V4IPStart, *route.V4IPEnd, route.Prefix, route.OriginASN, route.PeerIP) + if err != nil { + return false, err + } + affected, err := res.RowsAffected() + if err != nil { + return false, err + } + if affected > 0 { + return false, nil + } + + _, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN, + route.PeerIP, pathJSON, route.NextHop, route.LastUpdated, *route.V4IPStart, *route.V4IPEnd) + if err != nil { + return false, err + } + + return true, nil +} + +// upsertRouteRowV6 updates an IPv6 live route, inserting it when no row matched, +// and reports whether a new row was inserted. +func upsertRouteRowV6(upd, ins *sql.Stmt, route *LiveRoute, pathJSON string) (inserted 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, err + } + affected, err := res.RowsAffected() + if err != nil { + return false, err + } + if affected > 0 { + return false, nil + } + + _, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN, + route.PeerIP, pathJSON, route.NextHop, route.LastUpdated) + if err != nil { + return false, err + } + + return true, nil +} + // UpsertLiveRouteBatch inserts or updates multiple live routes in a single transaction func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error { if len(routes) == 0 { @@ -227,91 +310,65 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error { } }() - // Prepare statements for both IPv4 and IPv6 - queryV4 := ` - INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, - last_updated, ip_start, ip_end) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET - mask_length = excluded.mask_length, - as_path = excluded.as_path, - next_hop = excluded.next_hop, - last_updated = excluded.last_updated, - ip_start = excluded.ip_start, - ip_end = excluded.ip_end - ` - - queryV6 := ` - INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, - last_updated) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET - mask_length = excluded.mask_length, - as_path = excluded.as_path, - next_hop = excluded.next_hop, - last_updated = excluded.last_updated - ` - - stmtV4, err := tx.Prepare(queryV4) + // Prepare the update and insert statements for both tables. + updV4, err := tx.Prepare(updateLiveRouteV4SQL) if err != nil { - return fmt.Errorf("failed to prepare IPv4 statement: %w", err) + return fmt.Errorf("failed to prepare IPv4 update statement: %w", err) } - defer func() { _ = stmtV4.Close() }() + defer func() { _ = updV4.Close() }() - stmtV6, err := tx.Prepare(queryV6) + insV4, err := tx.Prepare(insertLiveRouteV4SQL) if err != nil { - return fmt.Errorf("failed to prepare IPv6 statement: %w", err) + return fmt.Errorf("failed to prepare IPv4 insert statement: %w", err) } - defer func() { _ = stmtV6.Close() }() + 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() }() + + var newV4, newV6 int for _, route := range routes { - // Encode AS path as JSON pathJSON, err := json.Marshal(route.ASPath) if err != nil { return fmt.Errorf("failed to encode AS path: %w", err) } - // Use appropriate statement based on IP version if route.IPVersion == ipVersionV4 { - // IPv4 routes must have range values - if route.V4IPStart == nil || route.V4IPEnd == nil { - return fmt.Errorf("IPv4 route %s missing range values", route.Prefix) + inserted, err := upsertRouteRowV4(updV4, insV4, route, string(pathJSON)) + if err != nil { + return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) + } + if inserted { + newV4++ } - _, err = stmtV4.Exec( - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - *route.V4IPStart, - *route.V4IPEnd, - ) - } else { - // IPv6 routes - _, err = stmtV6.Exec( - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - ) + continue } + + inserted, err := upsertRouteRowV6(updV6, insV6, route, string(pathJSON)) if err != nil { return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) } + if inserted { + newV6++ + } } if err = tx.Commit(); err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } + d.counts.addRoutes(newV4, newV6) + return nil } @@ -362,6 +419,7 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { defer func() { _ = stmtV6WithoutOrigin.Close() }() // Process deletions + var deletedV4, deletedV6 int64 for _, del := range deletions { var stmt *sql.Stmt @@ -382,21 +440,36 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { } // Execute deletion + var res sql.Result if del.OriginASN == 0 { - _, err = stmt.Exec(del.Prefix, del.PeerIP) + res, err = stmt.Exec(del.Prefix, del.PeerIP) } else { - _, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP) + 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 err = tx.Commit(); err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } + d.counts.addRoutes(-int(deletedV4), -int(deletedV6)) + return nil } @@ -456,6 +529,7 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { } defer func() { _ = insertV6Stmt.Close() }() + var newV4, newV6 int for prefix, timestamp := range prefixes { ipVersion := detectIPVersion(prefix) @@ -482,6 +556,11 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { 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) } @@ -491,6 +570,8 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { return fmt.Errorf("failed to commit transaction: %w", err) } + d.counts.addPrefixes(newV4, newV6) + return nil } @@ -534,6 +615,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error { } defer func() { _ = insertStmt.Close() }() + var newASNs int for number, timestamp := range asns { var asn ASN var handle, description sql.NullString @@ -568,6 +650,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error { if err != nil { return fmt.Errorf("failed to insert ASN %d: %w", number, err) } + newASNs++ continue } @@ -581,6 +664,8 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error { return fmt.Errorf("failed to commit transaction: %w", err) } + d.counts.addASNs(newASNs) + return nil } @@ -652,6 +737,8 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error) return nil, err } + d.counts.addASNs(1) + return &asn, nil } @@ -727,6 +814,12 @@ func (d *Database) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefi return nil, err } + if ipVersion == ipVersionV4 { + d.counts.addPrefixes(1, 0) + } else { + d.counts.addPrefixes(0, 1) + } + return &p, nil } @@ -803,6 +896,10 @@ func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error { return err } + if !exists { + d.counts.addPeerings(1) + } + return nil } @@ -846,6 +943,7 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error { } defer func() { _ = insertStmt.Close() }() + var newPeers int for _, update := range peers { var exists bool err = checkStmt.QueryRow(update.PeerIP).Scan(&exists) @@ -863,12 +961,17 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error { if err != nil { return fmt.Errorf("failed to update peer %s: %w", update.PeerIP, err) } + if !exists { + newPeers++ + } } if err = tx.Commit(); err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } + d.counts.addPeers(newPeers) + return nil } @@ -918,6 +1021,10 @@ func (d *Database) UpdatePeer(peerIP string, peerASN int, messageType string, ti return err } + if !exists { + d.counts.addPeers(1) + } + return nil } @@ -926,169 +1033,161 @@ func (d *Database) GetStats() (Stats, error) { return d.GetStatsContext(context.Background()) } -// GetStatsContext returns database statistics with context support +// GetStatsContext returns database statistics with context support. +// +// The row counts (ASNs, prefixes, peerings, peers, live routes) come from the +// in-memory counters, seeded at startup and kept current on every write, so a +// read runs no COUNT(*) over the tables. The oldest/newest route timestamps are +// read from the ends of the last_updated index, and the file size from a +// stat(); neither is a table scan. The only remaining query is the prefix +// distribution. func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) { var stats Stats - // Count ASNs - err := d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM asns").Scan(&stats.ASNs) - if err != nil { - return stats, err - } + // Row counts from memory, as a single consistent snapshot. + d.counts.fill(&stats) - // Count prefixes from both tables - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM prefixes_v4").Scan(&stats.IPv4Prefixes) - if err != nil { - return stats, err - } - - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM prefixes_v6").Scan(&stats.IPv6Prefixes) - if err != nil { - return stats, err - } - - stats.Prefixes = stats.IPv4Prefixes + stats.IPv6Prefixes - - // Count peerings - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM peerings").Scan(&stats.Peerings) - if err != nil { - return stats, err - } - - // Count peers - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM bgp_peers").Scan(&stats.Peers) - if err != nil { - return stats, err - } - - // Get database file size - fileInfo, err := os.Stat(d.path) - if err != nil { + // Database file size is a cheap stat() on the file. + if fileInfo, err := os.Stat(d.path); err != nil { d.logger.Warn("Failed to get database file size", "error", err) - stats.FileSizeBytes = 0 } else { stats.FileSizeBytes = fileInfo.Size() } - // Get live routes count from both tables - var v4Count, v6Count int - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v4").Scan(&v4Count) - if err != nil { - return stats, fmt.Errorf("failed to count IPv4 routes: %w", err) - } - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v6").Scan(&v6Count) - if err != nil { - return stats, fmt.Errorf("failed to count IPv6 routes: %w", err) - } - stats.LiveRoutes = v4Count + v6Count - - // Get oldest and newest route timestamps - routeTimestampQuery := ` - SELECT MIN(last_updated), MAX(last_updated) FROM ( - SELECT last_updated FROM live_routes_v4 - UNION ALL - SELECT last_updated FROM live_routes_v6 - ) - ` - var oldestRoute, newestRoute *time.Time - err = d.db.QueryRowContext(ctx, routeTimestampQuery).Scan(&oldestRoute, &newestRoute) + // Oldest and newest route timestamps read one row from each end of the + // last_updated index (a log-time lookup, not a scan). Selecting the column + // directly lets the driver parse the DATETIME into time.Time; the old + // MIN/MAX union scan returned an untyped string that failed to scan and + // logged a warning on every call. + oldest, newest, err := d.routeTimestampRange(ctx) if err != nil { + // Display-only fields; log but keep the rest of the stats. d.logger.Warn("Failed to get route timestamps", "error", err) } else { - stats.OldestRoute = oldestRoute - stats.NewestRoute = newestRoute + stats.OldestRoute = oldest + stats.NewestRoute = newest } - // Get prefix distribution + // Prefix distribution counts distinct prefixes per mask length. It stays a + // query over the covering (mask_length, prefix) index rather than an + // in-memory counter: maintaining distinct-prefix-per-mask in memory would + // need a per-prefix table of roughly a million entries, memory this service + // is tuned to avoid. stats.IPv4PrefixDistribution, stats.IPv6PrefixDistribution, err = d.GetPrefixDistributionContext(ctx) if err != nil { - // Log but don't fail + // Log but don't fail. d.logger.Warn("Failed to get prefix distribution", "error", err) } 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") defer d.unlock() - // Choose table based on IP version - tableName := "live_routes_v4" - if route.IPVersion == ipVersionV6 { - tableName = "live_routes_v6" - } - - var query string - if route.IPVersion == ipVersionV4 { - query = fmt.Sprintf(` - INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, - last_updated, ip_start, ip_end) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET - mask_length = excluded.mask_length, - as_path = excluded.as_path, - next_hop = excluded.next_hop, - last_updated = excluded.last_updated, - ip_start = excluded.ip_start, - ip_end = excluded.ip_end - `, tableName) - } else { - query = fmt.Sprintf(` - INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, - last_updated) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET - mask_length = excluded.mask_length, - as_path = excluded.as_path, - next_hop = excluded.next_hop, - last_updated = excluded.last_updated - `, tableName) - } - - // Encode AS path as JSON pathJSON, err := json.Marshal(route.ASPath) if err != nil { return fmt.Errorf("failed to encode AS path: %w", err) } - if route.IPVersion == ipVersionV4 { - // Convert v4_ip_start and v4_ip_end to interface{} for SQL NULL handling - var v4Start, v4End interface{} - if route.V4IPStart != nil { - v4Start = *route.V4IPStart - } - if route.V4IPEnd != nil { - v4End = *route.V4IPEnd - } - - _, err = d.db.Exec(query, - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - v4Start, - v4End, - ) - } else { - _, err = d.db.Exec(query, - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - ) + updateSQL, insertSQL := updateLiveRouteV4SQL, insertLiveRouteV4SQL + if route.IPVersion == ipVersionV6 { + updateSQL, insertSQL = updateLiveRouteV6SQL, insertLiveRouteV6SQL } - return err + // 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) + if err != nil { + return fmt.Errorf("failed to prepare update statement: %w", err) + } + defer func() { _ = upd.Close() }() + + ins, err := d.db.Prepare(insertSQL) + if err != nil { + return fmt.Errorf("failed to prepare insert statement: %w", err) + } + defer func() { _ = ins.Close() }() + + var inserted bool + if route.IPVersion == ipVersionV4 { + inserted, err = upsertRouteRowV4(upd, ins, route, string(pathJSON)) + } else { + inserted, err = upsertRouteRowV6(upd, ins, route, string(pathJSON)) + } + if err != nil { + return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) + } + + if inserted { + if route.IPVersion == ipVersionV4 { + d.counts.addRoutes(1, 0) + } else { + d.counts.addRoutes(0, 1) + } + } + + return nil } // DeleteLiveRoute deletes a live route @@ -1103,23 +1202,41 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) return fmt.Errorf("invalid prefix format: %w", err) } - tableName := "live_routes_v4" - if ipnet.IP.To4() == nil { - tableName = "live_routes_v6" + isV4 := ipnet.IP.To4() != nil + + // 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 = d.db.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP) + case isV4: + res, err = d.db.Exec( + `DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, + prefix, originASN, peerIP) + case originASN == 0: + res, err = d.db.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP) + default: + res, err = d.db.Exec( + `DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, + prefix, originASN, peerIP) + } + if err != nil { + return err } - var query string - if originASN == 0 { - // Delete all routes for this prefix from this peer - query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND peer_ip = ?`, tableName) - _, err = d.db.Exec(query, prefix, peerIP) + affected, err := res.RowsAffected() + if err != nil { + return err + } + if isV4 { + d.counts.addRoutes(-int(affected), 0) } else { - // Delete specific route - query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, tableName) - _, err = d.db.Exec(query, prefix, originASN, peerIP) + d.counts.addRoutes(0, -int(affected)) } - return err + return nil } // GetPrefixDistribution returns the distribution of unique prefixes by mask length diff --git a/internal/database/interface.go b/internal/database/interface.go index 7f3f8e9..038dfd5 100644 --- a/internal/database/interface.go +++ b/internal/database/interface.go @@ -18,6 +18,8 @@ type Stats struct { Peers int FileSizeBytes int64 LiveRoutes int + IPv4Routes int + IPv6Routes int OldestRoute *time.Time NewestRoute *time.Time IPv4PrefixDistribution []PrefixDistribution diff --git a/internal/server/handlers.go b/internal/server/handlers.go index d04d0fd..195cb33 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -219,13 +219,6 @@ func (s *Server) handleStatusJSON() http.HandlerFunc { const bitsPerMegabit = 1000000.0 - // Get route counts from database - ipv4Routes, ipv6Routes, err := s.db.GetLiveRouteCountsContext(ctx) - if err != nil { - s.logger.Warn("Failed to get live route counts", "error", err) - // Continue with zero counts - } - // Get route update metrics routeMetrics := s.streamer.GetMetricsTracker().GetRouteMetrics() @@ -259,8 +252,8 @@ func (s *Server) handleStatusJSON() http.HandlerFunc { Peers: dbStats.Peers, DatabaseSizeBytes: dbStats.FileSizeBytes, LiveRoutes: dbStats.LiveRoutes, - IPv4Routes: ipv4Routes, - IPv6Routes: ipv6Routes, + IPv4Routes: dbStats.IPv4Routes, + IPv6Routes: dbStats.IPv6Routes, OldestRoute: dbStats.OldestRoute, NewestRoute: dbStats.NewestRoute, IPv4UpdatesPerSec: routeMetrics.IPv4UpdatesPerSec, @@ -439,13 +432,6 @@ func (s *Server) handleStats() http.HandlerFunc { const bitsPerMegabit = 1000000.0 - // Get route counts from database - ipv4Routes, ipv6Routes, err := s.db.GetLiveRouteCountsContext(ctx) - if err != nil { - s.logger.Warn("Failed to get live route counts", "error", err) - // Continue with zero counts - } - // Get route update metrics routeMetrics := s.streamer.GetMetricsTracker().GetRouteMetrics() @@ -537,8 +523,8 @@ func (s *Server) handleStats() http.HandlerFunc { Peers: dbStats.Peers, DatabaseSizeBytes: dbStats.FileSizeBytes, LiveRoutes: dbStats.LiveRoutes, - IPv4Routes: ipv4Routes, - IPv6Routes: ipv6Routes, + IPv4Routes: dbStats.IPv4Routes, + IPv6Routes: dbStats.IPv6Routes, OldestRoute: dbStats.OldestRoute, NewestRoute: dbStats.NewestRoute, IPv4UpdatesPerSec: routeMetrics.IPv4UpdatesPerSec,