Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
31c74bc81c |
@@ -128,7 +128,7 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) {
|
|||||||
|
|
||||||
// A withdrawal removes one route.
|
// A withdrawal removes one route.
|
||||||
if err := db.DeleteLiveRouteBatch([]LiveRouteDeletion{
|
if err := db.DeleteLiveRouteBatch([]LiveRouteDeletion{
|
||||||
{Prefix: "203.0.113.0/24", PeerIP: "192.0.2.1", IPVersion: ipVersionV4},
|
{Prefix: "203.0.113.0/24", OriginASN: 64501, PeerIP: "192.0.2.1", IPVersion: ipVersionV4},
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
t.Fatalf("DeleteLiveRouteBatch: %v", err)
|
t.Fatalf("DeleteLiveRouteBatch: %v", err)
|
||||||
}
|
}
|
||||||
@@ -327,11 +327,13 @@ func announceRoutes(db *Database, batch bool, routes []*LiveRoute) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// withdrawRoutes removes routes in one DeleteLiveRouteBatch, or with one
|
// withdrawRoutes removes routes in one DeleteLiveRouteBatch, or with one
|
||||||
// DeleteLiveRoute each.
|
// 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.
|
||||||
func withdrawRoutes(db *Database, batch bool, routes []*LiveRoute) error {
|
func withdrawRoutes(db *Database, batch bool, routes []*LiveRoute) error {
|
||||||
if !batch {
|
if !batch {
|
||||||
for _, route := range routes {
|
for _, route := range routes {
|
||||||
if err := db.DeleteLiveRoute(route.Prefix, route.PeerIP); err != nil {
|
if err := db.DeleteLiveRoute(route.Prefix, 0, route.PeerIP); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+257
-64
@@ -8,6 +8,7 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"math/rand/v2"
|
||||||
"net"
|
"net"
|
||||||
"net/netip"
|
"net/netip"
|
||||||
"os"
|
"os"
|
||||||
@@ -61,6 +62,17 @@ const (
|
|||||||
sqliteHardHeapLimitBytes = 1610612736
|
sqliteHardHeapLimitBytes = 1610612736
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Bounds on the work GetRandomPrefixesByLength does to pick its sample.
|
||||||
|
const (
|
||||||
|
// maxPrefixesToShuffle is the most prefixes of one mask length that are
|
||||||
|
// read in full and shuffled to pick the sample. A mask length with more is
|
||||||
|
// sampled with random lookups.
|
||||||
|
maxPrefixesToShuffle = 5000
|
||||||
|
// sampleLookupsPerPrefix is the most random lookups made for each prefix
|
||||||
|
// asked for.
|
||||||
|
sampleLookupsPerPrefix = 4
|
||||||
|
)
|
||||||
|
|
||||||
// Common errors
|
// Common errors
|
||||||
var (
|
var (
|
||||||
// ErrInvalidIP is returned when an IP address is malformed
|
// ErrInvalidIP is returned when an IP address is malformed
|
||||||
@@ -228,22 +240,21 @@ func (d *Database) beginTx() (*loggingTx, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// A live-route upsert is an UPDATE followed, only when no row matched, by an
|
// A live-route upsert is an UPDATE followed, only when no row matched, by an
|
||||||
// INSERT. A live route is keyed by prefix and peer, so an announcement replaces
|
// INSERT. The UPDATE's rows-affected count (1 for an existing key, 0 for a new
|
||||||
// the peer's route for the prefix whatever its origin AS. The UPDATE's
|
// one) is what lets the in-memory route counters stay exact without a COUNT(*).
|
||||||
// rows-affected count (1 for an existing key, 0 for a new one) is what lets the
|
// Callers hold the database write lock, so no other writer can insert the same
|
||||||
// in-memory route counters stay exact without a COUNT(*). Callers hold the
|
// key between the two statements. The id column is set only on INSERT, so an
|
||||||
// database write lock, so no other writer can insert the same key between the
|
// updated route keeps its original id, exactly as the previous ON CONFLICT
|
||||||
// two statements. The id column is set only on INSERT, so an updated route
|
// upsert did.
|
||||||
// keeps its original id, exactly as the previous ON CONFLICT upsert did.
|
|
||||||
const (
|
const (
|
||||||
updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, origin_asn = ?, as_path = ?,
|
updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?,
|
||||||
next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
|
last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
|
||||||
|
|
||||||
insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip,
|
insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip,
|
||||||
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
||||||
|
|
||||||
updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, origin_asn = ?, as_path = ?,
|
updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?,
|
||||||
next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
|
last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
|
||||||
|
|
||||||
insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip,
|
insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip,
|
||||||
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
||||||
@@ -262,8 +273,8 @@ const (
|
|||||||
// live route for its prefix.
|
// live route for its prefix.
|
||||||
func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
|
func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
|
||||||
inserted, newPrefix bool, err error) {
|
inserted, newPrefix bool, err error) {
|
||||||
res, err := upd.Exec(route.MaskLength, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
|
res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
|
||||||
route.Prefix, route.PeerIP)
|
route.Prefix, route.OriginASN, route.PeerIP)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, false, err
|
return false, false, err
|
||||||
}
|
}
|
||||||
@@ -294,8 +305,8 @@ func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string
|
|||||||
// live route for its prefix.
|
// live route for its prefix.
|
||||||
func upsertRouteRowV6(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
|
func upsertRouteRowV6(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
|
||||||
inserted, newPrefix bool, err error) {
|
inserted, newPrefix bool, err error) {
|
||||||
res, err := upd.Exec(route.MaskLength, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
|
res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
|
||||||
route.Prefix, route.PeerIP)
|
route.Prefix, route.OriginASN, route.PeerIP)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, false, err
|
return false, false, err
|
||||||
}
|
}
|
||||||
@@ -442,18 +453,32 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
|
|||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// Prepare statements for both IPv4 and IPv6 tables
|
// No longer need to separate deletions since we handle them in the loop below
|
||||||
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 statement: %w", err)
|
|
||||||
}
|
|
||||||
defer func() { _ = deleteV4.Close() }()
|
|
||||||
|
|
||||||
deleteV6, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`)
|
// Prepare statements for both IPv4 and IPv6 tables
|
||||||
|
stmtV4WithOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to prepare IPv6 delete statement: %w", err)
|
return fmt.Errorf("failed to prepare IPv4 delete with origin statement: %w", err)
|
||||||
}
|
}
|
||||||
defer func() { _ = deleteV6.Close() }()
|
defer func() { _ = stmtV4WithOrigin.Close() }()
|
||||||
|
|
||||||
|
stmtV4WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to prepare IPv4 delete without origin statement: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = stmtV4WithoutOrigin.Close() }()
|
||||||
|
|
||||||
|
stmtV6WithOrigin, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to prepare IPv6 delete with origin statement: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = stmtV6WithOrigin.Close() }()
|
||||||
|
|
||||||
|
stmtV6WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to prepare IPv6 delete without origin statement: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = stmtV6WithoutOrigin.Close() }()
|
||||||
|
|
||||||
hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL)
|
hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -472,18 +497,38 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
|
|||||||
// Mask lengths of the prefixes this batch leaves with no live route.
|
// Mask lengths of the prefixes this batch leaves with no live route.
|
||||||
var gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6 []int
|
var gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6 []int
|
||||||
for _, del := range deletions {
|
for _, del := range deletions {
|
||||||
stmt := deleteV4
|
var stmt *sql.Stmt
|
||||||
if del.IPVersion != ipVersionV4 {
|
|
||||||
stmt = deleteV6
|
// 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)
|
||||||
}
|
}
|
||||||
|
|
||||||
res, err := stmt.Exec(del.Prefix, del.PeerIP)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err)
|
return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// The feed often withdraws a route that is not live, which deletes
|
// A deletion with no origin ASN can remove several rows, so use the
|
||||||
// nothing, so the in-memory route counters use the rows-affected count.
|
// exact rows-affected count to keep the in-memory route counters right.
|
||||||
affected, err := res.RowsAffected()
|
affected, err := res.RowsAffected()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err)
|
return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err)
|
||||||
@@ -1249,8 +1294,9 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// DeleteLiveRoute deletes the live route the peer has for the prefix
|
// DeleteLiveRoute deletes a live route
|
||||||
func (d *Database) DeleteLiveRoute(prefix, peerIP string) error {
|
// If originASN is 0, deletes all routes for the prefix/peer combination
|
||||||
|
func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) error {
|
||||||
d.lock("DeleteLiveRoute")
|
d.lock("DeleteLiveRoute")
|
||||||
defer d.unlock()
|
defer d.unlock()
|
||||||
|
|
||||||
@@ -1275,12 +1321,22 @@ func (d *Database) DeleteLiveRoute(prefix, peerIP string) error {
|
|||||||
}()
|
}()
|
||||||
|
|
||||||
// Literal per-table queries (rather than one formatted with the table name)
|
// Literal per-table queries (rather than one formatted with the table name)
|
||||||
// so the delete carries no dynamically built SQL.
|
// so the delete carries no dynamically built SQL. A delete with no origin
|
||||||
|
// ASN can remove several rows.
|
||||||
var res sql.Result
|
var res sql.Result
|
||||||
if isV4 {
|
switch {
|
||||||
|
case isV4 && originASN == 0:
|
||||||
res, err = tx.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
|
res, err = tx.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
|
||||||
} else {
|
case isV4:
|
||||||
|
res, err = tx.Exec(
|
||||||
|
`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
|
||||||
|
prefix, originASN, peerIP)
|
||||||
|
case originASN == 0:
|
||||||
res, err = tx.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
|
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 {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -1667,45 +1723,183 @@ func (d *Database) GetRandomPrefixesByLength(maskLength, ipVersion, limit int) (
|
|||||||
return d.GetRandomPrefixesByLengthContext(context.Background(), maskLength, ipVersion, limit)
|
return d.GetRandomPrefixesByLengthContext(context.Background(), maskLength, ipVersion, limit)
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetRandomPrefixesByLengthContext returns a random sample of prefixes with context support
|
// GetRandomPrefixesByLengthContext returns limit prefixes with the given mask
|
||||||
|
// length picked at random, or all of them when there are fewer, each once with
|
||||||
|
// its most recently updated live route, for a limit up to
|
||||||
|
// maxPrefixesToShuffle. However many routes there are, it reads at most
|
||||||
|
// maxPrefixesToShuffle+1 prefixes and makes at most sampleLookupsPerPrefix
|
||||||
|
// lookups for each prefix asked for.
|
||||||
func (d *Database) GetRandomPrefixesByLengthContext(
|
func (d *Database) GetRandomPrefixesByLengthContext(
|
||||||
ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error) {
|
ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error) {
|
||||||
// Select unique prefixes with their most recent route information
|
|
||||||
tableName := "live_routes_v4"
|
tableName := "live_routes_v4"
|
||||||
if ipVersion == ipVersionV6 {
|
if ipVersion == ipVersionV6 {
|
||||||
tableName = "live_routes_v6"
|
tableName = "live_routes_v6"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Reading one prefix more than maxPrefixesToShuffle tells whether these
|
||||||
|
// are all the prefixes with this mask length.
|
||||||
|
prefixes, err := d.firstPrefixesByLength(ctx, tableName, maskLength, maxPrefixesToShuffle+1)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
rand.Shuffle(len(prefixes), func(i, j int) {
|
||||||
|
prefixes[i], prefixes[j] = prefixes[j], prefixes[i]
|
||||||
|
})
|
||||||
|
if len(prefixes) > maxPrefixesToShuffle {
|
||||||
|
prefixes, err = d.samplePrefixesByLength(ctx, tableName, maskLength, limit, prefixes)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
prefixes = prefixes[:min(limit, len(prefixes))]
|
||||||
|
|
||||||
|
return d.newestRouteOfEachPrefix(ctx, tableName, ipVersion, prefixes)
|
||||||
|
}
|
||||||
|
|
||||||
|
// firstPrefixesByLength returns the first n distinct prefixes, in text order,
|
||||||
|
// with the given mask length. It reads only their entries in the mask length
|
||||||
|
// and prefix index.
|
||||||
|
func (d *Database) firstPrefixesByLength(
|
||||||
|
ctx context.Context, tableName string, maskLength, n int) ([]string, error) {
|
||||||
//nolint:gosec // Table name is hardcoded based on IP version
|
//nolint:gosec // Table name is hardcoded based on IP version
|
||||||
query := fmt.Sprintf(`
|
query := fmt.Sprintf(`
|
||||||
WITH unique_prefixes AS (
|
SELECT DISTINCT prefix FROM %s
|
||||||
SELECT prefix, MAX(last_updated) as max_updated
|
WHERE mask_length = ?
|
||||||
FROM %s
|
ORDER BY prefix
|
||||||
WHERE mask_length = ?
|
LIMIT ?
|
||||||
GROUP BY prefix
|
`, tableName)
|
||||||
ORDER BY RANDOM()
|
|
||||||
LIMIT ?
|
|
||||||
)
|
|
||||||
SELECT lr.prefix, lr.mask_length, lr.origin_asn, lr.as_path,
|
|
||||||
lr.peer_ip, lr.last_updated
|
|
||||||
FROM %s lr
|
|
||||||
INNER JOIN unique_prefixes up ON lr.prefix = up.prefix AND lr.last_updated = up.max_updated
|
|
||||||
WHERE lr.mask_length = ?
|
|
||||||
`, tableName, tableName)
|
|
||||||
|
|
||||||
rows, err := d.db.QueryContext(ctx, query, maskLength, limit, maskLength)
|
rows, err := d.db.QueryContext(ctx, query, maskLength, n)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("failed to query random prefixes: %w", err)
|
return nil, fmt.Errorf("failed to list prefixes: %w", err)
|
||||||
}
|
}
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
_ = rows.Close()
|
|
||||||
}()
|
|
||||||
|
|
||||||
var routes []LiveRoute
|
var prefixes []string
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
var route LiveRoute
|
var prefix string
|
||||||
|
if err := rows.Scan(&prefix); err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to scan prefix: %w", err)
|
||||||
|
}
|
||||||
|
prefixes = append(prefixes, prefix)
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to list prefixes: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return prefixes, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// samplePrefixesByLength picks limit distinct prefixes with the given mask
|
||||||
|
// length at random, in at most sampleLookupsPerPrefix lookups for each prefix
|
||||||
|
// asked for. ORDER BY RANDOM() would read every route with that mask length.
|
||||||
|
// Instead each lookup draws a row id between those of the first and last route
|
||||||
|
// with that mask length and takes the prefix of the first such route at or
|
||||||
|
// after it. SQLite keeps the row id at the end of every index entry, so the
|
||||||
|
// mask length index is ordered by mask length, then row id, and the lookup is
|
||||||
|
// one search of it.
|
||||||
|
//
|
||||||
|
// The sample is uneven, as the page needs a varied sample, not an even one: a
|
||||||
|
// route is picked in proportion to the row ids between it and the previous
|
||||||
|
// route with that mask length, including the row ids of routes with other mask
|
||||||
|
// lengths and of withdrawn routes. A route that follows a long run of
|
||||||
|
// withdrawn routes is picked far more often than one that directly follows
|
||||||
|
// another route with that mask length.
|
||||||
|
//
|
||||||
|
// Should the lookups keep landing on prefixes already picked, the rest are
|
||||||
|
// taken from shuffled, prefixes of that length in random order.
|
||||||
|
func (d *Database) samplePrefixesByLength(
|
||||||
|
ctx context.Context, tableName string, maskLength, limit int, shuffled []string) ([]string, error) {
|
||||||
|
//nolint:gosec // Table name is hardcoded based on IP version
|
||||||
|
rangeQuery := fmt.Sprintf(`
|
||||||
|
SELECT
|
||||||
|
(SELECT MIN(rowid) FROM %[1]s WHERE mask_length = ?),
|
||||||
|
(SELECT MAX(rowid) FROM %[1]s WHERE mask_length = ?)
|
||||||
|
`, tableName)
|
||||||
|
|
||||||
|
var first, last sql.NullInt64
|
||||||
|
if err := d.db.QueryRowContext(ctx, rangeQuery, maskLength, maskLength).Scan(&first, &last); err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to read row id range: %w", err)
|
||||||
|
}
|
||||||
|
if !first.Valid {
|
||||||
|
// Every route with this mask length was withdrawn since they were listed.
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
//nolint:gosec // Table name is hardcoded based on IP version
|
||||||
|
lookupQuery := fmt.Sprintf(`
|
||||||
|
SELECT prefix FROM %s
|
||||||
|
WHERE mask_length = ? AND rowid >= ?
|
||||||
|
ORDER BY rowid
|
||||||
|
LIMIT 1
|
||||||
|
`, tableName)
|
||||||
|
|
||||||
|
stmt, err := d.db.PrepareContext(ctx, lookupQuery)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to prepare prefix lookup: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = stmt.Close() }()
|
||||||
|
|
||||||
|
picked := make(map[string]bool, limit)
|
||||||
|
prefixes := make([]string, 0, limit)
|
||||||
|
pick := func(prefix string) {
|
||||||
|
if !picked[prefix] {
|
||||||
|
picked[prefix] = true
|
||||||
|
prefixes = append(prefixes, prefix)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for lookups := 0; lookups < limit*sampleLookupsPerPrefix && len(prefixes) < limit; lookups++ {
|
||||||
|
//nolint:gosec // The sample needs variety, not unpredictability.
|
||||||
|
rowID := first.Int64 + rand.Int64N(last.Int64-first.Int64+1)
|
||||||
|
|
||||||
|
var prefix string
|
||||||
|
err := stmt.QueryRowContext(ctx, maskLength, rowID).Scan(&prefix)
|
||||||
|
if errors.Is(err, sql.ErrNoRows) {
|
||||||
|
// The routes from rowID on were withdrawn since the range was read.
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to look up prefix: %w", err)
|
||||||
|
}
|
||||||
|
pick(prefix)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, prefix := range shuffled {
|
||||||
|
if len(prefixes) == limit {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
pick(prefix)
|
||||||
|
}
|
||||||
|
|
||||||
|
return prefixes, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// newestRouteOfEachPrefix returns the most recently updated live route of each
|
||||||
|
// prefix, one lookup on the prefix index per prefix. A prefix whose routes were
|
||||||
|
// all withdrawn since it was picked is left out.
|
||||||
|
func (d *Database) newestRouteOfEachPrefix(
|
||||||
|
ctx context.Context, tableName string, ipVersion int, prefixes []string) ([]LiveRoute, error) {
|
||||||
|
//nolint:gosec // Table name is hardcoded based on IP version
|
||||||
|
query := fmt.Sprintf(`
|
||||||
|
SELECT prefix, mask_length, origin_asn, as_path, peer_ip, last_updated
|
||||||
|
FROM %s
|
||||||
|
WHERE prefix = ?
|
||||||
|
ORDER BY last_updated DESC
|
||||||
|
LIMIT 1
|
||||||
|
`, tableName)
|
||||||
|
|
||||||
|
stmt, err := d.db.PrepareContext(ctx, query)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to prepare route query: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = stmt.Close() }()
|
||||||
|
|
||||||
|
routes := make([]LiveRoute, 0, len(prefixes))
|
||||||
|
for _, prefix := range prefixes {
|
||||||
|
route := LiveRoute{IPVersion: ipVersion}
|
||||||
var pathJSON string
|
var pathJSON string
|
||||||
err := rows.Scan(
|
err := stmt.QueryRowContext(ctx, prefix).Scan(
|
||||||
&route.Prefix,
|
&route.Prefix,
|
||||||
&route.MaskLength,
|
&route.MaskLength,
|
||||||
&route.OriginASN,
|
&route.OriginASN,
|
||||||
@@ -1713,17 +1907,16 @@ func (d *Database) GetRandomPrefixesByLengthContext(
|
|||||||
&route.PeerIP,
|
&route.PeerIP,
|
||||||
&route.LastUpdated,
|
&route.LastUpdated,
|
||||||
)
|
)
|
||||||
// Set IP version based on which table we queried
|
if errors.Is(err, sql.ErrNoRows) {
|
||||||
route.IPVersion = ipVersion
|
|
||||||
if err != nil {
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to query route of %s: %w", prefix, err)
|
||||||
|
}
|
||||||
|
|
||||||
// Decode AS path
|
|
||||||
if err := json.Unmarshal([]byte(pathJSON), &route.ASPath); err != nil {
|
if err := json.Unmarshal([]byte(pathJSON), &route.ASPath); err != nil {
|
||||||
route.ASPath = []int{}
|
route.ASPath = []int{}
|
||||||
}
|
}
|
||||||
|
|
||||||
routes = append(routes, route)
|
routes = append(routes, route)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net/netip"
|
"net/netip"
|
||||||
|
"slices"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -13,7 +14,6 @@ import (
|
|||||||
"git.eeqj.de/sneak/routewatch/internal/config"
|
"git.eeqj.de/sneak/routewatch/internal/config"
|
||||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
"github.com/mattn/go-sqlite3"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in
|
// tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in
|
||||||
@@ -46,6 +46,32 @@ const (
|
|||||||
asnSecondBand = 100
|
asnSecondBand = 100
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Parameters for the prefix-length sample tests.
|
||||||
|
const (
|
||||||
|
// sampleLimit is how many prefixes the tests ask the sample for.
|
||||||
|
sampleLimit = 10
|
||||||
|
// sampledPrefixCount is how many prefixes the sampling test stores, more
|
||||||
|
// than sampleLimit so the sample has to choose among them.
|
||||||
|
sampledPrefixCount = 30
|
||||||
|
// secondV4PeerIP and secondV6PeerIP are the second peer each prefix is
|
||||||
|
// learned from, besides the one mkV4Route and mkV6Route set.
|
||||||
|
secondV4PeerIP = "192.0.2.2"
|
||||||
|
secondV6PeerIP = "2001:db8::2"
|
||||||
|
// largeSamplePrefixCount is how many prefixes the test of a mask length
|
||||||
|
// with more than maxPrefixesToShuffle prefixes stores: twice that, so about
|
||||||
|
// half of them come after the first maxPrefixesToShuffle+1 in text order.
|
||||||
|
largeSamplePrefixCount = 2 * maxPrefixesToShuffle
|
||||||
|
// largeSampleLimit is how many prefixes that test asks for. Each random
|
||||||
|
// lookup lands after the first maxPrefixesToShuffle+1 in text order about
|
||||||
|
// half the time, so none of them doing so is about as likely as
|
||||||
|
// largeSampleLimit coin tosses all coming up heads.
|
||||||
|
largeSampleLimit = 30
|
||||||
|
// liveRoutesV4Table and liveRoutesV6Table are the live route tables the
|
||||||
|
// tests pass to the sampling steps.
|
||||||
|
liveRoutesV4Table = "live_routes_v4"
|
||||||
|
liveRoutesV6Table = "live_routes_v6"
|
||||||
|
)
|
||||||
|
|
||||||
// TestGetIPInfoFindsMostSpecificLiveRoute stores nested live prefixes for both
|
// TestGetIPInfoFindsMostSpecificLiveRoute stores nested live prefixes for both
|
||||||
// families and checks that a lookup returns the most specific one covering the
|
// families and checks that a lookup returns the most specific one covering the
|
||||||
// address, ErrNoRoute when none covers it, and the next less specific prefix
|
// address, ErrNoRoute when none covers it, and the next less specific prefix
|
||||||
@@ -123,8 +149,8 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) {
|
|||||||
lookup("2001:db9::1", "")
|
lookup("2001:db9::1", "")
|
||||||
|
|
||||||
err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
|
err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
|
||||||
{Prefix: "10.1.2.0/24", PeerIP: testPeerIP, IPVersion: ipVersionV4},
|
{Prefix: "10.1.2.0/24", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV4},
|
||||||
{Prefix: "2001:db8:1:2::/64", PeerIP: testPeerIP, IPVersion: ipVersionV6},
|
{Prefix: "2001:db8:1:2::/64", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV6},
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("DeleteLiveRouteBatch: %v", err)
|
t.Fatalf("DeleteLiveRouteBatch: %v", err)
|
||||||
@@ -206,40 +232,6 @@ func TestGetIPInfoIgnoresShortRoutes(t *testing.T) {
|
|||||||
lookup("2001:db8::1", "2001:db8::/32", 64502)
|
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
|
// TestConnectionPoolPragmas holds several pooled connections open at once and
|
||||||
// checks each one carries the per-connection settings from the DSN, plus the
|
// 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
|
// process-wide hard heap limit, and sees the new file with auto_vacuum
|
||||||
@@ -341,6 +333,7 @@ func TestVacuumReturnsFreePages(t *testing.T) {
|
|||||||
routes = append(routes, route)
|
routes = append(routes, route)
|
||||||
deletions = append(deletions, LiveRouteDeletion{
|
deletions = append(deletions, LiveRouteDeletion{
|
||||||
Prefix: route.Prefix,
|
Prefix: route.Prefix,
|
||||||
|
OriginASN: asn,
|
||||||
PeerIP: route.PeerIP,
|
PeerIP: route.PeerIP,
|
||||||
IPVersion: ipVersionV4,
|
IPVersion: ipVersionV4,
|
||||||
})
|
})
|
||||||
@@ -423,3 +416,281 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) {
|
|||||||
cancel()
|
cancel()
|
||||||
wg.Wait()
|
wg.Wait()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestGetRandomPrefixesByLengthListsPrefixOnce stores one prefix of each
|
||||||
|
// family learned from two peers in the same second and checks that the
|
||||||
|
// prefix-length sample lists each of them once.
|
||||||
|
func TestGetRandomPrefixesByLengthListsPrefixOnce(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)
|
||||||
|
v4First := mkV4Route(t, "198.51.100.0/24", 64500, ts)
|
||||||
|
v4Second := mkV4Route(t, "198.51.100.0/24", 64500, ts)
|
||||||
|
v4Second.PeerIP = secondV4PeerIP
|
||||||
|
v6First := mkV6Route("2001:db8::/32", 64500, ts)
|
||||||
|
v6Second := mkV6Route("2001:db8::/32", 64500, ts)
|
||||||
|
v6Second.PeerIP = secondV6PeerIP
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*LiveRoute{v4First, v4Second, v6First, v6Second}); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, want := range []*LiveRoute{v4First, v6First} {
|
||||||
|
routes, err := db.GetRandomPrefixesByLength(want.MaskLength, want.IPVersion, sampleLimit)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetRandomPrefixesByLength(%d, %d): %v", want.MaskLength, want.IPVersion, err)
|
||||||
|
}
|
||||||
|
if len(routes) != 1 || routes[0].Prefix != want.Prefix {
|
||||||
|
t.Errorf("GetRandomPrefixesByLength(%d, %d) = %+v, want %s listed once",
|
||||||
|
want.MaskLength, want.IPVersion, routes, want.Prefix)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestGetRandomPrefixesByLengthListsNewestRoute stores a prefix learned from
|
||||||
|
// two peers at different times, the older route first, and checks that the
|
||||||
|
// prefix-length sample lists the newer route.
|
||||||
|
func TestGetRandomPrefixesByLengthListsNewestRoute(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() }()
|
||||||
|
|
||||||
|
older := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
olderRoute := mkV4Route(t, "198.51.100.0/24", 64500, older)
|
||||||
|
newerRoute := mkV4Route(t, "198.51.100.0/24", 64500, older.Add(time.Minute))
|
||||||
|
newerRoute.PeerIP = secondV4PeerIP
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*LiveRoute{olderRoute, newerRoute}); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
routes, err := db.GetRandomPrefixesByLength(newerRoute.MaskLength, ipVersionV4, sampleLimit)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetRandomPrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
if len(routes) != 1 || routes[0].PeerIP != newerRoute.PeerIP {
|
||||||
|
t.Errorf("GetRandomPrefixesByLength = %+v, want only the newer route, from %s",
|
||||||
|
routes, newerRoute.PeerIP)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestGetRandomPrefixesByLengthSamplesManyPrefixes stores more /24 prefixes
|
||||||
|
// than the sample asks for, each followed by a /16 so that routes of another
|
||||||
|
// mask length sit between the /24 rows, and each prefix learned from two peers
|
||||||
|
// in the same second. It checks both the sample the pages get, picked from the
|
||||||
|
// shuffled prefixes, and the random lookups used for a mask length with more
|
||||||
|
// than maxPrefixesToShuffle prefixes.
|
||||||
|
func TestGetRandomPrefixesByLengthSamplesManyPrefixes(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)
|
||||||
|
var routes []*LiveRoute
|
||||||
|
var stored []string
|
||||||
|
for i := range sampledPrefixCount {
|
||||||
|
prefix := fmt.Sprintf("10.0.%d.0/24", i)
|
||||||
|
for _, p := range []string{prefix, fmt.Sprintf("172.%d.0.0/16", i)} {
|
||||||
|
first := mkV4Route(t, p, 64500, ts)
|
||||||
|
second := mkV4Route(t, p, 64500, ts)
|
||||||
|
second.PeerIP = secondV4PeerIP
|
||||||
|
routes = append(routes, first, second)
|
||||||
|
}
|
||||||
|
stored = append(stored, prefix)
|
||||||
|
}
|
||||||
|
if err := db.UpsertLiveRouteBatch(routes); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
slices.Sort(stored)
|
||||||
|
|
||||||
|
maskLength := routes[0].MaskLength
|
||||||
|
sample, err := db.GetRandomPrefixesByLength(maskLength, ipVersionV4, sampleLimit)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetRandomPrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
var listed []string
|
||||||
|
for _, route := range sample {
|
||||||
|
listed = append(listed, route.Prefix)
|
||||||
|
}
|
||||||
|
checkPrefixSample(t, "GetRandomPrefixesByLength", listed, stored)
|
||||||
|
|
||||||
|
looked, err := db.samplePrefixesByLength(
|
||||||
|
context.Background(), liveRoutesV4Table, maskLength, sampleLimit, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("samplePrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
checkPrefixSample(t, "samplePrefixesByLength", looked, stored)
|
||||||
|
}
|
||||||
|
|
||||||
|
// checkPrefixSample checks that sample lists sampleLimit prefixes, each once
|
||||||
|
// and each one of stored, and that it is not just the first sampleLimit of
|
||||||
|
// stored, which is in text order.
|
||||||
|
func checkPrefixSample(t *testing.T, name string, sample, stored []string) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
if len(sample) != sampleLimit {
|
||||||
|
t.Errorf("%s listed %d prefixes, want %d", name, len(sample), sampleLimit)
|
||||||
|
}
|
||||||
|
listed := make(map[string]bool)
|
||||||
|
for _, prefix := range sample {
|
||||||
|
if listed[prefix] || !slices.Contains(stored, prefix) {
|
||||||
|
t.Errorf("%s lists %s twice or it is not one of the /24s", name, prefix)
|
||||||
|
}
|
||||||
|
listed[prefix] = true
|
||||||
|
}
|
||||||
|
if slices.Equal(slices.Sorted(slices.Values(sample)), stored[:sampleLimit]) {
|
||||||
|
t.Errorf("%s lists just the first %d prefixes in text order", name, sampleLimit)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestGetRandomPrefixesByLengthSamplesPastFirstPrefixes stores more prefixes
|
||||||
|
// of one mask length than maxPrefixesToShuffle. It checks that
|
||||||
|
// firstPrefixesByLength reads just the first maxPrefixesToShuffle+1 in text
|
||||||
|
// order, and that the sample lists the prefixes asked for, each once, some of
|
||||||
|
// them after those first ones, which only the random lookups reach.
|
||||||
|
func TestGetRandomPrefixesByLengthSamplesPastFirstPrefixes(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)
|
||||||
|
var routes []*LiveRoute
|
||||||
|
var stored []string
|
||||||
|
for i := range largeSamplePrefixCount {
|
||||||
|
route := mkV6Route(fmt.Sprintf("2001:%x::/32", i), 64500, ts)
|
||||||
|
routes = append(routes, route)
|
||||||
|
stored = append(stored, route.Prefix)
|
||||||
|
}
|
||||||
|
if err := db.UpsertLiveRouteBatch(routes); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
slices.Sort(stored)
|
||||||
|
firstPrefixes := stored[:maxPrefixesToShuffle+1]
|
||||||
|
|
||||||
|
maskLength := routes[0].MaskLength
|
||||||
|
read, err := db.firstPrefixesByLength(
|
||||||
|
context.Background(), liveRoutesV6Table, maskLength, maxPrefixesToShuffle+1)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("firstPrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
if !slices.Equal(read, firstPrefixes) {
|
||||||
|
t.Errorf("firstPrefixesByLength read %d prefixes, want the first %d in text order",
|
||||||
|
len(read), len(firstPrefixes))
|
||||||
|
}
|
||||||
|
|
||||||
|
sample, err := db.GetRandomPrefixesByLength(maskLength, ipVersionV6, largeSampleLimit)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetRandomPrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
if len(sample) != largeSampleLimit {
|
||||||
|
t.Errorf("GetRandomPrefixesByLength listed %d prefixes, want %d", len(sample), largeSampleLimit)
|
||||||
|
}
|
||||||
|
listed := make(map[string]bool)
|
||||||
|
pastFirstPrefixes := 0
|
||||||
|
for _, route := range sample {
|
||||||
|
if listed[route.Prefix] {
|
||||||
|
t.Errorf("GetRandomPrefixesByLength lists %s twice", route.Prefix)
|
||||||
|
}
|
||||||
|
listed[route.Prefix] = true
|
||||||
|
if !slices.Contains(firstPrefixes, route.Prefix) {
|
||||||
|
pastFirstPrefixes++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if pastFirstPrefixes == 0 {
|
||||||
|
t.Errorf("GetRandomPrefixesByLength lists only prefixes among the first %d in text order",
|
||||||
|
len(firstPrefixes))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestSamplePrefixesByLengthFillsUpFromShuffled stores a single prefix, so
|
||||||
|
// every random lookup lands on it, and checks that the sample still lists
|
||||||
|
// sampleLimit prefixes: the stored one, then the shuffled ones in order. The
|
||||||
|
// stored prefix is the second shuffled one, so the fill-up must skip it.
|
||||||
|
func TestSamplePrefixesByLengthFillsUpFromShuffled(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() }()
|
||||||
|
|
||||||
|
var shuffled []string
|
||||||
|
for i := range sampledPrefixCount {
|
||||||
|
shuffled = append(shuffled, fmt.Sprintf("10.0.%d.0/24", i))
|
||||||
|
}
|
||||||
|
route := mkV4Route(t, shuffled[1], 64500, time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC))
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*LiveRoute{route}); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
sample, err := db.samplePrefixesByLength(
|
||||||
|
context.Background(), liveRoutesV4Table, route.MaskLength, sampleLimit, shuffled)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("samplePrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
want := append([]string{shuffled[1], shuffled[0]}, shuffled[2:sampleLimit]...)
|
||||||
|
if !slices.Equal(sample, want) {
|
||||||
|
t.Errorf("samplePrefixesByLength = %v, want %v", sample, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestGetRandomPrefixesByLengthSkipsWithdrawnRoutes withdraws a route after
|
||||||
|
// storing it, as can happen between the steps of GetRandomPrefixesByLength. It
|
||||||
|
// checks that the random lookups pick nothing for a mask length with no route
|
||||||
|
// left, and that only the prefixes that still have a route get their newest
|
||||||
|
// route listed.
|
||||||
|
func TestGetRandomPrefixesByLengthSkipsWithdrawnRoutes(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)
|
||||||
|
kept := mkV4Route(t, "198.51.100.0/24", 64500, ts)
|
||||||
|
withdrawn := mkV4Route(t, "172.16.0.0/16", 64500, ts)
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*LiveRoute{kept, withdrawn}); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
if err := db.DeleteLiveRoute(withdrawn.Prefix, withdrawn.OriginASN, withdrawn.PeerIP); err != nil {
|
||||||
|
t.Fatalf("DeleteLiveRoute: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
sample, err := db.samplePrefixesByLength(
|
||||||
|
ctx, liveRoutesV4Table, withdrawn.MaskLength, sampleLimit, []string{withdrawn.Prefix})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("samplePrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
if len(sample) != 0 {
|
||||||
|
t.Errorf("samplePrefixesByLength = %v, want none for a mask length with no route left", sample)
|
||||||
|
}
|
||||||
|
|
||||||
|
routes, err := db.newestRouteOfEachPrefix(
|
||||||
|
ctx, liveRoutesV4Table, ipVersionV4, []string{withdrawn.Prefix, kept.Prefix})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("newestRouteOfEachPrefix: %v", err)
|
||||||
|
}
|
||||||
|
if len(routes) != 1 || routes[0].Prefix != kept.Prefix {
|
||||||
|
t.Errorf("newestRouteOfEachPrefix = %+v, want only the route of %s", routes, kept.Prefix)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -55,7 +55,7 @@ type Store interface {
|
|||||||
// Live route operations
|
// Live route operations
|
||||||
UpsertLiveRoute(route *LiveRoute) error
|
UpsertLiveRoute(route *LiveRoute) error
|
||||||
UpsertLiveRouteBatch(routes []*LiveRoute) error
|
UpsertLiveRouteBatch(routes []*LiveRoute) error
|
||||||
DeleteLiveRoute(prefix, peerIP string) error
|
DeleteLiveRoute(prefix string, originASN int, peerIP string) error
|
||||||
DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error
|
DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error
|
||||||
GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
|
GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
|
||||||
GetPrefixDistributionContext(ctx context.Context) (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
|
GetPrefixDistributionContext(ctx context.Context) (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
|
||||||
|
|||||||
@@ -114,6 +114,7 @@ type IPInfo struct {
|
|||||||
// LiveRouteDeletion represents parameters for deleting a live route
|
// LiveRouteDeletion represents parameters for deleting a live route
|
||||||
type LiveRouteDeletion struct {
|
type LiveRouteDeletion struct {
|
||||||
Prefix string
|
Prefix string
|
||||||
|
OriginASN int
|
||||||
PeerIP string
|
PeerIP string
|
||||||
IPVersion int
|
IPVersion int
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -107,7 +107,7 @@ CREATE TABLE IF NOT EXISTS live_routes_v4 (
|
|||||||
as_path TEXT NOT NULL, -- JSON array
|
as_path TEXT NOT NULL, -- JSON array
|
||||||
next_hop TEXT NOT NULL,
|
next_hop TEXT NOT NULL,
|
||||||
last_updated DATETIME NOT NULL,
|
last_updated DATETIME NOT NULL,
|
||||||
UNIQUE(prefix, peer_ip) -- a peer has at most one route for a prefix
|
UNIQUE(prefix, origin_asn, peer_ip)
|
||||||
);
|
);
|
||||||
|
|
||||||
-- IPv6 routing table maintained by PrefixHandler
|
-- 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
|
as_path TEXT NOT NULL, -- JSON array
|
||||||
next_hop TEXT NOT NULL,
|
next_hop TEXT NOT NULL,
|
||||||
last_updated DATETIME NOT NULL,
|
last_updated DATETIME NOT NULL,
|
||||||
UNIQUE(prefix, peer_ip) -- a peer has at most one route for a prefix
|
UNIQUE(prefix, origin_asn, peer_ip)
|
||||||
);
|
);
|
||||||
|
|
||||||
-- Indexes for live_routes_v4 table
|
-- Indexes for live_routes_v4 table
|
||||||
|
|||||||
@@ -205,7 +205,7 @@ func (m *mockStore) UpsertLiveRoute(route *database.LiveRoute) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// DeleteLiveRoute mock implementation
|
// DeleteLiveRoute mock implementation
|
||||||
func (m *mockStore) DeleteLiveRoute(prefix, peerIP string) error {
|
func (m *mockStore) DeleteLiveRoute(prefix string, originASN int, peerIP string) error {
|
||||||
// Simple mock - just return nil
|
// Simple mock - just return nil
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -108,14 +108,14 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
|||||||
// Withdrawals are queued before announcements, because the batch keeps
|
// Withdrawals are queued before announcements, because the batch keeps
|
||||||
// each peer's last update for a prefix: a message that lists a prefix as
|
// 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).
|
// 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 {
|
for _, prefix := range msg.Withdrawals {
|
||||||
h.batch = append(h.batch, prefixUpdate{
|
h.batch = append(h.batch, prefixUpdate{
|
||||||
prefix: canonicalPrefix(prefix),
|
prefix: canonicalPrefix(prefix),
|
||||||
|
originASN: originASN, // Use the originASN from path if available
|
||||||
peer: msg.Peer,
|
peer: msg.Peer,
|
||||||
messageType: "withdrawal",
|
messageType: "withdrawal",
|
||||||
timestamp: timestamp,
|
timestamp: timestamp,
|
||||||
|
path: msg.Path,
|
||||||
})
|
})
|
||||||
// Record withdrawal in metrics
|
// Record withdrawal in metrics
|
||||||
if h.metrics != nil {
|
if h.metrics != nil {
|
||||||
@@ -221,6 +221,7 @@ func (h *PrefixHandler) flushBatchLocked() {
|
|||||||
// Create deletion record for batch delete
|
// Create deletion record for batch delete
|
||||||
routesToDelete = append(routesToDelete, database.LiveRouteDeletion{
|
routesToDelete = append(routesToDelete, database.LiveRouteDeletion{
|
||||||
Prefix: update.prefix,
|
Prefix: update.prefix,
|
||||||
|
OriginASN: update.originASN,
|
||||||
PeerIP: update.peer,
|
PeerIP: update.peer,
|
||||||
IPVersion: ipVersion,
|
IPVersion: ipVersion,
|
||||||
})
|
})
|
||||||
@@ -424,24 +425,51 @@ func (h *PrefixHandler) processAnnouncementDirect(update prefixUpdate) {
|
|||||||
// processWithdrawalDirect handles removing a route directly without prefix table
|
// processWithdrawalDirect handles removing a route directly without prefix table
|
||||||
// nolint:unused // kept for potential future use
|
// nolint:unused // kept for potential future use
|
||||||
func (h *PrefixHandler) processWithdrawalDirect(update prefixUpdate) {
|
func (h *PrefixHandler) processWithdrawalDirect(update prefixUpdate) {
|
||||||
if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil {
|
// For withdrawals, we need to delete the route from live_routes
|
||||||
h.logger.Error("Failed to delete live route",
|
if update.originASN > 0 {
|
||||||
"prefix", update.prefix,
|
if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
|
||||||
"peer", update.peer,
|
h.logger.Error("Failed to delete live route",
|
||||||
"error", err,
|
"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,
|
||||||
|
)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// processWithdrawal handles removing a route from the live routing table
|
// processWithdrawal handles removing a route from the live routing table
|
||||||
// nolint:unused // kept for potential future use
|
// nolint:unused // kept for potential future use
|
||||||
func (h *PrefixHandler) processWithdrawal(_ *database.Prefix, update prefixUpdate) {
|
func (h *PrefixHandler) processWithdrawal(_ *database.Prefix, update prefixUpdate) {
|
||||||
if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil {
|
// For withdrawals, we need to delete the route from live_routes
|
||||||
h.logger.Error("Failed to delete live route",
|
// Since we have the origin ASN from the update, we can delete the specific route
|
||||||
"prefix", update.prefix,
|
if update.originASN > 0 {
|
||||||
"peer", update.peer,
|
if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
|
||||||
"error", err,
|
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,
|
||||||
|
)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -3,7 +3,6 @@ package routewatch
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
|
||||||
"slices"
|
"slices"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -183,155 +182,32 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
|
|||||||
t.Errorf("peers with a live route for %s = %v, want %v", prefix, gotPeers, tt.wantPeers)
|
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
|
var wantDistribution []database.PrefixDistribution
|
||||||
if len(tt.wantPeers) > 0 {
|
if len(tt.wantPeers) > 0 {
|
||||||
wantDistribution = []database.PrefixDistribution{{MaskLength: prefixMaskLength, Count: 1}}
|
wantDistribution = []database.PrefixDistribution{{MaskLength: prefixMaskLength, Count: 1}}
|
||||||
}
|
}
|
||||||
assertCounts(t, db, ipv4Version, len(tt.wantPeers), wantDistribution)
|
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)
|
||||||
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 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)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
Reference in New Issue
Block a user