From 5edfca1f535b6eeefe39b1c6eb1d819a59abc8b5 Mon Sep 17 00:00:00 2001 From: sneak Date: Mon, 19 Oct 2026 22:24:46 +0000 Subject: [PATCH] Read the AS page from bounded lookups (closes #104) The AS page and /api/v1/as/{asn} grouped every route the AS originates by prefix, reading one table row per peer and prefix and sorting them all. They now take the AS's prefixes one at a time: one search of a new index on origin_asn and prefix, which replaces the origin_asn index and is written exactly when it was, and one search of the prefix and last_updated index for the time of the prefix's newest route. Neither reads the table. For a prefix more than one AS originates, that time is now the newest route's from any origin. A hand-run benchmark reads an AS of 5,000 prefixes from 500 peers. Model: opus-5-5 --- internal/database/database.go | 121 ++++++------ internal/database/database_internal_test.go | 196 ++++++++++++++++++++ internal/database/schema.sql | 9 +- 3 files changed, 265 insertions(+), 61 deletions(-) diff --git a/internal/database/database.go b/internal/database/database.go index 3d4e89e..279940d 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -1673,8 +1673,9 @@ func (d *Database) GetASDetails(asn int) (*ASN, []LiveRoute, error) { return d.GetASDetailsContext(context.Background(), asn) } -// GetASDetailsContext returns detailed information about an AS including -// prefixes with context support +// GetASDetailsContext returns the AS and the prefixes it originates, the IPv4 +// ones first, each once with the time of its newest route. It makes two index +// searches per prefix, however many peers carry the prefix. func (d *Database) GetASDetailsContext( ctx context.Context, asn int, ) (*ASN, []LiveRoute, error) { @@ -1701,27 +1702,12 @@ func (d *Database) GetASDetailsContext( asnInfo.Handle = handle.String asnInfo.Description = description.String - // Get prefixes announced by this AS from both tables - queryV4 := ` - SELECT prefix, mask_length, MAX(last_updated) as last_updated - FROM live_routes_v4 - WHERE origin_asn = ? - GROUP BY prefix, mask_length - ` - - allPrefixes, err := d.originPrefixes(ctx, queryV4, asn, ipVersionV4) + allPrefixes, err := d.originPrefixes(ctx, asn, ipVersionV4) if err != nil { return &asnInfo, nil, err } - queryV6 := ` - SELECT prefix, mask_length, MAX(last_updated) as last_updated - FROM live_routes_v6 - WHERE origin_asn = ? - GROUP BY prefix, mask_length - ` - - prefixesV6, err := d.originPrefixes(ctx, queryV6, asn, ipVersionV6) + prefixesV6, err := d.originPrefixes(ctx, asn, ipVersionV6) if err != nil { return &asnInfo, allPrefixes, err } @@ -2452,66 +2438,87 @@ func (d *Database) scanRouteTimestamp( } } -// originPrefixes runs query, which selects the prefix, mask length and latest -// last_updated of every prefix asn originates in the live route table of -// ipVersion. Rows that fail to scan or whose timestamp does not parse are -// logged and skipped. +// The AS page lists the prefixes an AS originates one at a time, each in two +// index searches that read no table row: one of the origin_asn and prefix +// index for the AS's first prefix after the previous one, and one of the +// prefix and last_updated index for the time of its newest route. Neither +// reads more for a prefix that more peers carry. +const ( + nextOriginPrefixV4SQL = `SELECT prefix FROM live_routes_v4 + WHERE origin_asn = ? AND prefix > ? ORDER BY prefix LIMIT 1` + newestRouteTimeV4SQL = `SELECT last_updated FROM live_routes_v4 + WHERE prefix = ? ORDER BY last_updated DESC LIMIT 1` + + nextOriginPrefixV6SQL = `SELECT prefix FROM live_routes_v6 + WHERE origin_asn = ? AND prefix > ? ORDER BY prefix LIMIT 1` + newestRouteTimeV6SQL = `SELECT last_updated FROM live_routes_v6 + WHERE prefix = ? ORDER BY last_updated DESC LIMIT 1` +) + +// originPrefixes returns every prefix asn originates in the live route table +// of ipVersion, in text order, each with its mask length and the time of the +// prefix's newest route, whichever AS that route names as origin. A prefix +// whose routes were all withdrawn between the two searches is left out. func (d *Database) originPrefixes( - ctx context.Context, query string, asn, ipVersion int, + ctx context.Context, asn, ipVersion int, ) ([]LiveRoute, error) { - family := ipv4Family + family, nextSQL, newestSQL := ipv4Family, nextOriginPrefixV4SQL, + newestRouteTimeV4SQL if ipVersion == ipVersionV6 { - family = ipv6Family + family, nextSQL, newestSQL = ipv6Family, nextOriginPrefixV6SQL, + newestRouteTimeV6SQL } - rows, err := d.db.QueryContext(ctx, query, asn) + nextStmt, err := d.db.PrepareContext(ctx, nextSQL) if err != nil { - return nil, fmt.Errorf("failed to query %s prefixes: %w", family, err) + return nil, fmt.Errorf("failed to prepare %s prefix query: %w", + family, err) } - defer func() { _ = rows.Close() }() + defer func() { _ = nextStmt.Close() }() + + newestStmt, err := d.db.PrepareContext(ctx, newestSQL) + if err != nil { + return nil, fmt.Errorf("failed to prepare %s route time query: %w", + family, err) + } + defer func() { _ = newestStmt.Close() }() var routes []LiveRoute - for rows.Next() { - var ( - route LiveRoute - lastUpdatedStr string - ) + previous := "" // every prefix sorts after the empty string + + for { + var prefix string + + err := nextStmt.QueryRowContext(ctx, asn, previous).Scan(&prefix) + if errors.Is(err, sql.ErrNoRows) { + return routes, nil + } - err := rows.Scan(&route.Prefix, &route.MaskLength, &lastUpdatedStr) if err != nil { - d.logger.Error("Failed to scan "+family+" prefix row", - "error", err, "asn", asn) + return nil, fmt.Errorf("failed to list %s prefixes: %w", family, err) + } + previous = prefix + route := LiveRoute{Prefix: prefix, IPVersion: ipVersion, OriginASN: asn} + + err = newestStmt.QueryRowContext(ctx, prefix).Scan(&route.LastUpdated) + if errors.Is(err, sql.ErrNoRows) { continue } - // Parse the timestamp string - route.LastUpdated, err = time.Parse("2006-01-02 15:04:05-07:00", - lastUpdatedStr) if err != nil { - // Try without timezone - route.LastUpdated, err = time.Parse("2006-01-02 15:04:05", - lastUpdatedStr) - if err != nil { - d.logger.Error("Failed to parse timestamp", - "error", err, "timestamp", lastUpdatedStr) + return nil, fmt.Errorf("failed to query route time of %s: %w", + prefix, err) + } - continue - } + route.MaskLength, err = prefixMaskLength(prefix) + if err != nil { + return nil, err } - route.OriginASN = asn - route.IPVersion = ipVersion routes = append(routes, route) } - - err = rows.Err() - if err != nil { - return nil, fmt.Errorf("failed to read %s prefixes: %w", family, err) - } - - return routes, nil } // firstPrefixesByLength returns the first n prefixes, in text order, with the diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index 7a48559..41d2c22 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -10,6 +10,7 @@ import ( "net/netip" "slices" "strconv" + "strings" "sync" "testing" "time" @@ -1214,6 +1215,138 @@ func TestLiveRouteUpdateSetsOriginWhenItChanges(t *testing.T) { } } +// TestGetASDetailsListsEachPrefixOnce stores two IPv4 prefixes and an IPv6 +// prefix an AS originates, two of them from two peers at different times, and +// a prefix another AS originates. The AS details list each of the AS's +// prefixes once, IPv4 first and in text order, with its mask length and the +// time of its newest route. +func TestGetASDetailsListsEachPrefixOnce(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() }() + + const asn, otherASN = 64500, 64501 + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + _, err = db.GetOrCreateASN(asn, ts) + if err != nil { + t.Fatalf("GetOrCreateASN: %v", err) + } + + v4Older := mkV4Route(t, "192.0.2.0/24", asn, ts) + v4Newer := mkV4Route(t, "192.0.2.0/24", asn, ts.Add(time.Minute)) + v4Newer.PeerIP = secondV4PeerIP + v4Single := mkV4Route(t, "198.51.100.128/25", asn, ts) + v6Newer := mkV6Route(testV6Prefix, asn, ts.Add(time.Minute)) + v6Older := mkV6Route(testV6Prefix, asn, ts) + v6Older.PeerIP = secondV6PeerIP + other := mkV4Route(t, "203.0.113.0/24", otherASN, ts) + + err = db.UpsertLiveRouteBatch([]*LiveRoute{ + v4Older, v4Newer, v4Single, v6Newer, v6Older, other, + }) + if err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + _, prefixes, err := db.GetASDetails(asn) + if err != nil { + t.Fatalf("GetASDetails: %v", err) + } + + want := []*LiveRoute{v4Newer, v4Single, v6Newer} + + listedOnce := slices.EqualFunc(prefixes, want, + func(got LiveRoute, want *LiveRoute) bool { + return got.Prefix == want.Prefix && + got.MaskLength == want.MaskLength && + got.IPVersion == want.IPVersion && + got.OriginASN == asn && + got.LastUpdated.Equal(want.LastUpdated) + }) + if !listedOnce { + t.Errorf("GetASDetails listed %+v, want the prefixes and times of %+v", + prefixes, want) + } +} + +// TestASPrefixQueriesUseCoveringIndexes checks the plan of each query that +// lists the prefixes an AS originates: a search of a covering index and no +// temporary B-tree, so the list reads one index entry per prefix, not one +// table row per route. Grouping the AS's routes by prefix read every route's +// row and sorted them in a temporary B-tree +// (https://git.eeqj.de/sneak/routewatch/issues/104). +func TestASPrefixQueriesUseCoveringIndexes(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() }() + + tests := []struct { + query string + args []any + index string + }{ + {nextOriginPrefixV4SQL, []any{0, ""}, "idx_live_routes_v4_origin_prefix"}, + {newestRouteTimeV4SQL, []any{""}, "idx_live_routes_v4_prefix_last_updated"}, + {nextOriginPrefixV6SQL, []any{0, ""}, "idx_live_routes_v6_origin_prefix"}, + {newestRouteTimeV6SQL, []any{""}, "idx_live_routes_v6_prefix_last_updated"}, + } + + for _, tt := range tests { + plan := queryPlan(t, db, tt.query, tt.args...) + + if !strings.Contains(plan, "USING COVERING INDEX "+tt.index+" ") || + strings.Contains(plan, "TEMP B-TREE") { + t.Errorf("plan of %s:\n%s\nwant a search of covering index %s "+ + "and no temporary B-tree", tt.query, plan, tt.index) + } + } +} + +// queryPlan returns the steps of EXPLAIN QUERY PLAN for query, one per line. +func queryPlan(t *testing.T, db *Database, query string, args ...any) string { + t.Helper() + + rows, err := db.db.QueryContext(t.Context(), "EXPLAIN QUERY PLAN "+query, + args...) + if err != nil { + t.Fatalf("EXPLAIN QUERY PLAN %s: %v", query, err) + } + defer func() { _ = rows.Close() }() + + var steps []string + + for rows.Next() { + var ( + id, parent, unused int + detail string + ) + + err := rows.Scan(&id, &parent, &unused, &detail) + if err != nil { + t.Fatalf("reading the plan of %s: %v", query, err) + } + + steps = append(steps, detail) + } + + err = rows.Err() + if err != nil { + t.Fatalf("reading the plan of %s: %v", query, err) + } + + return strings.Join(steps, "\n") +} + // TestUpdatePrefixesBatchKeepsFirstSeen writes a prefix, then a batch with it // and a new prefix. Only the new prefix is inserted and counted, and the known // one keeps its first_seen time. @@ -1518,3 +1651,66 @@ func writeBenchBatch( b.Fatalf("UpdatePrefixesBatch: %v", err) } } + +// Parameters for BenchmarkGetASDetailsWithFiveHundredPeers: the AS originates +// benchASPrefixes prefixes and every one of benchASPeers peers has a route for +// each, 2.5 million routes in all. +const ( + benchASPrefixes = 5000 + benchASPeers = 500 +) + +// BenchmarkGetASDetailsWithFiveHundredPeers times the read behind the AS page +// for an AS that originates 5,000 IPv4 prefixes, each from 500 peers. It +// builds that database first, one peer at a time, which takes some minutes and +// about a GB of disk under TMPDIR. It is run by hand, not by make test, for +// example in the test image: +// +// go test -run '^$' -bench GetASDetails -benchtime 10x ./internal/database/ +func BenchmarkGetASDetailsWithFiveHundredPeers(b *testing.B) { + db, err := New(&config.Config{StateDir: b.TempDir()}, logger.New()) + if err != nil { + b.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + _, err = db.GetOrCreateASN(benchFirstASN, ts) + if err != nil { + b.Fatalf("GetOrCreateASN: %v", err) + } + + for peer := range benchASPeers { + peerIP := fmt.Sprintf("198.18.%d.%d", peer/256, peer%256) + + routes := make([]*LiveRoute, 0, benchASPrefixes) + for prefix := range benchASPrefixes { + routes = append(routes, &LiveRoute{ + Prefix: fmt.Sprintf("%d.%d.%d.0/%d", + 1+prefix>>16, (prefix>>8)%256, prefix%256, benchMaskLength), + MaskLength: benchMaskLength, + IPVersion: ipVersionV4, + OriginASN: benchFirstASN, + PeerIP: peerIP, + ASPath: []int{benchFirstASN + 1 + peer, benchFirstASN}, + NextHop: peerIP, + LastUpdated: ts, + }) + } + + writeBenchBatch(b, db, routes, nil, ts) + } + + for b.Loop() { + _, prefixes, err := db.GetASDetails(benchFirstASN) + if err != nil { + b.Fatalf("GetASDetails: %v", err) + } + + if len(prefixes) != benchASPrefixes { + b.Fatalf("GetASDetails listed %d prefixes, want %d", + len(prefixes), benchASPrefixes) + } + } +} diff --git a/internal/database/schema.sql b/internal/database/schema.sql index 6bd65a6..5c0ca22 100644 --- a/internal/database/schema.sql +++ b/internal/database/schema.sql @@ -128,7 +128,7 @@ CREATE TABLE IF NOT EXISTS live_routes_v6 ( -- prefix, last_updated: a prefix's routes, for the prefix page and its API -- answer, the IP lookup and its peer count, and the check whether a prefix -- still has a live route; and a prefix's newest route for the prefix-length --- pages, in one search whatever the number of peers. +-- pages and the AS page, in one search whatever the number of peers. CREATE INDEX IF NOT EXISTS idx_live_routes_v4_prefix_last_updated ON live_routes_v4(prefix, last_updated); CREATE INDEX IF NOT EXISTS idx_live_routes_v6_prefix_last_updated ON live_routes_v6(prefix, last_updated); -- mask_length: the prefix-length pages' random sample, the row id range of a @@ -139,9 +139,10 @@ CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_length ON live_routes_v6(mask -- prefixes in text order, and the prefix distribution read at startup. CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_prefix ON live_routes_v4(mask_length, prefix); CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_prefix ON live_routes_v6(mask_length, prefix); --- origin_asn: the prefixes an AS originates, for the AS page and its API answer. -CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_asn ON live_routes_v4(origin_asn); -CREATE INDEX IF NOT EXISTS idx_live_routes_v6_origin_asn ON live_routes_v6(origin_asn); +-- origin_asn, prefix: the prefixes an AS originates, each in one search +-- whatever the number of peers, for the AS page and its API answer. +CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_prefix ON live_routes_v4(origin_asn, prefix); +CREATE INDEX IF NOT EXISTS idx_live_routes_v6_origin_prefix ON live_routes_v6(origin_asn, prefix); -- last_updated: the oldest and newest route times in /api/v1/stats and -- /status.json. CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated);