diff --git a/internal/database/counts_test.go b/internal/database/counts_internal_test.go similarity index 80% rename from internal/database/counts_test.go rename to internal/database/counts_internal_test.go index 8922619..c3f0a0e 100644 --- a/internal/database/counts_test.go +++ b/internal/database/counts_internal_test.go @@ -12,6 +12,12 @@ import ( "github.com/google/uuid" ) +// testV4PeerIP is the peer the IPv4 routes in these tests are learned from. +const testV4PeerIP = "192.0.2.1" + +// testV6Prefix is the prefix of every IPv6 route in these tests. +const testV6Prefix = "2001:db8::/32" + // mkV4Route builds an IPv4 live route with its mask length taken from the // prefix. func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute { @@ -28,18 +34,18 @@ func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute { MaskLength: maskLength, IPVersion: ipVersionV4, OriginASN: asn, - PeerIP: "192.0.2.1", + PeerIP: testV4PeerIP, ASPath: []int{asn}, NextHop: "192.0.2.254", LastUpdated: ts, } } -// mkV6Route builds an IPv6 live route. -func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute { +// mkV6Route builds an IPv6 live route for testV6Prefix. +func mkV6Route(asn int, ts time.Time) *LiveRoute { return &LiveRoute{ ID: uuid.New(), - Prefix: prefix, + Prefix: testV6Prefix, MaskLength: 32, IPVersion: ipVersionV6, OriginASN: asn, @@ -56,6 +62,8 @@ func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute { // 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) { + t.Parallel() + cfg := &config.Config{StateDir: t.TempDir()} db, err := New(cfg, logger.New()) @@ -70,6 +78,7 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) { 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) @@ -77,30 +86,16 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) { 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) - } + writeASNsPrefixesAndPeers(t, db, ts) 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)), + mkV6Route(64502, ts.Add(2*time.Minute)), } - if err := db.UpsertLiveRouteBatch(routes); err != nil { + + err = db.UpsertLiveRouteBatch(routes) + if err != nil { t.Fatalf("UpsertLiveRouteBatch: %v", err) } @@ -108,44 +103,88 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) { 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 { + err = db.UpsertLiveRouteBatch(routes) + if 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 { + err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{{ + Prefix: "203.0.113.0/24", OriginASN: 64501, + PeerIP: testV4PeerIP, IPVersion: ipVersionV4, + }}) + if 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, }) } +// writeASNsPrefixesAndPeers writes two ASNs, two prefixes, one peer and one +// peering: the rows besides live routes that the stats count. +func writeASNsPrefixesAndPeers(t *testing.T, db *Database, ts time.Time) { + t.Helper() + + err := db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts}) + if err != nil { + t.Fatalf("GetOrCreateASNBatch: %v", err) + } + + err = db.UpdatePrefixesBatch(map[string]time.Time{ + "198.51.100.0/24": ts, + testV6Prefix: ts, + }) + if err != nil { + t.Fatalf("UpdatePrefixesBatch: %v", err) + } + + err = db.UpdatePeerBatch(map[string]PeerUpdate{ + testV4PeerIP: { + PeerIP: testV4PeerIP, PeerASN: 64500, + MessageType: "UPDATE", Timestamp: ts, + }, + }) + if err != nil { + t.Fatalf("UpdatePeerBatch: %v", err) + } + + err = db.RecordPeering(64500, 64501, ts) + if err != nil { + t.Fatalf("RecordPeering: %v", err) + } +} + // TestLiveCountsSeededFromDatabaseAtStartup writes rows, reopens the same // database file, and checks the counts and the prefix distribution come back // from the seed scan rather than starting at zero. func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) { + t.Parallel() + cfg := &config.Config{StateDir: t.TempDir()} db, err := New(cfg, logger.New()) @@ -154,16 +193,22 @@ func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) { } 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 { + + err = db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts, 64502: ts}) + if err != nil { t.Fatalf("GetOrCreateASNBatch: %v", err) } - if err := db.UpsertLiveRouteBatch([]*LiveRoute{ + + err = db.UpsertLiveRouteBatch([]*LiveRoute{ mkV4Route(t, "198.51.100.0/24", 64500, ts), - mkV6Route("2001:db8::/32", 64502, ts), - }); err != nil { + mkV6Route(64502, ts), + }) + if err != nil { t.Fatalf("UpsertLiveRouteBatch: %v", err) } - if err := db.Close(); err != nil { + + err = db.Close() + if err != nil { t.Fatalf("Close: %v", err) } @@ -177,13 +222,16 @@ func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) { 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) } + assertDistribution(t, "seeded IPv4 distribution", stats.IPv4PrefixDistribution, []PrefixDistribution{{MaskLength: 24, Count: 1}}) assertDistribution(t, "seeded IPv6 distribution", stats.IPv6PrefixDistribution, @@ -196,26 +244,91 @@ func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) { // tables. The steps run once through the batch methods the prefix handler uses // and once through the single-route methods. func TestPrefixDistributionTracksWrites(t *testing.T) { + t.Parallel() + + steps := prefixDistributionSteps(t) + + for _, batch := range []bool{true, false} { + name := "single-route writes" + if batch { + name = "batch writes" + } + + t.Run(name, func(t *testing.T) { + t.Parallel() + + db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + ctx := context.Background() + + for _, step := range steps { + err := announceRoutes(db, batch, step.announce) + if err != nil { + t.Fatalf("%s: announce: %v", step.name, err) + } + + err = withdrawRoutes(db, batch, step.withdraw) + if err != nil { + t.Fatalf("%s: withdraw: %v", step.name, err) + } + + stats, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("%s: GetStatsContext: %v", step.name, err) + } + + queryV4, queryV6, err := db.GetPrefixDistributionContext(ctx) + if err != nil { + t.Fatalf("%s: GetPrefixDistributionContext: %v", step.name, err) + } + + assertDistribution(t, step.name+": IPv4 distribution", + stats.IPv4PrefixDistribution, step.wantV4) + assertDistribution(t, step.name+": IPv6 distribution", + stats.IPv6PrefixDistribution, step.wantV6) + assertDistribution(t, step.name+": IPv4 distribution query", + queryV4, step.wantV4) + assertDistribution(t, step.name+": IPv6 distribution query", + queryV6, step.wantV6) + } + }) + } +} + +// prefixDistributionStep is one step of TestPrefixDistributionTracksWrites: +// the routes it announces, then the routes it withdraws, then the +// distributions both the stats and the distribution query must report. +type prefixDistributionStep struct { + name string + announce []*LiveRoute + withdraw []*LiveRoute + wantV4 []PrefixDistribution + wantV6 []PrefixDistribution +} + +// prefixDistributionSteps returns the steps of +// TestPrefixDistributionTracksWrites, in the order they run. +func prefixDistributionSteps(t *testing.T) []prefixDistributionStep { + t.Helper() + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) shared := mkV4Route(t, "198.51.100.0/24", 64500, ts) sharedSecondPeer := mkV4Route(t, "198.51.100.0/24", 64500, ts) sharedSecondPeer.PeerIP = "192.0.2.2" other := mkV4Route(t, "203.0.113.0/24", 64501, ts) wide := mkV4Route(t, "172.16.0.0/16", 64502, ts) - v6 := mkV6Route("2001:db8::/32", 64503, ts) - v6SecondPeer := mkV6Route("2001:db8::/32", 64503, ts) + v6 := mkV6Route(64503, ts) + v6SecondPeer := mkV6Route(64503, ts) v6SecondPeer.PeerIP = "2001:db8::2" all := []PrefixDistribution{{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 2}} v6Only := []PrefixDistribution{{MaskLength: 32, Count: 1}} - steps := []struct { - name string - announce []*LiveRoute - withdraw []*LiveRoute - wantV4 []PrefixDistribution - wantV6 []PrefixDistribution - }{ + return []prefixDistributionStep{ { name: "new routes", announce: []*LiveRoute{shared, other, wide, v6}, @@ -239,7 +352,9 @@ func TestPrefixDistributionTracksWrites(t *testing.T) { { name: "withdrawal of the last route for a prefix", withdraw: []*LiveRoute{sharedSecondPeer, v6SecondPeer}, - wantV4: []PrefixDistribution{{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 1}}, + wantV4: []PrefixDistribution{ + {MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 1}, + }, }, { name: "withdrawal of every remaining route", @@ -270,45 +385,6 @@ func TestPrefixDistributionTracksWrites(t *testing.T) { wantV6: v6Only, }, } - - for _, batch := range []bool{true, false} { - name := "single-route writes" - if batch { - name = "batch writes" - } - - t.Run(name, func(t *testing.T) { - db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New()) - if err != nil { - t.Fatalf("failed to create database: %v", err) - } - defer func() { _ = db.Close() }() - - ctx := context.Background() - for _, step := range steps { - if err := announceRoutes(db, batch, step.announce); err != nil { - t.Fatalf("%s: announce: %v", step.name, err) - } - if err := withdrawRoutes(db, batch, step.withdraw); err != nil { - t.Fatalf("%s: withdraw: %v", step.name, err) - } - - stats, err := db.GetStatsContext(ctx) - if err != nil { - t.Fatalf("%s: GetStatsContext: %v", step.name, err) - } - queryV4, queryV6, err := db.GetPrefixDistributionContext(ctx) - if err != nil { - t.Fatalf("%s: GetPrefixDistributionContext: %v", step.name, err) - } - - assertDistribution(t, step.name+": IPv4 distribution", stats.IPv4PrefixDistribution, step.wantV4) - assertDistribution(t, step.name+": IPv6 distribution", stats.IPv6PrefixDistribution, step.wantV6) - assertDistribution(t, step.name+": IPv4 distribution query", queryV4, step.wantV4) - assertDistribution(t, step.name+": IPv6 distribution query", queryV6, step.wantV6) - } - }) - } } // announceRoutes writes routes in one UpsertLiveRouteBatch, or with one @@ -317,8 +393,10 @@ func announceRoutes(db *Database, batch bool, routes []*LiveRoute) error { if batch { return db.UpsertLiveRouteBatch(routes) } + for _, route := range routes { - if err := db.UpsertLiveRoute(route); err != nil { + err := db.UpsertLiveRoute(route) + if err != nil { return err } } @@ -333,7 +411,8 @@ func announceRoutes(db *Database, batch bool, routes []*LiveRoute) error { func withdrawRoutes(db *Database, batch bool, routes []*LiveRoute) error { if !batch { for _, route := range routes { - if err := db.DeleteLiveRoute(route.Prefix, 0, route.PeerIP); err != nil { + err := db.DeleteLiveRoute(route.Prefix, 0, route.PeerIP) + if err != nil { return err } } @@ -356,6 +435,8 @@ func withdrawRoutes(db *Database, batch bool, routes []*LiveRoute) error { // 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) { + t.Parallel() + cfg := &config.Config{StateDir: t.TempDir()} db, err := New(cfg, logger.New()) @@ -370,6 +451,7 @@ func TestStatsRouteTimestamps(t *testing.T) { 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) @@ -378,11 +460,13 @@ func TestStatsRouteTimestamps(t *testing.T) { 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{ + + 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 { + mkV6Route(64502, newest), + }) + if err != nil { t.Fatalf("UpsertLiveRouteBatch: %v", err) } @@ -390,9 +474,11 @@ func TestStatsRouteTimestamps(t *testing.T) { 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) } @@ -401,6 +487,8 @@ func TestStatsRouteTimestamps(t *testing.T) { // 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) { + t.Parallel() + cfg := &config.Config{StateDir: t.TempDir()} db, err := New(cfg, logger.New()) @@ -412,17 +500,21 @@ func TestLiveCountsConcurrentReadWrite(t *testing.T) { ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) const writers = 4 - var wg sync.WaitGroup + 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 := mkV6Route(asn, ts) route.PeerIP = "2001:db8::" + uuid.NewString() - if err := db.UpsertLiveRoute(route); err != nil { + + err := db.UpsertLiveRoute(route) + if err != nil { t.Errorf("UpsertLiveRoute: %v", err) return @@ -433,15 +525,19 @@ func TestLiveCountsConcurrentReadWrite(t *testing.T) { 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 { + _, err := db.GetStatsContext(context.Background()) + if err != nil { t.Errorf("GetStatsContext: %v", err) return @@ -458,6 +554,7 @@ func TestLiveCountsConcurrentReadWrite(t *testing.T) { if err != nil { t.Fatalf("GetStatsContext: %v", err) } + if want := writers * 25; stats.IPv6Routes != want { t.Errorf("IPv6Routes = %d, want %d", stats.IPv6Routes, want) } @@ -482,21 +579,27 @@ func assertCounts(t *testing.T, when string, got Stats, want wantCounts) { 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_test.go b/internal/database/database_internal_test.go similarity index 79% rename from internal/database/database_test.go rename to internal/database/database_internal_test.go index 82753a9..e0d73c6 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_internal_test.go @@ -50,6 +50,8 @@ const ( // address, ErrNoRoute when none covers it, and the next less specific prefix // once the only route of the most specific one is withdrawn. func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) { + t.Parallel() + cfg := &config.Config{StateDir: t.TempDir()} db, err := New(cfg, logger.New()) @@ -67,24 +69,7 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) { "2001:db8:1::/48": 64501, "2001:db8:1:2::/64": 64502, } - ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) - routes := make([]*LiveRoute, 0, len(origins)) - for prefix, asn := range origins { - routes = append(routes, &LiveRoute{ - ID: uuid.New(), - Prefix: prefix, - MaskLength: netip.MustParsePrefix(prefix).Bits(), - IPVersion: detectIPVersion(prefix), - OriginASN: asn, - PeerIP: testPeerIP, - ASPath: []int{asn}, - NextHop: testPeerIP, - LastUpdated: ts, - }) - } - if err := db.UpsertLiveRouteBatch(routes); err != nil { - t.Fatalf("UpsertLiveRouteBatch: %v", err) - } + writeOriginRoutes(t, db, origins) // lookup checks that ip resolves to the live prefix want, or to ErrNoRoute // when want is empty. @@ -99,11 +84,13 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) { return } + if err != nil { t.Errorf("GetIPInfo(%s): %v", ip, err) return } + if info.Netblock != want || info.MaskLength != netip.MustParsePrefix(want).Bits() || info.ASN != origins[want] { t.Errorf("GetIPInfo(%s) = %s (mask %d) AS%d, want %s AS%d", @@ -122,8 +109,14 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) { lookup("2001:db9::1", "") err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{ - {Prefix: "10.1.2.0/24", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV4}, - {Prefix: "2001:db8:1:2::/64", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV6}, + { + Prefix: "10.1.2.0/24", OriginASN: 64502, + PeerIP: testPeerIP, IPVersion: ipVersionV4, + }, + { + Prefix: "2001:db8:1:2::/64", OriginASN: 64502, + PeerIP: testPeerIP, IPVersion: ipVersionV6, + }, }) if err != nil { t.Fatalf("DeleteLiveRouteBatch: %v", err) @@ -133,11 +126,41 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) { lookup("2001:db8:1:2::3", "2001:db8:1::/48") } +// writeOriginRoutes writes one live route from testPeerIP for each prefix in +// origins, originated by the AS it maps to. +func writeOriginRoutes(t *testing.T, db *Database, origins map[string]int) { + t.Helper() + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + routes := make([]*LiveRoute, 0, len(origins)) + + for prefix, asn := range origins { + routes = append(routes, &LiveRoute{ + ID: uuid.New(), + Prefix: prefix, + MaskLength: netip.MustParsePrefix(prefix).Bits(), + IPVersion: detectIPVersion(prefix), + OriginASN: asn, + PeerIP: testPeerIP, + ASPath: []int{asn}, + NextHop: testPeerIP, + LastUpdated: ts, + }) + } + + err := db.UpsertLiveRouteBatch(routes) + if err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } +} + // TestConnectionPoolPragmas holds several pooled connections open at once and // checks each one carries the per-connection settings from the DSN, plus the // process-wide hard heap limit, and sees the new file with auto_vacuum // incremental. func TestConnectionPoolPragmas(t *testing.T) { + t.Parallel() + cfg := &config.Config{StateDir: t.TempDir()} db, err := New(cfg, logger.New()) @@ -157,57 +180,64 @@ func TestConnectionPoolPragmas(t *testing.T) { } }() - for i := 0; i < heldConnections; i++ { + for i := range heldConnections { c, err := db.db.Conn(ctx) if err != nil { t.Fatalf("failed to open connection %d: %v", i, err) } + conns = append(conns, c) } for i, c := range conns { - var cacheSize int - if err := c.QueryRowContext(ctx, "PRAGMA cache_size").Scan(&cacheSize); err != nil { - t.Fatalf("conn %d: failed to read cache_size: %v", i, err) - } + cacheSize := readPragma(ctx, t, c, i, "cache_size") if cacheSize != sqliteCacheSizeKiB { - t.Errorf("conn %d: cache_size = %d, want %d", i, cacheSize, sqliteCacheSizeKiB) + t.Errorf("conn %d: cache_size = %d, want %d", + i, cacheSize, sqliteCacheSizeKiB) } - var busyTimeout int - if err := c.QueryRowContext(ctx, "PRAGMA busy_timeout").Scan(&busyTimeout); err != nil { - t.Fatalf("conn %d: failed to read busy_timeout: %v", i, err) - } + busyTimeout := readPragma(ctx, t, c, i, "busy_timeout") if busyTimeout != sqliteBusyTimeoutMs { - t.Errorf("conn %d: busy_timeout = %d, want %d", i, busyTimeout, sqliteBusyTimeoutMs) + t.Errorf("conn %d: busy_timeout = %d, want %d", + i, busyTimeout, sqliteBusyTimeoutMs) } - var tempStore int - if err := c.QueryRowContext(ctx, "PRAGMA temp_store").Scan(&tempStore); err != nil { - t.Fatalf("conn %d: failed to read temp_store: %v", i, err) - } + tempStore := readPragma(ctx, t, c, i, "temp_store") if tempStore == tempStoreMemory { - t.Errorf("conn %d: temp_store = %d, want anything but %d (MEMORY)", i, tempStore, tempStoreMemory) + t.Errorf("conn %d: temp_store = %d, want anything but %d (MEMORY)", + i, tempStore, tempStoreMemory) } - var hardHeapLimit int64 - if err := c.QueryRowContext(ctx, "PRAGMA hard_heap_limit").Scan(&hardHeapLimit); err != nil { - t.Fatalf("conn %d: failed to read hard_heap_limit: %v", i, err) - } + hardHeapLimit := readPragma(ctx, t, c, i, "hard_heap_limit") if hardHeapLimit != sqliteHardHeapLimitBytes { - t.Errorf("conn %d: hard_heap_limit = %d, want %d", i, hardHeapLimit, sqliteHardHeapLimitBytes) + t.Errorf("conn %d: hard_heap_limit = %d, want %d", + i, hardHeapLimit, sqliteHardHeapLimitBytes) } - var autoVacuum int - if err := c.QueryRowContext(ctx, "PRAGMA auto_vacuum").Scan(&autoVacuum); err != nil { - t.Fatalf("conn %d: failed to read auto_vacuum: %v", i, err) - } + autoVacuum := readPragma(ctx, t, c, i, "auto_vacuum") if autoVacuum != autoVacuumIncremental { - t.Errorf("conn %d: auto_vacuum = %d, want %d (incremental)", i, autoVacuum, autoVacuumIncremental) + t.Errorf("conn %d: auto_vacuum = %d, want %d (incremental)", + i, autoVacuum, autoVacuumIncremental) } } } +// readPragma reads the integer value of PRAGMA name on connection i, c. +func readPragma( + ctx context.Context, t *testing.T, c *sql.Conn, i int, name string, +) int64 { + t.Helper() + + var value int64 + + err := c.QueryRowContext(ctx, "PRAGMA "+name).Scan(&value) + if err != nil { + t.Fatalf("conn %d: failed to read %s: %v", i, name, err) + } + + return value +} + // TestVacuumReturnsFreePages checks that after routes are deleted from a new // database, one Vacuum call returns every page they used. The deletes leave // fewer free pages than the 1000 Vacuum frees per call, so none may remain. @@ -215,6 +245,8 @@ func TestConnectionPoolPragmas(t *testing.T) { // the free pages stayed in the file, and with the PRAGMA run by ExecContext // Vacuum freed only one page per call. func TestVacuumReturnsFreePages(t *testing.T) { + t.Parallel() + cfg := &config.Config{StateDir: t.TempDir()} db, err := New(cfg, logger.New()) @@ -225,10 +257,12 @@ func TestVacuumReturnsFreePages(t *testing.T) { ctx := context.Background() ts := time.Now().UTC() + const asn = 64500 routes := make([]*LiveRoute, 0, vacuumTestRoutes) deletions := make([]LiveRouteDeletion, 0, vacuumTestRoutes) + for i := range vacuumTestRoutes { route := mkV4Route(t, fmt.Sprintf("10.%d.%d.0/24", i/256, i%256), asn, ts) routes = append(routes, route) @@ -240,29 +274,39 @@ func TestVacuumReturnsFreePages(t *testing.T) { }) } - if err := db.UpsertLiveRouteBatch(routes); err != nil { + err = db.UpsertLiveRouteBatch(routes) + if err != nil { t.Fatalf("failed to write routes: %v", err) } - if err := db.DeleteLiveRouteBatch(deletions); err != nil { + + err = db.DeleteLiveRouteBatch(deletions) + if err != nil { t.Fatalf("failed to delete routes: %v", err) } var before int - if err := db.db.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&before); err != nil { + + err = db.db.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&before) + if err != nil { t.Fatalf("failed to read freelist_count: %v", err) } + if before == 0 { t.Fatalf("no free pages after deleting %d routes", vacuumTestRoutes) } - if err := db.Vacuum(ctx); err != nil { + err = db.Vacuum(ctx) + if err != nil { t.Fatalf("Vacuum failed: %v", err) } var after int - if err := db.db.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&after); err != nil { + + err = db.db.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&after) + if err != nil { t.Fatalf("failed to read freelist_count: %v", err) } + if after != 0 { t.Errorf("free pages after Vacuum = %d of %d, want 0", after, before) } @@ -277,6 +321,8 @@ func TestVacuumReturnsFreePages(t *testing.T) { // no batch is dropped. The checkpoint runs without the Database mutex, exactly // as the background maintainer does in production. func TestBatchWriteDuringCheckpoint(t *testing.T) { + t.Parallel() + cfg := &config.Config{StateDir: t.TempDir()} db, err := New(cfg, logger.New()) @@ -289,8 +335,10 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) { var wg sync.WaitGroup wg.Add(1) + go func() { defer wg.Done() + for { select { case <-ctx.Done(): @@ -302,12 +350,14 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) { }() ts := time.Now().UTC() - for i := 0; i < contentionIterations; i++ { + for i := range contentionIterations { asns := map[int]time.Time{ i % contendedASNCount: ts, (i % contendedASNCount) + asnSecondBand: ts, } - if err := db.GetOrCreateASNBatch(asns); err != nil { + + err := db.GetOrCreateASNBatch(asns) + if err != nil { cancel() wg.Wait() t.Fatalf("batch write failed under checkpoint contention: %v", err) diff --git a/internal/database/models.go b/internal/database/models.go index a8e98b0..ed2274c 100644 --- a/internal/database/models.go +++ b/internal/database/models.go @@ -1,4 +1,3 @@ -// Package database provides SQLite storage for BGP routing data. package database import ( @@ -9,6 +8,8 @@ import ( // ASN represents an Autonomous System Number with its metadata including // handle, description, WHOIS data, and first/last seen timestamps. +// +//nolint:tagliatelle // field names are the existing JSON API type ASN struct { ASN int `json:"asn"` Handle string `json:"handle"` @@ -36,37 +37,39 @@ type ASN struct { // Prefix represents an IP prefix (CIDR block) with its IP version (4 or 6) // and first/last seen timestamps. type Prefix struct { - ID uuid.UUID `json:"id"` - Prefix string `json:"prefix"` - IPVersion int `json:"ip_version"` // 4 or 6 - FirstSeen time.Time `json:"first_seen"` - LastSeen time.Time `json:"last_seen"` + ID uuid.UUID + Prefix string + IPVersion int // 4 or 6 + FirstSeen time.Time + LastSeen time.Time } // Announcement represents a BGP announcement or withdrawal event, // containing the prefix, AS path, origin ASN, peer ASN, next hop, and timestamp. type Announcement struct { - ID uuid.UUID `json:"id"` - PrefixID uuid.UUID `json:"prefix_id"` - PeerASN int `json:"peer_asn"` - OriginASN int `json:"origin_asn"` - Path string `json:"path"` // JSON-encoded AS path - NextHop string `json:"next_hop"` - Timestamp time.Time `json:"timestamp"` - IsWithdrawal bool `json:"is_withdrawal"` + ID uuid.UUID + PrefixID uuid.UUID + PeerASN int + OriginASN int + Path string // JSON-encoded AS path + NextHop string + Timestamp time.Time + IsWithdrawal bool } // ASNPeering represents a peering relationship between two ASNs, // stored with the lower ASN as ASA and the higher as ASB. type ASNPeering struct { - ID uuid.UUID `json:"id"` - ASA int `json:"as_a"` - ASB int `json:"as_b"` - FirstSeen time.Time `json:"first_seen"` - LastSeen time.Time `json:"last_seen"` + ID uuid.UUID + ASA int + ASB int + FirstSeen time.Time + LastSeen time.Time } // LiveRoute represents a route in the live routing table +// +//nolint:tagliatelle // field names are the existing JSON API type LiveRoute struct { ID uuid.UUID `json:"id"` Prefix string `json:"prefix"` @@ -80,12 +83,16 @@ type LiveRoute struct { } // PrefixDistribution represents the distribution of prefixes by mask length +// +//nolint:tagliatelle // field names are the existing JSON API type PrefixDistribution struct { MaskLength int `json:"mask_length"` Count int `json:"count"` } // IPInfo represents comprehensive IP information for the /ip endpoint +// +//nolint:tagliatelle // field names are the existing JSON API type IPInfo struct { IP string `json:"ip"` PTR []string `json:"ptr,omitempty"`