diff --git a/internal/database/counts_test.go b/internal/database/counts_test.go index 8922619..b3259ba 100644 --- a/internal/database/counts_test.go +++ b/internal/database/counts_test.go @@ -128,7 +128,7 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) { // 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}, + {Prefix: "203.0.113.0/24", PeerIP: "192.0.2.1", IPVersion: ipVersionV4}, }); err != nil { t.Fatalf("DeleteLiveRouteBatch: %v", err) } @@ -327,13 +327,11 @@ func announceRoutes(db *Database, batch bool, routes []*LiveRoute) error { } // withdrawRoutes removes routes in one DeleteLiveRouteBatch, or with one -// DeleteLiveRoute each. It names each route by prefix and peer only, with no -// origin ASN, as a withdrawal from the feed does when its message carries no AS -// path. +// DeleteLiveRoute each. 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 { + if err := db.DeleteLiveRoute(route.Prefix, route.PeerIP); err != nil { return err } } diff --git a/internal/database/database.go b/internal/database/database.go index 468bd2b..6461336 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -228,21 +228,22 @@ func (d *Database) beginTx() (*loggingTx, error) { } // 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. +// INSERT. A live route is keyed by prefix and peer, so an announcement replaces +// the peer's route for the prefix whatever its origin AS. The UPDATE's +// rows-affected count (1 for an existing key, 0 for a new one) is what lets the +// in-memory route counters stay exact without a COUNT(*). Callers hold the +// database write lock, so no other writer can insert the same key between the +// two statements. The id column is set only on INSERT, so an updated route +// keeps its original id, exactly as the previous ON CONFLICT upsert did. 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 = ?, origin_asn = ?, as_path = ?, + next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?` insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` - 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 = ?, origin_asn = ?, as_path = ?, + next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?` insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` @@ -261,8 +262,8 @@ const ( // 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) + res, err := upd.Exec(route.MaskLength, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated, + route.Prefix, route.PeerIP) if err != nil { return false, false, err } @@ -293,8 +294,8 @@ func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string // 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) + res, err := upd.Exec(route.MaskLength, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated, + route.Prefix, route.PeerIP) if err != nil { return false, false, err } @@ -441,32 +442,18 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { } }() - // No longer need to separate deletions since we handle them in the loop below - // 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 = ?`) + deleteV4, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`) if err != nil { - return fmt.Errorf("failed to prepare IPv4 delete with origin statement: %w", err) + return fmt.Errorf("failed to prepare IPv4 delete statement: %w", err) } - defer func() { _ = stmtV4WithOrigin.Close() }() + defer func() { _ = deleteV4.Close() }() - stmtV4WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`) + deleteV6, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`) if err != nil { - return fmt.Errorf("failed to prepare IPv4 delete without origin statement: %w", err) + return fmt.Errorf("failed to prepare IPv6 delete 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() }() + defer func() { _ = deleteV6.Close() }() hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL) if err != nil { @@ -485,38 +472,18 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { // 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 - } - } 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) + stmt := deleteV4 + if del.IPVersion != ipVersionV4 { + stmt = deleteV6 } + res, err := stmt.Exec(del.Prefix, 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. + // The feed often withdraws a route that is not live, which deletes + // nothing, so the in-memory route counters use the rows-affected count. affected, err := res.RowsAffected() if err != nil { return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err) @@ -1282,9 +1249,8 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error { return nil } -// 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 { +// DeleteLiveRoute deletes the live route the peer has for the prefix +func (d *Database) DeleteLiveRoute(prefix, peerIP string) error { d.lock("DeleteLiveRoute") defer d.unlock() @@ -1309,22 +1275,12 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) }() // 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. + // so the delete carries no dynamically built SQL. var res sql.Result - switch { - case isV4 && originASN == 0: + if isV4 { 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: + } else { 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) } if err != nil { return err diff --git a/internal/database/database_test.go b/internal/database/database_test.go index 1b52c08..6ecaf75 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -13,6 +13,7 @@ import ( "git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/logger" "github.com/google/uuid" + "github.com/mattn/go-sqlite3" ) // tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in @@ -122,8 +123,8 @@ 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", PeerIP: testPeerIP, IPVersion: ipVersionV4}, + {Prefix: "2001:db8:1:2::/64", PeerIP: testPeerIP, IPVersion: ipVersionV6}, }) if err != nil { t.Fatalf("DeleteLiveRouteBatch: %v", err) @@ -205,6 +206,40 @@ func TestGetIPInfoIgnoresShortRoutes(t *testing.T) { lookup("2001:db8::1", "2001:db8::/32", 64502) } +// TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer checks that each live route +// table refuses a second route from a peer for a prefix, even with another +// origin AS. An upsert updates the peer's route before it would insert one, so +// only an insert straight into the table shows what the table allows. +func TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer(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() }() + + const otherOriginASN = 64501 + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + for _, tt := range []struct { + route *LiveRoute + insertSQL string + }{ + {route: mkV4Route(t, "198.51.100.0/24", 64500, ts), insertSQL: insertLiveRouteV4SQL}, + {route: mkV6Route("2001:db8::/32", 64500, ts), insertSQL: insertLiveRouteV6SQL}, + } { + if err := db.UpsertLiveRoute(tt.route); err != nil { + t.Fatalf("UpsertLiveRoute(%s): %v", tt.route.Prefix, err) + } + + _, err := db.db.Exec(tt.insertSQL, uuid.NewString(), tt.route.Prefix, tt.route.MaskLength, + otherOriginASN, tt.route.PeerIP, "[]", tt.route.NextHop, ts) + var sqliteErr sqlite3.Error + if !errors.As(err, &sqliteErr) || sqliteErr.ExtendedCode != sqlite3.ErrConstraintUnique { + t.Errorf("second route from peer %s for %s: error %v, want a UNIQUE constraint error", + tt.route.PeerIP, tt.route.Prefix, 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 @@ -306,7 +341,6 @@ func TestVacuumReturnsFreePages(t *testing.T) { routes = append(routes, route) deletions = append(deletions, LiveRouteDeletion{ Prefix: route.Prefix, - OriginASN: asn, PeerIP: route.PeerIP, IPVersion: ipVersionV4, }) diff --git a/internal/database/interface.go b/internal/database/interface.go index 504c252..bbbe7f4 100644 --- a/internal/database/interface.go +++ b/internal/database/interface.go @@ -55,7 +55,7 @@ type Store interface { // Live route operations UpsertLiveRoute(route *LiveRoute) error UpsertLiveRouteBatch(routes []*LiveRoute) error - DeleteLiveRoute(prefix string, originASN int, peerIP string) error + DeleteLiveRoute(prefix, peerIP string) error DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) GetPrefixDistributionContext(ctx context.Context) (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) diff --git a/internal/database/models.go b/internal/database/models.go index a8e98b0..54c54f8 100644 --- a/internal/database/models.go +++ b/internal/database/models.go @@ -114,7 +114,6 @@ type IPInfo struct { // LiveRouteDeletion represents parameters for deleting a live route type LiveRouteDeletion struct { Prefix string - OriginASN int PeerIP string IPVersion int } diff --git a/internal/database/schema.sql b/internal/database/schema.sql index 6543c5b..a45c855 100644 --- a/internal/database/schema.sql +++ b/internal/database/schema.sql @@ -107,7 +107,7 @@ CREATE TABLE IF NOT EXISTS live_routes_v4 ( as_path TEXT NOT NULL, -- JSON array next_hop TEXT NOT NULL, last_updated DATETIME NOT NULL, - UNIQUE(prefix, origin_asn, peer_ip) + UNIQUE(prefix, peer_ip) -- a peer has at most one route for a prefix ); -- IPv6 routing table maintained by PrefixHandler @@ -120,7 +120,7 @@ CREATE TABLE IF NOT EXISTS live_routes_v6 ( as_path TEXT NOT NULL, -- JSON array next_hop TEXT NOT NULL, last_updated DATETIME NOT NULL, - UNIQUE(prefix, origin_asn, peer_ip) + UNIQUE(prefix, peer_ip) -- a peer has at most one route for a prefix ); -- Indexes for live_routes_v4 table diff --git a/internal/routewatch/app_integration_test.go b/internal/routewatch/app_integration_test.go index a359d03..634e9c6 100644 --- a/internal/routewatch/app_integration_test.go +++ b/internal/routewatch/app_integration_test.go @@ -205,7 +205,7 @@ func (m *mockStore) UpsertLiveRoute(route *database.LiveRoute) error { } // DeleteLiveRoute mock implementation -func (m *mockStore) DeleteLiveRoute(prefix string, originASN int, peerIP string) error { +func (m *mockStore) DeleteLiveRoute(prefix, peerIP string) error { // Simple mock - just return nil return nil } diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 59f3991..9833b37 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -108,14 +108,14 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { // Withdrawals are queued before announcements, because the batch keeps // each peer's last update for a prefix: a message that lists a prefix as // both withdrawn and announced keeps the route (RFC 4271 section 4.3). + // A withdrawal names only the prefix and peer: the message's AS path, and + // so its origin AS, belongs to the announced prefixes. for _, prefix := range msg.Withdrawals { h.batch = append(h.batch, prefixUpdate{ prefix: canonicalPrefix(prefix), - originASN: originASN, // Use the originASN from path if available peer: msg.Peer, messageType: "withdrawal", timestamp: timestamp, - path: msg.Path, }) // Record withdrawal in metrics if h.metrics != nil { @@ -221,7 +221,6 @@ func (h *PrefixHandler) flushBatchLocked() { // Create deletion record for batch delete routesToDelete = append(routesToDelete, database.LiveRouteDeletion{ Prefix: update.prefix, - OriginASN: update.originASN, PeerIP: update.peer, IPVersion: ipVersion, }) @@ -425,51 +424,24 @@ func (h *PrefixHandler) processAnnouncementDirect(update prefixUpdate) { // processWithdrawalDirect handles removing a route directly without prefix table // nolint:unused // kept for potential future use func (h *PrefixHandler) processWithdrawalDirect(update prefixUpdate) { - // For withdrawals, we need to delete the route from live_routes - if update.originASN > 0 { - if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil { - h.logger.Error("Failed to delete live route", - "prefix", update.prefix, - "origin_asn", update.originASN, - "peer", update.peer, - "error", err, - ) - } - } else { - // If no origin ASN, just delete all routes for this prefix from this peer - if err := h.db.DeleteLiveRoute(update.prefix, 0, update.peer); err != nil { - h.logger.Error("Failed to delete live route (no origin ASN)", - "prefix", update.prefix, - "peer", update.peer, - "error", err, - ) - } + if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil { + h.logger.Error("Failed to delete live route", + "prefix", update.prefix, + "peer", update.peer, + "error", err, + ) } } // processWithdrawal handles removing a route from the live routing table // nolint:unused // kept for potential future use func (h *PrefixHandler) processWithdrawal(_ *database.Prefix, update prefixUpdate) { - // For withdrawals, we need to delete the route from live_routes - // Since we have the origin ASN from the update, we can delete the specific route - if update.originASN > 0 { - if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil { - h.logger.Error("Failed to delete live route", - "prefix", update.prefix, - "origin_asn", update.originASN, - "peer", update.peer, - "error", err, - ) - } - } else { - // If no origin ASN, just delete all routes for this prefix from this peer - if err := h.db.DeleteLiveRoute(update.prefix, 0, update.peer); err != nil { - h.logger.Error("Failed to delete live route (no origin ASN)", - "prefix", update.prefix, - "peer", update.peer, - "error", err, - ) - } + if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil { + h.logger.Error("Failed to delete live route", + "prefix", update.prefix, + "peer", update.peer, + "error", err, + ) } } diff --git a/internal/routewatch/prefixhandler_test.go b/internal/routewatch/prefixhandler_test.go index fcfa9a3..54c33dd 100644 --- a/internal/routewatch/prefixhandler_test.go +++ b/internal/routewatch/prefixhandler_test.go @@ -3,6 +3,7 @@ package routewatch import ( "context" "errors" + "fmt" "slices" "testing" "time" @@ -182,32 +183,155 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) { t.Errorf("peers with a live route for %s = %v, want %v", prefix, gotPeers, tt.wantPeers) } - stats, err := db.GetStatsContext(ctx) - if err != nil { - t.Fatalf("GetStatsContext: %v", err) - } - tableRoutesV4, _, err := db.GetLiveRouteCountsContext(ctx) - if err != nil { - t.Fatalf("GetLiveRouteCountsContext: %v", err) - } - if stats.IPv4Routes != len(tt.wantPeers) || tableRoutesV4 != len(tt.wantPeers) { - t.Errorf("IPv4 live routes: %d in memory, %d in the table, want %d", - stats.IPv4Routes, tableRoutesV4, len(tt.wantPeers)) - } - var wantDistribution []database.PrefixDistribution if len(tt.wantPeers) > 0 { wantDistribution = []database.PrefixDistribution{{MaskLength: prefixMaskLength, Count: 1}} } - queryDistributionV4, _, err := db.GetPrefixDistributionContext(ctx) - if err != nil { - t.Fatalf("GetPrefixDistributionContext: %v", err) - } - if !slices.Equal(stats.IPv4PrefixDistribution, wantDistribution) || - !slices.Equal(queryDistributionV4, wantDistribution) { - t.Errorf("IPv4 prefix distribution: %v in memory, %v from the table, want %v", - stats.IPv4PrefixDistribution, queryDistributionV4, wantDistribution) - } + assertCounts(t, db, ipv4Version, len(tt.wantPeers), wantDistribution) }) } } + +// TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer runs updates from one peer +// through the prefix handler into a real database, flushing each message as its +// own batch, once with IPv4 prefixes and once with IPv6 prefixes. A peer has at +// most one route for a prefix: a later announcement replaces it whatever its +// origin AS, and a withdrawal removes it whatever AS the message's AS path ends +// in. The in-memory route count and prefix distribution must match the route +// table. +func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) { + families := []struct { + ipVersion int + prefix, otherPrefix string + // The mask length of both prefixes. + maskLength int + }{ + {ipVersion: ipv4Version, prefix: "198.51.100.0/24", otherPrefix: "203.0.113.0/24", maskLength: 24}, + {ipVersion: ipv6Version, prefix: "2001:db8:1::/48", otherPrefix: "2001:db8:2::/48", maskLength: 48}, + } + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + announce := func(path ristypes.ASPath, announced string) *ristypes.RISMessage { + return &ristypes.RISMessage{ + Peer: testPeerIP, + Path: path, + ParsedTimestamp: ts, + Announcements: []ristypes.RISAnnouncement{{NextHop: testPeerIP, Prefixes: []string{announced}}}, + } + } + + for _, family := range families { + prefix, otherPrefix := family.prefix, family.otherPrefix + + // One message that announces otherPrefix with origin AS testASNB and + // withdraws prefix. + announceOtherAndWithdraw := announce(ristypes.ASPath{testASNA, testASNB}, otherPrefix) + announceOtherAndWithdraw.Withdrawals = []string{prefix} + + tests := []struct { + name string + messages []*ristypes.RISMessage + // The origin AS of the peer's live route for each prefix. A prefix + // left out must have no live route. + wantOrigins map[string]int + }{ + { + name: "the peer announces the prefix with a new origin AS", + messages: []*ristypes.RISMessage{ + announce(ristypes.ASPath{testASNA}, prefix), + announce(ristypes.ASPath{testASNA, testASNB}, prefix), + }, + wantOrigins: map[string]int{prefix: testASNB}, + }, + { + name: "one message announces a prefix with another origin AS and withdraws the prefix", + messages: []*ristypes.RISMessage{ + announce(ristypes.ASPath{testASNA}, prefix), + announceOtherAndWithdraw, + }, + wantOrigins: map[string]int{otherPrefix: testASNB}, + }, + } + + for _, tt := range tests { + t.Run(fmt.Sprintf("IPv%d %s", family.ipVersion, tt.name), func(t *testing.T) { + db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + // Built without NewPrefixHandler's flush timer, so the test flushes itself. + h := &PrefixHandler{db: db, logger: logger.New()} + for _, msg := range tt.messages { + h.HandleMessage(msg) + h.mu.Lock() + h.flushBatchLocked() + h.mu.Unlock() + } + + for _, p := range []string{prefix, otherPrefix} { + routes, _, err := db.GetPrefixDetailsContext(context.Background(), p) + if err != nil && !errors.Is(err, database.ErrNoRoute) { + t.Fatalf("GetPrefixDetailsContext(%s): %v", p, err) + } + var gotOrigins []int + for _, route := range routes { + gotOrigins = append(gotOrigins, route.OriginASN) + } + var wantOrigins []int + if origin, ok := tt.wantOrigins[p]; ok { + wantOrigins = []int{origin} + } + if !slices.Equal(gotOrigins, wantOrigins) { + t.Errorf("origin ASes of the live routes for %s = %v, want %v", p, gotOrigins, wantOrigins) + } + } + + wantDistribution := []database.PrefixDistribution{ + {MaskLength: family.maskLength, Count: len(tt.wantOrigins)}, + } + assertCounts(t, db, family.ipVersion, len(tt.wantOrigins), wantDistribution) + }) + } + } +} + +// assertCounts checks that the live route count and prefix distribution of one +// IP version are wantRoutes and wantDistribution, both as held in memory and as +// read from the route table. +func assertCounts(t *testing.T, db *database.Database, ipVersion, wantRoutes int, + wantDistribution []database.PrefixDistribution) { + t.Helper() + + ctx := context.Background() + stats, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + tableRoutesV4, tableRoutesV6, err := db.GetLiveRouteCountsContext(ctx) + if err != nil { + t.Fatalf("GetLiveRouteCountsContext: %v", err) + } + queryDistributionV4, queryDistributionV6, err := db.GetPrefixDistributionContext(ctx) + if err != nil { + t.Fatalf("GetPrefixDistributionContext: %v", err) + } + + memoryRoutes, tableRoutes := stats.IPv4Routes, tableRoutesV4 + memoryDistribution, queryDistribution := stats.IPv4PrefixDistribution, queryDistributionV4 + if ipVersion == ipv6Version { + memoryRoutes, tableRoutes = stats.IPv6Routes, tableRoutesV6 + memoryDistribution, queryDistribution = stats.IPv6PrefixDistribution, queryDistributionV6 + } + + if memoryRoutes != wantRoutes || tableRoutes != wantRoutes { + t.Errorf("IPv%d live routes: %d in memory, %d in the table, want %d", + ipVersion, memoryRoutes, tableRoutes, wantRoutes) + } + if !slices.Equal(memoryDistribution, wantDistribution) || + !slices.Equal(queryDistribution, wantDistribution) { + t.Errorf("IPv%d prefix distribution: %v in memory, %v from the table, want %v", + ipVersion, memoryDistribution, queryDistribution, wantDistribution) + } +}