1 Commits
Author SHA1 Message Date
sneak 74baf66dde Serve the /api/v1/stats prefix distribution from memory (closes #30)
check / check (push) Successful in 2m54s
The prefix distribution was the last query on the stats path. It read one
index entry per live route on every request, so on a large database it took
the whole 4-second deadline and /api/v1/stats answered 500.

The distribution now lives next to the in-memory counts: seeded once at
startup from the same query, then kept exact by every live-route write. A new
route whose prefix had no live route adds one at its mask length; a delete
that leaves a prefix with no live route takes one away. Each check is one
lookup on the prefix index, made only for new and removed routes, within the
same write. The single-route delete now runs its delete and that lookup in one
transaction, so a failed lookup cannot leave the counts off.

Model: opus-5-5
2026-10-03 12:54:12 +00:00
12 changed files with 726 additions and 385 deletions
+2 -5
View File
@@ -142,10 +142,7 @@ routewatch/
- **Backpressure**: Probabilistic message dropping when queues exceed 50% capacity - **Backpressure**: Probabilistic message dropping when queues exceed 50% capacity
- **Graceful Shutdown**: 60-second timeout, flushes all pending batches - **Graceful Shutdown**: 60-second timeout, flushes all pending batches
- **Reconnection**: Exponential backoff (5s-320s) with reset after 30s of stable connection - **Reconnection**: Exponential backoff (5s-320s) with reset after 30s of stable connection
- **IP Lookup**: the most specific live route is found by looking up the - **IPv4 Optimization**: IP ranges stored as uint32 for O(1) lookups
address's prefix at each mask length, longest first, in the prefix index
(at most 33 lookups for IPv4, 129 for IPv6); prefixes are stored in the
text form Go's `net/netip` prints
### Database Schema ### Database Schema
@@ -157,7 +154,7 @@ prefixes_v6(id, prefix, mask_length, first_seen, last_seen)
-- Live routing tables (one per IP version) -- Live routing tables (one per IP version)
live_routes_v4(id, prefix, mask_length, origin_asn, peer_ip, as_path, live_routes_v4(id, prefix, mask_length, origin_asn, peer_ip, as_path,
next_hop, last_updated) next_hop, last_updated, v4_ip_start, v4_ip_end)
live_routes_v6(id, prefix, mask_length, origin_asn, peer_ip, as_path, live_routes_v6(id, prefix, mask_length, origin_asn, peer_ip, as_path,
next_hop, last_updated) next_hop, last_updated)
-10
View File
@@ -30,16 +30,6 @@ real 5 GiB limit (https://git.eeqj.de/sneak/routewatch/issues/3).
- 2026-10-03: `/api/v1/stats` serves the prefix distribution from memory, - 2026-10-03: `/api/v1/stats` serves the prefix distribution from memory,
seeded at startup and adjusted on every live-route write, so a request no seeded at startup and adjusted on every live-route write, so a request no
longer reads every live route (closes #30) longer reads every live route (closes #30)
- 2026-10-03: a new database is created with `auto_vacuum` set to
incremental, through the connection string so it is set before the file
is first written, and each periodic incremental vacuum now returns up to
1000 free pages; the late `PRAGMA auto_vacuum` in `Initialize`, which
SQLite ignored, is gone (closes #43)
- 2026-10-03: looking up an IP address no longer reads every IPv6 route:
both families find the most specific live route with at most 33 or 129
lookups on the prefix index, and the IPv4 range columns are gone. Prefixes
from the feed are stored in one text form, so an IPv6 withdrawal, which the
feed sends uncompressed, now removes its route (closes #48)
- 2026-10-02: a plain `docker build .` stamps the commit's tag or short - 2026-10-02: a plain `docker build .` stamps the commit's tag or short
commit (`git describe --tags --always`) into the page footer instead of commit (`git describe --tags --always`) into the page footer instead of
`unknown`: `.dockerignore` sends `.git` without `.git/config`, a `VERSION` `unknown`: `.dockerignore` sends `.git` without `.git/config`, a `VERSION`
+15 -25
View File
@@ -12,11 +12,15 @@ import (
"github.com/google/uuid" "github.com/google/uuid"
) )
// mkV4Route builds an IPv4 live route with its mask length taken from the // mkV4Route builds an IPv4 live route with its mask length and range columns
// prefix. // taken from the prefix.
func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute { func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute {
t.Helper() t.Helper()
start, end, err := CalculateIPv4Range(prefix)
if err != nil {
t.Fatalf("CalculateIPv4Range(%s): %v", prefix, err)
}
maskLength, err := prefixMaskLength(prefix) maskLength, err := prefixMaskLength(prefix)
if err != nil { if err != nil {
t.Fatalf("prefixMaskLength(%s): %v", prefix, err) t.Fatalf("prefixMaskLength(%s): %v", prefix, err)
@@ -32,6 +36,8 @@ func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute {
ASPath: []int{asn}, ASPath: []int{asn},
NextHop: "192.0.2.254", NextHop: "192.0.2.254",
LastUpdated: ts, LastUpdated: ts,
V4IPStart: &start,
V4IPEnd: &end,
} }
} }
@@ -203,8 +209,6 @@ func TestPrefixDistributionTracksWrites(t *testing.T) {
other := mkV4Route(t, "203.0.113.0/24", 64501, ts) other := mkV4Route(t, "203.0.113.0/24", 64501, ts)
wide := mkV4Route(t, "172.16.0.0/16", 64502, ts) wide := mkV4Route(t, "172.16.0.0/16", 64502, ts)
v6 := mkV6Route("2001:db8::/32", 64503, ts) v6 := mkV6Route("2001:db8::/32", 64503, ts)
v6SecondPeer := mkV6Route("2001:db8::/32", 64503, ts)
v6SecondPeer.PeerIP = "2001:db8::2"
all := []PrefixDistribution{{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 2}} all := []PrefixDistribution{{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 2}}
v6Only := []PrefixDistribution{{MaskLength: 32, Count: 1}} v6Only := []PrefixDistribution{{MaskLength: 32, Count: 1}}
@@ -228,46 +232,32 @@ func TestPrefixDistributionTracksWrites(t *testing.T) {
}, },
{ {
name: "second peer announces a prefix that has a live route", name: "second peer announces a prefix that has a live route",
announce: []*LiveRoute{sharedSecondPeer, v6SecondPeer}, announce: []*LiveRoute{sharedSecondPeer},
wantV4: all, wantV6: v6Only, wantV4: all, wantV6: v6Only,
}, },
{ {
name: "withdrawal of a route that is not the last for its prefix", name: "withdrawal of a route that is not the last for its prefix",
withdraw: []*LiveRoute{shared, v6}, withdraw: []*LiveRoute{shared},
wantV4: all, wantV6: v6Only, wantV4: all, wantV6: v6Only,
}, },
{ {
name: "withdrawal of the last route for a prefix", name: "withdrawal of the last route for a prefix",
withdraw: []*LiveRoute{sharedSecondPeer, v6SecondPeer}, withdraw: []*LiveRoute{sharedSecondPeer},
wantV4: []PrefixDistribution{{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 1}}, wantV4: []PrefixDistribution{{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 1}},
wantV6: v6Only,
}, },
{ {
name: "withdrawal of every remaining route", name: "withdrawal of every remaining route",
withdraw: []*LiveRoute{other, wide}, withdraw: []*LiveRoute{other, wide, v6},
}, },
{ {
name: "two peers announce a new prefix together", name: "two peers announce a new prefix together",
announce: []*LiveRoute{shared, sharedSecondPeer, v6, v6SecondPeer}, announce: []*LiveRoute{shared, sharedSecondPeer},
wantV4: []PrefixDistribution{{MaskLength: 24, Count: 1}}, wantV4: []PrefixDistribution{{MaskLength: 24, Count: 1}},
wantV6: v6Only,
}, },
{ {
name: "both routes for a prefix withdrawn together", name: "both routes for a prefix withdrawn together",
withdraw: []*LiveRoute{shared, sharedSecondPeer, v6, v6SecondPeer}, withdraw: []*LiveRoute{shared, sharedSecondPeer},
},
// The feed often withdraws a route that is not live. That must not take
// the prefix out of the distribution. A count wrongly taken below zero is
// left out of the answer, so the next step shows it: its announcement
// at the same mask length would then not be counted.
{
name: "withdrawal of routes that are not live",
withdraw: []*LiveRoute{other, v6},
},
{
name: "announcement after a withdrawal of routes that are not live",
announce: []*LiveRoute{other, v6},
wantV4: []PrefixDistribution{{MaskLength: 24, Count: 1}},
wantV6: v6Only,
}, },
} }
+335 -92
View File
@@ -9,7 +9,6 @@ import (
"errors" "errors"
"fmt" "fmt"
"net" "net"
"net/netip"
"os" "os"
"path/filepath" "path/filepath"
"runtime" "runtime"
@@ -33,8 +32,11 @@ const (
dirPermissions = 0750 // rwxr-x--- dirPermissions = 0750 // rwxr-x---
ipVersionV4 = 4 ipVersionV4 = 4
ipVersionV6 = 6 ipVersionV6 = 6
ipv6Length = 16
ipv4Offset = 12
ipv4Bits = 32 ipv4Bits = 32
ipv6Bits = 128 ipv6Bits = 128
maxIPv4 = 0xFFFFFFFF
) )
// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every // SQLite memory tuning. cache_size and busy_timeout go in the DSN so every
@@ -93,13 +95,8 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
// a transaction that reads before writing starts as a reader and, when it // a transaction that reads before writing starts as a reader and, when it
// then writes while another connection holds the write lock, fails at once // then writes while another connection holds the write lock, fails at once
// with "database is locked" without waiting for _busy_timeout. // with "database is locked" without waiting for _busy_timeout.
// _auto_vacuum=incremental lets Vacuum return free pages to the filesystem.
// SQLite only accepts it before the database file is first written, and the
// switch to WAL writes it, so it must be here: the driver applies it on open,
// before _journal_mode. On an existing file it changes nothing.
dsn := fmt.Sprintf( dsn := fmt.Sprintf(
"file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d"+ "file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d&_journal_mode=WAL&_txlock=immediate",
"&_auto_vacuum=incremental&_journal_mode=WAL&_txlock=immediate",
dbPath, dbPath,
sqliteCacheSizeKiB, sqliteCacheSizeKiB,
sqliteBusyTimeoutMs, sqliteBusyTimeoutMs,
@@ -144,8 +141,9 @@ func (d *Database) Initialize() error {
// heap. The heap limits below are process-wide, so setting them once here is // heap. The heap limits below are process-wide, so setting them once here is
// enough for the whole pool. // enough for the whole pool.
pragmas := []string{ pragmas := []string{
"PRAGMA journal_mode=WAL", // Write-Ahead Logging "PRAGMA journal_mode=WAL", // Write-Ahead Logging
"PRAGMA analysis_limit=0", // Disable automatic ANALYZE "PRAGMA analysis_limit=0", // Disable automatic ANALYZE
"PRAGMA auto_vacuum=INCREMENTAL", // Enable incremental vacuum
fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes), fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes),
fmt.Sprintf("PRAGMA hard_heap_limit=%d", sqliteHardHeapLimitBytes), fmt.Sprintf("PRAGMA hard_heap_limit=%d", sqliteHardHeapLimitBytes),
} }
@@ -228,10 +226,10 @@ func (d *Database) beginTx() (*loggingTx, error) {
// upsert did. // upsert did.
const ( const (
updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?, updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?,
last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?` last_updated = ?, ip_start = ?, ip_end = ? 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, ip_start, ip_end) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?, updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?,
last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?` last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
@@ -253,8 +251,12 @@ 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) {
if route.V4IPStart == nil || route.V4IPEnd == nil {
return false, false, fmt.Errorf("IPv4 route %s missing range values", route.Prefix)
}
res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated, res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
route.Prefix, route.OriginASN, route.PeerIP) *route.V4IPStart, *route.V4IPEnd, route.Prefix, route.OriginASN, route.PeerIP)
if err != nil { if err != nil {
return false, false, err return false, false, err
} }
@@ -272,7 +274,7 @@ func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string
} }
_, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN, _, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN,
route.PeerIP, pathJSON, route.NextHop, route.LastUpdated) route.PeerIP, pathJSON, route.NextHop, route.LastUpdated, *route.V4IPStart, *route.V4IPEnd)
if err != nil { if err != nil {
return false, false, err return false, false, err
} }
@@ -1395,10 +1397,6 @@ func (d *Database) GetPrefixDistributionContext(ctx context.Context) (
} }
ipv4 = append(ipv4, dist) ipv4 = append(ipv4, dist)
} }
// A read that stops partway ends the loop without an error of its own.
if err := rows4.Err(); err != nil {
return nil, nil, fmt.Errorf("failed to read IPv4 distribution: %w", err)
}
// IPv6 distribution - count unique prefixes from v6 table // IPv6 distribution - count unique prefixes from v6 table
query = ` query = `
@@ -1424,9 +1422,6 @@ func (d *Database) GetPrefixDistributionContext(ctx context.Context) (
} }
ipv6 = append(ipv6, dist) ipv6 = append(ipv6, dist)
} }
if err := rows6.Err(); err != nil {
return nil, nil, fmt.Errorf("failed to read IPv6 distribution: %w", err)
}
return ipv4, ipv6, nil return ipv4, ipv6, nil
} }
@@ -1453,6 +1448,176 @@ func (d *Database) GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ip
return ipv4Count, ipv6Count, nil return ipv4Count, ipv6Count, nil
} }
// GetASInfoForIP returns AS information for the given IP address
func (d *Database) GetASInfoForIP(ip string) (*ASInfo, error) {
return d.GetASInfoForIPContext(context.Background(), ip)
}
// GetASInfoForIPContext returns AS information for the given IP address with context support
func (d *Database) GetASInfoForIPContext(ctx context.Context, ip string) (*ASInfo, error) {
// Parse the IP to validate it
parsedIP := net.ParseIP(ip)
if parsedIP == nil {
return nil, fmt.Errorf("%w: %s", ErrInvalidIP, ip)
}
// Determine IP version
ipVersion := ipVersionV4
ipv4 := parsedIP.To4()
if ipv4 == nil {
ipVersion = ipVersionV6
}
// For IPv4, use optimized range query
if ipVersion == ipVersionV4 {
// Convert IP to 32-bit unsigned integer
ipUint := ipToUint32(ipv4)
query := `
SELECT DISTINCT lr.prefix, lr.mask_length, lr.origin_asn, lr.last_updated, a.handle, a.description
FROM live_routes_v4 lr
LEFT JOIN asns a ON a.asn = lr.origin_asn
WHERE lr.ip_start <= ? AND lr.ip_end >= ?
ORDER BY lr.mask_length DESC
LIMIT 1
`
var prefix string
var maskLength, originASN int
var lastUpdated time.Time
var handle, description sql.NullString
err := d.db.QueryRowContext(ctx, query, ipUint, ipUint).Scan(
&prefix, &maskLength, &originASN, &lastUpdated, &handle, &description)
if err != nil {
if err == sql.ErrNoRows {
return nil, fmt.Errorf("%w for IP %s", ErrNoRoute, ip)
}
return nil, fmt.Errorf("failed to query routes: %w", err)
}
age := time.Since(lastUpdated).Round(time.Second).String()
return &ASInfo{
ASN: originASN,
Handle: handle.String,
Description: description.String,
Prefix: prefix,
LastUpdated: lastUpdated,
Age: age,
}, nil
}
// For IPv6, use the original method since we don't have range optimization
query := `
SELECT DISTINCT lr.prefix, lr.mask_length, lr.origin_asn, lr.last_updated, a.handle, a.description
FROM live_routes_v6 lr
LEFT JOIN asns a ON a.asn = lr.origin_asn
ORDER BY lr.mask_length DESC
`
rows, err := d.db.QueryContext(ctx, query)
if err != nil {
return nil, fmt.Errorf("failed to query routes: %w", err)
}
defer func() { _ = rows.Close() }()
// Find the most specific matching prefix
var bestMatch struct {
prefix string
maskLength int
originASN int
lastUpdated time.Time
handle sql.NullString
description sql.NullString
}
bestMaskLength := -1
for rows.Next() {
var prefix string
var maskLength, originASN int
var lastUpdated time.Time
var handle, description sql.NullString
if err := rows.Scan(&prefix, &maskLength, &originASN, &lastUpdated, &handle, &description); err != nil {
continue
}
// Parse the prefix CIDR
_, ipNet, err := net.ParseCIDR(prefix)
if err != nil {
continue
}
// Check if the IP is in this prefix
if ipNet.Contains(parsedIP) && maskLength > bestMaskLength {
bestMatch.prefix = prefix
bestMatch.maskLength = maskLength
bestMatch.originASN = originASN
bestMatch.lastUpdated = lastUpdated
bestMatch.handle = handle
bestMatch.description = description
bestMaskLength = maskLength
}
}
if bestMaskLength == -1 {
return nil, fmt.Errorf("%w for IP %s", ErrNoRoute, ip)
}
age := time.Since(bestMatch.lastUpdated).Round(time.Second).String()
return &ASInfo{
ASN: bestMatch.originASN,
Handle: bestMatch.handle.String,
Description: bestMatch.description.String,
Prefix: bestMatch.prefix,
LastUpdated: bestMatch.lastUpdated,
Age: age,
}, nil
}
// ipToUint32 converts an IPv4 address to a 32-bit unsigned integer
func ipToUint32(ip net.IP) uint32 {
if len(ip) == ipv6Length {
// Convert to 4-byte representation
ip = ip[ipv4Offset:ipv6Length]
}
return uint32(ip[0])<<24 | uint32(ip[1])<<16 | uint32(ip[2])<<8 | uint32(ip[3])
}
// CalculateIPv4Range calculates the start and end IP addresses for an IPv4 CIDR block
func CalculateIPv4Range(cidr string) (start, end uint32, err error) {
_, ipNet, err := net.ParseCIDR(cidr)
if err != nil {
return 0, 0, err
}
// Get the network address (start of range)
ip := ipNet.IP.To4()
if ip == nil {
return 0, 0, fmt.Errorf("not an IPv4 address")
}
start = ipToUint32(ip)
// Calculate the end of the range
ones, bits := ipNet.Mask.Size()
hostBits := bits - ones
if hostBits >= ipv4Bits {
// Special case for /0 - entire IPv4 space
end = maxIPv4
} else {
// Safe to convert since we checked hostBits < 32
//nolint:gosec // hostBits is guaranteed to be < 32 from the check above
end = start | ((1 << uint(hostBits)) - 1)
}
return start, end, nil
}
// GetASDetails returns detailed information about an AS including prefixes // GetASDetails returns detailed information about an AS including prefixes
func (d *Database) GetASDetails(asn int) (*ASN, []LiveRoute, error) { func (d *Database) GetASDetails(asn int) (*ASN, []LiveRoute, error) {
return d.GetASDetailsContext(context.Background(), asn) return d.GetASDetailsContext(context.Background(), asn)
@@ -1874,37 +2039,39 @@ func (d *Database) GetIPInfo(ip string) (*IPInfo, error) {
} }
// GetIPInfoContext returns comprehensive IP information with context support. // GetIPInfoContext returns comprehensive IP information with context support.
//
// It finds the most specific live route for the address with point lookups on
// the prefix index: for each mask length from the longest (32 or 128) down to 0
// it looks up the address's network prefix at that length, and the first length
// with a live route wins. That is at most 33 or 129 indexed lookups, however
// many routes are live. The prefix is looked up in the text form net/netip
// prints, which is the form the prefix handler stores.
func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error) { func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error) {
addr, err := netip.ParseAddr(ip) // Parse the IP to validate it
if err != nil { parsedIP := net.ParseIP(ip)
if parsedIP == nil {
return nil, fmt.Errorf("%w: %s", ErrInvalidIP, ip) return nil, fmt.Errorf("%w: %s", ErrInvalidIP, ip)
} }
// An IPv4 address written in IPv6 form (::ffff:192.0.2.1) is an IPv4 lookup.
addr = addr.Unmap()
info := &IPInfo{IP: ip, IPVersion: ipVersionV6} // Determine IP version
routesTable, prefixesTable := "live_routes_v6", "prefixes_v6" ipv4 := parsedIP.To4()
if addr.Is4() { if ipv4 != nil {
info.IPVersion = ipVersionV4 return d.getIPv4Info(ctx, ip, ipv4)
routesTable, prefixesTable = "live_routes_v4", "prefixes_v4"
} }
// Route info with peer count and prefix first_seen return d.getIPv6Info(ctx, ip, parsedIP)
//nolint:gosec // Table names are hardcoded based on IP version }
query := fmt.Sprintf(`
// getIPv4Info returns comprehensive IP information for an IPv4 address.
func (d *Database) getIPv4Info(ctx context.Context, ip string, ipv4 net.IP) (*IPInfo, error) {
info := &IPInfo{
IP: ip,
IPVersion: ipVersionV4,
}
ipUint := ipToUint32(ipv4)
// Get route info with peer count and prefix first_seen
query := `
SELECT SELECT
lr.prefix, lr.prefix,
lr.mask_length, lr.mask_length,
lr.origin_asn, lr.origin_asn,
lr.last_updated, lr.last_updated,
(SELECT COUNT(DISTINCT peer_ip) FROM %[1]s WHERE prefix = lr.prefix) as num_peers, (SELECT COUNT(DISTINCT peer_ip) FROM live_routes_v4 WHERE prefix = lr.prefix) as num_peers,
p.first_seen, p.first_seen,
a.handle, a.handle,
a.description, a.description,
@@ -1916,56 +2083,42 @@ func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, er
a.abuse_email, a.abuse_email,
a.rir, a.rir,
a.whois_updated_at a.whois_updated_at
FROM %[1]s lr FROM live_routes_v4 lr
LEFT JOIN %[2]s p ON p.prefix = lr.prefix LEFT JOIN prefixes_v4 p ON p.prefix = lr.prefix
LEFT JOIN asns a ON a.asn = lr.origin_asn LEFT JOIN asns a ON a.asn = lr.origin_asn
WHERE lr.prefix = ? WHERE lr.ip_start <= ? AND lr.ip_end >= ?
ORDER BY lr.mask_length DESC
LIMIT 1 LIMIT 1
`, routesTable, prefixesTable) `
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() }()
var handle, description, asName, orgName, orgID, address, countryCode, abuseEmail, rir sql.NullString var handle, description, asName, orgName, orgID, address, countryCode, abuseEmail, rir sql.NullString
var prefixFirstSeen sql.NullTime var prefixFirstSeen sql.NullTime
var whoisUpdatedAt sql.NullTime var whoisUpdatedAt sql.NullTime
found := false err := d.db.QueryRowContext(ctx, query, ipUint, ipUint).Scan(
for bits := addr.BitLen(); bits >= 0; bits-- { &info.Netblock,
prefix := netip.PrefixFrom(addr, bits).Masked().String() &info.MaskLength,
err := stmt.QueryRowContext(ctx, prefix).Scan( &info.ASN,
&info.Netblock, &info.LastSeen,
&info.MaskLength, &info.NumPeers,
&info.ASN, &prefixFirstSeen,
&info.LastSeen, &handle,
&info.NumPeers, &description,
&prefixFirstSeen, &asName,
&handle, &orgName,
&description, &orgID,
&asName, &address,
&orgName, &countryCode,
&orgID, &abuseEmail,
&address, &rir,
&countryCode, &whoisUpdatedAt,
&abuseEmail, )
&rir, if err != nil {
&whoisUpdatedAt, if err == sql.ErrNoRows {
) return nil, fmt.Errorf("%w for IP %s", ErrNoRoute, ip)
if errors.Is(err, sql.ErrNoRows) {
continue
} }
if err != nil {
return nil, fmt.Errorf("failed to query routes: %w", err)
}
found = true
break return nil, fmt.Errorf("failed to query routes: %w", err)
}
if !found {
return nil, fmt.Errorf("%w for IP %s", ErrNoRoute, ip)
} }
info.Handle = handle.String info.Handle = handle.String
@@ -1989,26 +2142,116 @@ func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, er
return info, nil return info, nil
} }
// getIPv6Info returns comprehensive IP information for an IPv6 address.
func (d *Database) getIPv6Info(ctx context.Context, ip string, parsedIP net.IP) (*IPInfo, error) {
info := &IPInfo{
IP: ip,
IPVersion: ipVersionV6,
}
// For IPv6, scan all routes and find best match
query := `
SELECT DISTINCT
lr.prefix,
lr.mask_length,
lr.origin_asn,
lr.last_updated,
a.handle,
a.description,
a.as_name,
a.org_name,
a.org_id,
a.address,
a.country_code,
a.abuse_email,
a.rir,
a.whois_updated_at
FROM live_routes_v6 lr
LEFT JOIN asns a ON a.asn = lr.origin_asn
ORDER BY lr.mask_length DESC
`
rows, err := d.db.QueryContext(ctx, query)
if err != nil {
return nil, fmt.Errorf("failed to query routes: %w", err)
}
defer func() { _ = rows.Close() }()
bestMaskLength := -1
for rows.Next() {
var prefix string
var maskLength, originASN int
var lastUpdated time.Time
var handle, description, asName, orgName, orgID, address, countryCode, abuseEmail, rir sql.NullString
var whoisUpdatedAt sql.NullTime
if err := rows.Scan(
&prefix, &maskLength, &originASN, &lastUpdated,
&handle, &description, &asName, &orgName, &orgID,
&address, &countryCode, &abuseEmail, &rir, &whoisUpdatedAt,
); err != nil {
continue
}
_, ipNet, err := net.ParseCIDR(prefix)
if err != nil {
continue
}
if ipNet.Contains(parsedIP) && maskLength > bestMaskLength {
info.Netblock = prefix
info.MaskLength = maskLength
info.ASN = originASN
info.LastSeen = lastUpdated
info.Handle = handle.String
info.Description = description.String
info.ASName = asName.String
info.OrgName = orgName.String
info.OrgID = orgID.String
info.Address = address.String
info.CountryCode = countryCode.String
info.AbuseEmail = abuseEmail.String
info.RIR = rir.String
bestMaskLength = maskLength
if !whoisUpdatedAt.Valid {
info.NeedsWHOISRefresh = true
} else {
const staleThreshold = 30 * 24 * time.Hour
info.NeedsWHOISRefresh = time.Since(whoisUpdatedAt.Time) > staleThreshold
}
}
}
if bestMaskLength == -1 {
return nil, fmt.Errorf("%w for IP %s", ErrNoRoute, ip)
}
// Get peer count and first_seen for IPv6
countQuery := `SELECT COUNT(DISTINCT peer_ip) FROM live_routes_v6 WHERE prefix = ?`
_ = d.db.QueryRowContext(ctx, countQuery, info.Netblock).Scan(&info.NumPeers)
firstSeenQuery := `SELECT first_seen FROM prefixes_v6 WHERE prefix = ?`
var prefixFirstSeen sql.NullTime
err = d.db.QueryRowContext(ctx, firstSeenQuery, info.Netblock).Scan(&prefixFirstSeen)
if err == nil && prefixFirstSeen.Valid {
info.FirstSeen = prefixFirstSeen.Time
}
return info, nil
}
// Vacuum runs incremental vacuum to reclaim unused pages without blocking writes. // Vacuum runs incremental vacuum to reclaim unused pages without blocking writes.
// It frees up to the specified number of pages per call (0 = all freeable pages). // It frees up to the specified number of pages per call (0 = all freeable pages).
func (d *Database) Vacuum(ctx context.Context) error { func (d *Database) Vacuum(ctx context.Context) error {
// Free up to 1000 pages per call (~4MB with default 4KB page size) // Free up to 1000 pages per call (~4MB with default 4KB page size)
// This keeps each vacuum operation quick and non-blocking // This keeps each vacuum operation quick and non-blocking
const pagesToFree = 1000 const pagesToFree = 1000
rows, err := d.db.QueryContext(ctx, fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree)) _, err := d.db.ExecContext(ctx, fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree))
if err != nil { if err != nil {
return fmt.Errorf("failed to run incremental vacuum: %w", err) return fmt.Errorf("failed to run incremental vacuum: %w", err)
} }
defer func() { _ = rows.Close() }()
// SQLite frees one page each time the statement steps, and each step
// returns a row, so every row must be read for the pragma to free more
// than one page.
for rows.Next() {
}
if err := rows.Err(); err != nil {
return fmt.Errorf("failed to run incremental vacuum: %w", err)
}
return nil return nil
} }
+285 -160
View File
@@ -3,36 +3,23 @@ package database
import ( import (
"context" "context"
"database/sql" "database/sql"
"errors" "net"
"fmt"
"net/netip"
"sync" "sync"
"testing" "testing"
"time" "time"
"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"
) )
// 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
// memory"; the DSN change must leave temp_store below this so they spill to disk. // memory"; the DSN change must leave temp_store below this so they spill to disk.
const tempStoreMemory = 2 const tempStoreMemory = 2
// autoVacuumIncremental is the PRAGMA auto_vacuum value meaning "incremental".
const autoVacuumIncremental = 2
// vacuumTestRoutes is how many routes the vacuum test writes and then deletes,
// enough to leave many free pages in the file.
const vacuumTestRoutes = 2000
// heldConnections is how many pooled connections the pragma test holds open at // heldConnections is how many pooled connections the pragma test holds open at
// once so each is a distinct SQLite connection that parsed the DSN. // once so each is a distinct SQLite connection that parsed the DSN.
const heldConnections = 5 const heldConnections = 5
// testPeerIP is the peer every route in the IP lookup test is learned from.
const testPeerIP = "192.0.2.254"
// Parameters for the checkpoint-contention regression test. // Parameters for the checkpoint-contention regression test.
const ( const (
// contentionIterations is how many batch writes race the checkpoint loop. // contentionIterations is how many batch writes race the checkpoint loop.
@@ -45,98 +32,286 @@ const (
asnSecondBand = 100 asnSecondBand = 100
) )
// TestGetIPInfoFindsMostSpecificLiveRoute stores nested live prefixes for both func TestIPToUint32(t *testing.T) {
// families and checks that a lookup returns the most specific one covering the tests := []struct {
// address, ErrNoRoute when none covers it, and the next less specific prefix name string
// once the only route of the most specific one is withdrawn. ip string
func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) { expected uint32
cfg := &config.Config{StateDir: t.TempDir()} }{
{
db, err := New(cfg, logger.New()) name: "Simple IP",
if err != nil { ip: "192.168.1.1",
t.Fatalf("failed to create database: %v", err) expected: 3232235777, // 192<<24 + 168<<16 + 1<<8 + 1
} },
defer func() { _ = db.Close() }() {
name: "Minimum IP",
// Nested live prefixes, each originated by its own AS. ip: "0.0.0.0",
origins := map[string]int{ expected: 0,
"10.0.0.0/8": 64500, },
"10.1.0.0/16": 64501, {
"10.1.2.0/24": 64502, name: "Maximum IP",
"2001:db8::/32": 64500, ip: "255.255.255.255",
"2001:db8:1::/48": 64501, expected: 4294967295,
"2001:db8:1:2::/64": 64502, },
} {
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) name: "10.0.0.0",
routes := make([]*LiveRoute, 0, len(origins)) ip: "10.0.0.0",
for prefix, asn := range origins { expected: 167772160,
routes = append(routes, &LiveRoute{ },
ID: uuid.New(), {
Prefix: prefix, name: "172.16.0.0",
MaskLength: netip.MustParsePrefix(prefix).Bits(), ip: "172.16.0.0",
IPVersion: detectIPVersion(prefix), expected: 2886729728,
OriginASN: asn, },
PeerIP: testPeerIP, {
ASPath: []int{asn}, name: "8.8.8.8",
NextHop: testPeerIP, ip: "8.8.8.8",
LastUpdated: ts, expected: 134744072,
}) },
} {
if err := db.UpsertLiveRouteBatch(routes); err != nil { name: "1.2.3.4",
t.Fatalf("UpsertLiveRouteBatch: %v", err) ip: "1.2.3.4",
expected: 16909060,
},
} }
// lookup checks that ip resolves to the live prefix want, or to ErrNoRoute for _, tt := range tests {
// when want is empty. t.Run(tt.name, func(t *testing.T) {
lookup := func(ip, want string) { ip := net.ParseIP(tt.ip)
t.Helper() if ip == nil {
t.Fatalf("Failed to parse IP: %s", tt.ip)
info, err := db.GetIPInfo(ip)
if want == "" {
if !errors.Is(err, ErrNoRoute) {
t.Errorf("GetIPInfo(%s) = %+v, %v; want ErrNoRoute", ip, info, err)
} }
return result := ipToUint32(ip)
} if result != tt.expected {
if err != nil { t.Errorf("ipToUint32(%s) = %d, want %d", tt.ip, result, tt.expected)
t.Errorf("GetIPInfo(%s): %v", ip, err) }
return // Test with IPv4-mapped IPv6 address
} ip6 := net.ParseIP(tt.ip).To16()
if info.Netblock != want || info.MaskLength != netip.MustParsePrefix(want).Bits() || if ip6 != nil {
info.ASN != origins[want] { result6 := ipToUint32(ip6)
t.Errorf("GetIPInfo(%s) = %s (mask %d) AS%d, want %s AS%d", if result6 != tt.expected {
ip, info.Netblock, info.MaskLength, info.ASN, want, origins[want]) t.Errorf("ipToUint32(%s as IPv6) = %d, want %d", tt.ip, result6, tt.expected)
} }
}
})
}
}
func TestCalculateIPv4Range(t *testing.T) {
tests := []struct {
name string
cidr string
wantStart uint32
wantEnd uint32
wantErr bool
}{
{
name: "Single IP /32",
cidr: "192.168.1.1/32",
wantStart: 3232235777,
wantEnd: 3232235777,
},
{
name: "Class C /24",
cidr: "192.168.1.0/24",
wantStart: 3232235776, // 192.168.1.0
wantEnd: 3232236031, // 192.168.1.255
},
{
name: "Class B /16",
cidr: "192.168.0.0/16",
wantStart: 3232235520, // 192.168.0.0
wantEnd: 3232301055, // 192.168.255.255
},
{
name: "Class A /8",
cidr: "10.0.0.0/8",
wantStart: 167772160, // 10.0.0.0
wantEnd: 184549375, // 10.255.255.255
},
{
name: "Entire IPv4 space /0",
cidr: "0.0.0.0/0",
wantStart: 0,
wantEnd: 4294967295,
},
{
name: "Small subnet /30",
cidr: "192.168.1.0/30",
wantStart: 3232235776, // 192.168.1.0
wantEnd: 3232235779, // 192.168.1.3
},
{
name: "Medium subnet /20",
cidr: "172.16.0.0/20",
wantStart: 2886729728, // 172.16.0.0
wantEnd: 2886733823, // 172.16.15.255
},
{
name: "Private range 172.16/12",
cidr: "172.16.0.0/12",
wantStart: 2886729728, // 172.16.0.0
wantEnd: 2887778303, // 172.31.255.255
},
{
name: "Google DNS /29",
cidr: "8.8.8.8/29",
wantStart: 134744072, // 8.8.8.8 (network is actually 8.8.8.8 with /29)
wantEnd: 134744079, // 8.8.8.15
},
{
name: "Non-zero host bits",
cidr: "192.168.1.5/24",
wantStart: 3232235776, // 192.168.1.0 (network address)
wantEnd: 3232236031, // 192.168.1.255
},
{
name: "Invalid CIDR",
cidr: "192.168.1.1/33",
wantErr: true,
},
{
name: "Invalid IP",
cidr: "256.256.256.256/24",
wantErr: true,
},
{
name: "IPv6 CIDR",
cidr: "2001:db8::/32",
wantErr: true,
},
{
name: "Empty CIDR",
cidr: "",
wantErr: true,
},
{
name: "Missing mask",
cidr: "192.168.1.1",
wantErr: true,
},
} }
lookup("10.1.2.3", "10.1.2.0/24") for _, tt := range tests {
lookup("::ffff:10.1.2.3", "10.1.2.0/24") t.Run(tt.name, func(t *testing.T) {
lookup("10.1.3.4", "10.1.0.0/16") start, end, err := CalculateIPv4Range(tt.cidr)
lookup("10.2.0.1", "10.0.0.0/8")
lookup("192.0.2.1", "")
lookup("2001:db8:1:2::3", "2001:db8:1:2::/64")
lookup("2001:db8:1:3::4", "2001:db8:1::/48")
lookup("2001:db8:2::1", "2001:db8::/32")
lookup("2001:db9::1", "")
err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{ if tt.wantErr {
{Prefix: "10.1.2.0/24", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV4}, if err == nil {
{Prefix: "2001:db8:1:2::/64", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV6}, t.Errorf("CalculateIPv4Range(%s) expected error, got nil", tt.cidr)
}) }
if err != nil { return
t.Fatalf("DeleteLiveRouteBatch: %v", err) }
if err != nil {
t.Errorf("CalculateIPv4Range(%s) unexpected error: %v", tt.cidr, err)
return
}
if start != tt.wantStart {
t.Errorf("CalculateIPv4Range(%s) start = %d, want %d", tt.cidr, start, tt.wantStart)
}
if end != tt.wantEnd {
t.Errorf("CalculateIPv4Range(%s) end = %d, want %d", tt.cidr, end, tt.wantEnd)
}
// Verify that start <= end
if start > end {
t.Errorf("CalculateIPv4Range(%s) start (%d) > end (%d)", tt.cidr, start, end)
}
// Verify the range size matches the CIDR mask
if !tt.wantErr && tt.cidr != "" {
_, ipNet, _ := net.ParseCIDR(tt.cidr)
if ipNet != nil {
ones, bits := ipNet.Mask.Size()
expectedSize := uint32(1) << uint(bits-ones)
actualSize := end - start + 1
if actualSize != expectedSize {
t.Errorf("CalculateIPv4Range(%s) range size = %d, want %d", tt.cidr, actualSize, expectedSize)
}
}
}
})
}
}
func TestIPv4RangeIntegration(t *testing.T) {
// Test that our functions work correctly together
tests := []struct {
name string
cidr string
testIPs []string
shouldContain []bool
}{
{
name: "192.168.1.0/24",
cidr: "192.168.1.0/24",
testIPs: []string{
"192.168.1.0",
"192.168.1.1",
"192.168.1.255",
"192.168.0.255",
"192.168.2.0",
},
shouldContain: []bool{true, true, true, false, false},
},
{
name: "10.0.0.0/8",
cidr: "10.0.0.0/8",
testIPs: []string{
"10.0.0.0",
"10.255.255.255",
"10.1.2.3",
"9.255.255.255",
"11.0.0.0",
},
shouldContain: []bool{true, true, true, false, false},
},
{
name: "172.16.0.0/12",
cidr: "172.16.0.0/12",
testIPs: []string{
"172.16.0.0",
"172.31.255.255",
"172.20.1.1",
"172.15.255.255",
"172.32.0.0",
},
shouldContain: []bool{true, true, true, false, false},
},
} }
lookup("10.1.2.3", "10.1.0.0/16") for _, tt := range tests {
lookup("2001:db8:1:2::3", "2001:db8:1::/48") t.Run(tt.name, func(t *testing.T) {
start, end, err := CalculateIPv4Range(tt.cidr)
if err != nil {
t.Fatalf("Failed to calculate range for %s: %v", tt.cidr, err)
}
for i, testIP := range tt.testIPs {
ip := net.ParseIP(testIP)
if ip == nil {
t.Fatalf("Failed to parse test IP: %s", testIP)
}
ipUint := ipToUint32(ip)
contained := ipUint >= start && ipUint <= end
if contained != tt.shouldContain[i] {
t.Errorf("IP %s in range %s: got %v, want %v", testIP, tt.cidr, contained, tt.shouldContain[i])
}
}
})
}
} }
// 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.
// incremental.
func TestConnectionPoolPragmas(t *testing.T) { func TestConnectionPoolPragmas(t *testing.T) {
cfg := &config.Config{StateDir: t.TempDir()} cfg := &config.Config{StateDir: t.TempDir()}
@@ -197,74 +372,6 @@ func TestConnectionPoolPragmas(t *testing.T) {
if hardHeapLimit != sqliteHardHeapLimitBytes { if hardHeapLimit != sqliteHardHeapLimitBytes {
t.Errorf("conn %d: hard_heap_limit = %d, want %d", i, hardHeapLimit, sqliteHardHeapLimitBytes) t.Errorf("conn %d: hard_heap_limit = %d, want %d", i, hardHeapLimit, sqliteHardHeapLimitBytes)
} }
var autoVacuum int
if err := c.QueryRowContext(ctx, "PRAGMA auto_vacuum").Scan(&autoVacuum); err != nil {
t.Fatalf("conn %d: failed to read auto_vacuum: %v", i, err)
}
if autoVacuum != autoVacuumIncremental {
t.Errorf("conn %d: auto_vacuum = %d, want %d (incremental)", i, autoVacuum, autoVacuumIncremental)
}
}
}
// TestVacuumReturnsFreePages checks that after routes are deleted from a new
// database, one Vacuum call returns every page they used. The deletes leave
// fewer free pages than the 1000 Vacuum frees per call, so none may remain.
// With auto_vacuum off (issue https://git.eeqj.de/sneak/routewatch/issues/43)
// the free pages stayed in the file, and with the PRAGMA run by ExecContext
// Vacuum freed only one page per call.
func TestVacuumReturnsFreePages(t *testing.T) {
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() }()
ctx := context.Background()
ts := time.Now().UTC()
const asn = 64500
routes := make([]*LiveRoute, 0, vacuumTestRoutes)
deletions := make([]LiveRouteDeletion, 0, vacuumTestRoutes)
for i := range vacuumTestRoutes {
route := mkV4Route(t, fmt.Sprintf("10.%d.%d.0/24", i/256, i%256), asn, ts)
routes = append(routes, route)
deletions = append(deletions, LiveRouteDeletion{
Prefix: route.Prefix,
OriginASN: asn,
PeerIP: route.PeerIP,
IPVersion: ipVersionV4,
})
}
if err := db.UpsertLiveRouteBatch(routes); err != nil {
t.Fatalf("failed to write routes: %v", err)
}
if err := db.DeleteLiveRouteBatch(deletions); err != nil {
t.Fatalf("failed to delete routes: %v", err)
}
var before int
if err := db.db.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&before); err != nil {
t.Fatalf("failed to read freelist_count: %v", err)
}
if before == 0 {
t.Fatalf("no free pages after deleting %d routes", vacuumTestRoutes)
}
if err := db.Vacuum(ctx); err != nil {
t.Fatalf("Vacuum failed: %v", err)
}
var after int
if err := db.db.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&after); err != nil {
t.Fatalf("failed to read freelist_count: %v", err)
}
if after != 0 {
t.Errorf("free pages after Vacuum = %d of %d, want 0", after, before)
} }
} }
@@ -317,3 +424,21 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) {
cancel() cancel()
wg.Wait() wg.Wait()
} }
func BenchmarkIPToUint32(b *testing.B) {
ip := net.ParseIP("192.168.1.1")
b.ResetTimer()
for i := 0; i < b.N; i++ {
_ = ipToUint32(ip)
}
}
func BenchmarkCalculateIPv4Range(b *testing.B) {
cidr := "192.168.0.0/16"
b.ResetTimer()
for i := 0; i < b.N; i++ {
_, _, _ = CalculateIPv4Range(cidr)
}
}
+2
View File
@@ -63,6 +63,8 @@ type Store interface {
GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error) GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error)
// IP lookup operations // IP lookup operations
GetASInfoForIP(ip string) (*ASInfo, error)
GetASInfoForIPContext(ctx context.Context, ip string) (*ASInfo, error)
GetIPInfo(ip string) (*IPInfo, error) GetIPInfo(ip string) (*IPInfo, error)
GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error)
+13
View File
@@ -77,6 +77,9 @@ type LiveRoute struct {
ASPath []int `json:"as_path"` ASPath []int `json:"as_path"`
NextHop string `json:"next_hop"` NextHop string `json:"next_hop"`
LastUpdated time.Time `json:"last_updated"` LastUpdated time.Time `json:"last_updated"`
// IPv4 range fields for fast lookups (nil for IPv6)
V4IPStart *uint32 `json:"v4_ip_start,omitempty"`
V4IPEnd *uint32 `json:"v4_ip_end,omitempty"`
} }
// PrefixDistribution represents the distribution of prefixes by mask length // PrefixDistribution represents the distribution of prefixes by mask length
@@ -85,6 +88,16 @@ type PrefixDistribution struct {
Count int `json:"count"` Count int `json:"count"`
} }
// ASInfo represents AS information for an IP lookup (legacy format)
type ASInfo struct {
ASN int `json:"asn"`
Handle string `json:"handle"`
Description string `json:"description"`
Prefix string `json:"prefix"`
LastUpdated time.Time `json:"last_updated"`
Age string `json:"age"`
}
// IPInfo represents comprehensive IP information for the /ip endpoint // IPInfo represents comprehensive IP information for the /ip endpoint
type IPInfo struct { type IPInfo struct {
IP string `json:"ip"` IP string `json:"ip"`
+6
View File
@@ -107,6 +107,9 @@ 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,
-- IPv4 range columns for fast lookups
ip_start INTEGER NOT NULL, -- Start of IPv4 range as 32-bit unsigned int
ip_end INTEGER NOT NULL, -- End of IPv4 range as 32-bit unsigned int
UNIQUE(prefix, origin_asn, peer_ip) UNIQUE(prefix, origin_asn, peer_ip)
); );
@@ -120,6 +123,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,
-- Note: IPv6 doesn't use integer range columns
UNIQUE(prefix, origin_asn, peer_ip) UNIQUE(prefix, origin_asn, peer_ip)
); );
@@ -128,6 +132,8 @@ CREATE INDEX IF NOT EXISTS idx_live_routes_v4_prefix ON live_routes_v4(prefix);
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_length ON live_routes_v4(mask_length); CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_length ON live_routes_v4(mask_length);
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_v4_origin_asn ON live_routes_v4(origin_asn);
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);
-- Indexes for IPv4 range queries
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_ip_range ON live_routes_v4(ip_start, ip_end);
-- Index to optimize prefix distribution queries -- Index to optimize prefix distribution queries
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);
@@ -232,6 +232,25 @@ func (m *mockStore) GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, i
return m.GetLiveRouteCounts() return m.GetLiveRouteCounts()
} }
// GetASInfoForIP mock implementation
func (m *mockStore) GetASInfoForIP(ip string) (*database.ASInfo, error) {
// Simple mock - return a test AS
now := time.Now()
return &database.ASInfo{
ASN: 15169,
Handle: "GOOGLE",
Description: "Google LLC",
Prefix: "8.8.8.0/24",
LastUpdated: now.Add(-5 * time.Minute),
Age: "5m0s",
}, nil
}
// GetASInfoForIPContext mock implementation with context support
func (m *mockStore) GetASInfoForIPContext(ctx context.Context, ip string) (*database.ASInfo, error) {
return m.GetASInfoForIP(ip)
}
// GetASDetails mock implementation // GetASDetails mock implementation
func (m *mockStore) GetASDetails(asn int) (*database.ASN, []database.LiveRoute, error) { func (m *mockStore) GetASDetails(asn int) (*database.ASN, []database.LiveRoute, error) {
m.mu.Lock() m.mu.Lock()
+44 -18
View File
@@ -2,7 +2,6 @@ package routewatch
import ( import (
"net" "net"
"net/netip"
"strings" "strings"
"sync" "sync"
"time" "time"
@@ -109,7 +108,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
for _, announcement := range msg.Announcements { for _, announcement := range msg.Announcements {
for _, prefix := range announcement.Prefixes { for _, prefix := range announcement.Prefixes {
h.batch = append(h.batch, prefixUpdate{ h.batch = append(h.batch, prefixUpdate{
prefix: canonicalPrefix(prefix), prefix: prefix,
originASN: originASN, originASN: originASN,
peer: msg.Peer, peer: msg.Peer,
messageType: "announcement", messageType: "announcement",
@@ -126,7 +125,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
// Process withdrawals // Process withdrawals
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: prefix,
originASN: originASN, // Use the originASN from path if available originASN: originASN, // Use the originASN from path if available
peer: msg.Peer, peer: msg.Peer,
messageType: "withdrawal", messageType: "withdrawal",
@@ -265,21 +264,6 @@ func (h *PrefixHandler) flushBatchLocked() {
h.lastFlush = time.Now() h.lastFlush = time.Now()
} }
// canonicalPrefix returns prefix in the text form net/netip prints, the form
// the IP lookup builds when it looks a prefix up. The feed sends IPv6
// announcements compressed ("2001:db8::/32") but IPv6 withdrawals uncompressed
// ("2001:db8:0:0:0:0:0:0/32"); stored as received, a withdrawal would not match
// the route its announcement stored. A prefix that does not parse is returned
// unchanged, and the batch flush reports it.
func canonicalPrefix(prefix string) string {
p, err := netip.ParsePrefix(prefix)
if err != nil {
return prefix
}
return p.Masked().String()
}
// parseCIDR extracts the mask length and IP version from a prefix string // parseCIDR extracts the mask length and IP version from a prefix string
func parseCIDR(prefix string) (maskLength int, ipVersion int, err error) { func parseCIDR(prefix string) (maskLength int, ipVersion int, err error) {
_, ipNet, err := net.ParseCIDR(prefix) _, ipNet, err := net.ParseCIDR(prefix)
@@ -331,6 +315,20 @@ func (h *PrefixHandler) processAnnouncement(_ *database.Prefix, update prefixUpd
LastUpdated: update.timestamp, LastUpdated: update.timestamp,
} }
// For IPv4, calculate the IP range
if ipVersion == ipv4Version {
start, end, err := database.CalculateIPv4Range(update.prefix)
if err == nil {
liveRoute.V4IPStart = &start
liveRoute.V4IPEnd = &end
} else {
h.logger.Error("Failed to calculate IPv4 range",
"prefix", update.prefix,
"error", err,
)
}
}
if err := h.db.UpsertLiveRoute(liveRoute); err != nil { if err := h.db.UpsertLiveRoute(liveRoute); err != nil {
h.logger.Error("Failed to upsert live route", h.logger.Error("Failed to upsert live route",
"prefix", update.prefix, "prefix", update.prefix,
@@ -374,6 +372,20 @@ func (h *PrefixHandler) createLiveRoute(update prefixUpdate) *database.LiveRoute
LastUpdated: update.timestamp, LastUpdated: update.timestamp,
} }
// For IPv4, calculate the IP range
if ipVersion == ipv4Version {
start, end, err := database.CalculateIPv4Range(update.prefix)
if err == nil {
liveRoute.V4IPStart = &start
liveRoute.V4IPEnd = &end
} else {
h.logger.Error("Failed to calculate IPv4 range",
"prefix", update.prefix,
"error", err,
)
}
}
return liveRoute return liveRoute
} }
@@ -413,6 +425,20 @@ func (h *PrefixHandler) processAnnouncementDirect(update prefixUpdate) {
LastUpdated: update.timestamp, LastUpdated: update.timestamp,
} }
// For IPv4, calculate the IP range
if ipVersion == ipv4Version {
start, end, err := database.CalculateIPv4Range(update.prefix)
if err == nil {
liveRoute.V4IPStart = &start
liveRoute.V4IPEnd = &end
} else {
h.logger.Error("Failed to calculate IPv4 range",
"prefix", update.prefix,
"error", err,
)
}
}
if err := h.db.UpsertLiveRoute(liveRoute); err != nil { if err := h.db.UpsertLiveRoute(liveRoute); err != nil {
h.logger.Error("Failed to upsert live route", h.logger.Error("Failed to upsert live route",
"prefix", update.prefix, "prefix", update.prefix,
-74
View File
@@ -1,74 +0,0 @@
package routewatch
import (
"errors"
"testing"
"time"
"git.eeqj.de/sneak/routewatch/internal/config"
"git.eeqj.de/sneak/routewatch/internal/database"
"git.eeqj.de/sneak/routewatch/internal/logger"
"git.eeqj.de/sneak/routewatch/internal/ristypes"
)
const testPeerIP = "2001:db8:ffff::1"
// TestPrefixHandlerStoresPrefixesTheIPLookupFinds runs announcements and
// withdrawals through the prefix handler into a real database and looks the
// addresses up. The prefixes are written the way the feed sends them: IPv6
// announcements compressed, IPv6 withdrawals uncompressed. The withdrawal must
// remove the route the announcement stored, and the IP lookup must find the
// stored prefix.
func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(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()}
flush := func() {
h.mu.Lock()
defer h.mu.Unlock()
h.flushBatchLocked()
}
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
h.HandleMessage(&ristypes.RISMessage{
Peer: testPeerIP,
Path: ristypes.ASPath{testASNA, testASNB},
ParsedTimestamp: ts,
Announcements: []ristypes.RISAnnouncement{{
NextHop: testPeerIP,
Prefixes: []string{"2001:db8:1::/48", "192.0.2.0/24"},
}},
})
flush()
for ip, want := range map[string]string{
"2001:db8:1::1": "2001:db8:1::/48",
"192.0.2.1": "192.0.2.0/24",
} {
info, err := db.GetIPInfo(ip)
if err != nil {
t.Fatalf("GetIPInfo(%s) after announcement: %v", ip, err)
}
if info.Netblock != want || info.ASN != testASNB {
t.Errorf("GetIPInfo(%s) = %s AS%d, want %s AS%d", ip, info.Netblock, info.ASN, want, testASNB)
}
}
h.HandleMessage(&ristypes.RISMessage{
Peer: testPeerIP,
ParsedTimestamp: ts.Add(time.Minute),
Withdrawals: []string{"2001:db8:1:0:0:0:0:0/48", "192.0.2.0/24"},
})
flush()
for _, ip := range []string{"2001:db8:1::1", "192.0.2.1"} {
if info, err := db.GetIPInfo(ip); !errors.Is(err, database.ErrNoRoute) {
t.Errorf("GetIPInfo(%s) after withdrawal = %+v, %v; want ErrNoRoute", ip, info, err)
}
}
}
+5 -1
View File
@@ -89,11 +89,15 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) {
} }
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
start, end, err := database.CalculateIPv4Range("198.51.100.0/24")
if err != nil {
t.Fatalf("CalculateIPv4Range: %v", err)
}
if err := db.UpsertLiveRouteBatch([]*database.LiveRoute{ if err := db.UpsertLiveRouteBatch([]*database.LiveRoute{
{ {
ID: uuid.New(), Prefix: "198.51.100.0/24", MaskLength: 24, IPVersion: 4, ID: uuid.New(), Prefix: "198.51.100.0/24", MaskLength: 24, IPVersion: 4,
OriginASN: 64500, PeerIP: "192.0.2.1", ASPath: []int{64500}, NextHop: "192.0.2.1", OriginASN: 64500, PeerIP: "192.0.2.1", ASPath: []int{64500}, NextHop: "192.0.2.1",
LastUpdated: ts, LastUpdated: ts, V4IPStart: &start, V4IPEnd: &end,
}, },
{ {
ID: uuid.New(), Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6, ID: uuid.New(), Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6,