Read the AS page from bounded lookups (closes #104)
check / check (push) Waiting to run

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
This commit is contained in:
2026-10-19 22:24:46 +00:00
parent 69d477159c
commit 5edfca1f53
3 changed files with 265 additions and 61 deletions
+64 -57
View File
@@ -1673,8 +1673,9 @@ func (d *Database) GetASDetails(asn int) (*ASN, []LiveRoute, error) {
return d.GetASDetailsContext(context.Background(), asn) return d.GetASDetailsContext(context.Background(), asn)
} }
// GetASDetailsContext returns detailed information about an AS including // GetASDetailsContext returns the AS and the prefixes it originates, the IPv4
// prefixes with context support // 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( func (d *Database) GetASDetailsContext(
ctx context.Context, asn int, ctx context.Context, asn int,
) (*ASN, []LiveRoute, error) { ) (*ASN, []LiveRoute, error) {
@@ -1701,27 +1702,12 @@ func (d *Database) GetASDetailsContext(
asnInfo.Handle = handle.String asnInfo.Handle = handle.String
asnInfo.Description = description.String asnInfo.Description = description.String
// Get prefixes announced by this AS from both tables allPrefixes, err := d.originPrefixes(ctx, asn, ipVersionV4)
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)
if err != nil { if err != nil {
return &asnInfo, nil, err return &asnInfo, nil, err
} }
queryV6 := ` prefixesV6, err := d.originPrefixes(ctx, asn, ipVersionV6)
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)
if err != nil { if err != nil {
return &asnInfo, allPrefixes, err return &asnInfo, allPrefixes, err
} }
@@ -2452,66 +2438,87 @@ func (d *Database) scanRouteTimestamp(
} }
} }
// originPrefixes runs query, which selects the prefix, mask length and latest // The AS page lists the prefixes an AS originates one at a time, each in two
// last_updated of every prefix asn originates in the live route table of // index searches that read no table row: one of the origin_asn and prefix
// ipVersion. Rows that fail to scan or whose timestamp does not parse are // index for the AS's first prefix after the previous one, and one of the
// logged and skipped. // 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( func (d *Database) originPrefixes(
ctx context.Context, query string, asn, ipVersion int, ctx context.Context, asn, ipVersion int,
) ([]LiveRoute, error) { ) ([]LiveRoute, error) {
family := ipv4Family family, nextSQL, newestSQL := ipv4Family, nextOriginPrefixV4SQL,
newestRouteTimeV4SQL
if ipVersion == ipVersionV6 { 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 { 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 var routes []LiveRoute
for rows.Next() { previous := "" // every prefix sorts after the empty string
var (
route LiveRoute for {
lastUpdatedStr string 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 { if err != nil {
d.logger.Error("Failed to scan "+family+" prefix row", return nil, fmt.Errorf("failed to list %s prefixes: %w", family, err)
"error", err, "asn", asn) }
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 continue
} }
// Parse the timestamp string
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05-07:00",
lastUpdatedStr)
if err != nil { if err != nil {
// Try without timezone return nil, fmt.Errorf("failed to query route time of %s: %w",
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05", prefix, err)
lastUpdatedStr) }
if err != nil {
d.logger.Error("Failed to parse timestamp",
"error", err, "timestamp", lastUpdatedStr)
continue route.MaskLength, err = prefixMaskLength(prefix)
} if err != nil {
return nil, err
} }
route.OriginASN = asn
route.IPVersion = ipVersion
routes = append(routes, route) 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 // firstPrefixesByLength returns the first n prefixes, in text order, with the
+196
View File
@@ -10,6 +10,7 @@ import (
"net/netip" "net/netip"
"slices" "slices"
"strconv" "strconv"
"strings"
"sync" "sync"
"testing" "testing"
"time" "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 // TestUpdatePrefixesBatchKeepsFirstSeen writes a prefix, then a batch with it
// and a new prefix. Only the new prefix is inserted and counted, and the known // and a new prefix. Only the new prefix is inserted and counted, and the known
// one keeps its first_seen time. // one keeps its first_seen time.
@@ -1518,3 +1651,66 @@ func writeBenchBatch(
b.Fatalf("UpdatePrefixesBatch: %v", err) 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)
}
}
}
+5 -4
View File
@@ -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 -- 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 -- 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 -- 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_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); 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 -- 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. -- 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_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); 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. -- origin_asn, prefix: the prefixes an AS originates, each in one search
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_asn ON live_routes_v4(origin_asn); -- whatever the number of peers, for the AS page and its API answer.
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_origin_asn ON live_routes_v6(origin_asn); 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 -- last_updated: the oldest and newest route times in /api/v1/stats and
-- /status.json. -- /status.json.
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated); CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated);