Compare commits

..
2 Commits
Author SHA1 Message Date
sneak 9ca9c3c9fa Bring README and TODO.md up to date with the code (closes #72)
check / check (push) Canceled after 0s
README "HTTP Endpoints" lists every route setupRoutes registers with what
it answers, and "Database Schema" matches schema.sql. Other sentences the
code contradicts are corrected: the handlers' batch sizes, the stream
client, the macOS state directory, the build command and what
script/precommit checks; "Code Structure" gains the version and whois
packages.

TODO.md gets one line for each issue landed on next since 2026-10-03, and
a current Status, Next Step and Future Steps.

Model: opus-5-5
2026-10-19 13:49:00 +00:00
clawbot d84dd56460 Make live route writes cheaper so the prefix handler keeps up (closes #96)
check / check (push) Canceled after 0s
The live route tables lose their random UUID key; the row id the
prefix-length sample draws is declared as the integer key. The prefix
tables are keyed by prefix and keep only first_seen; a batch inserts only
prefixes not stored yet. A route update sets only the columns that
change. Each kept index names the queries using it. Every write goes
through one connection kept for writing, with a 1 GiB page cache; reading
connections get 16 MiB, heap limits rise to 1.5 and 2 GiB (README
"Memory"). Batch writes go in key order. A benchmark run by hand writes
250,000 updates on 5 million routes faster than the feed's 10,000 a
second; a live run follows.

Model: opus-5-5
2026-10-19 15:43:38 +02:00
11 changed files with 741 additions and 269 deletions

No files matched your search

+36 -16
View File
@@ -142,8 +142,7 @@ routewatch/
- Signal handlers registered for graceful shutdown - Signal handlers registered for graceful shutdown
2. **Initialization** (`internal/routewatch/app.go`) 2. **Initialization** (`internal/routewatch/app.go`)
- Database created with SQLite schema (WAL mode, 64 MiB page cache per - Database created with SQLite schema (WAL mode, page caches as in "Memory")
connection)
- Message handlers registered with the streamer - Message handlers registered with the streamer
- HTTP server started on configured port - HTTP server started on configured port
@@ -199,21 +198,35 @@ routewatch/
```sql ```sql
-- Core tables -- Core tables
asns(id, number, handle, description, first_seen, last_seen) asns(asn PRIMARY KEY, handle, description,
prefixes_v4(id, prefix, mask_length, first_seen, last_seen) as_name, org_name, org_id, address, country_code, abuse_email,
prefixes_v6(id, prefix, mask_length, first_seen, last_seen) abuse_phone, tech_email, tech_phone, rir, rir_registration_date,
rir_last_modified, whois_raw,
first_seen, last_seen, whois_updated_at)
prefixes_v4(prefix PRIMARY KEY, first_seen) WITHOUT ROWID
prefixes_v6(prefix PRIMARY KEY, first_seen) WITHOUT ROWID
-- Live routing tables (one per IP version) -- Live routing tables (one per IP version), one route per peer and prefix
live_routes_v4(id, prefix, mask_length, origin_asn, peer_ip, as_path, live_routes_v4(id INTEGER PRIMARY KEY, prefix, mask_length, origin_asn,
next_hop, last_updated) peer_ip, as_path, next_hop, last_updated,
live_routes_v6(id, prefix, mask_length, origin_asn, peer_ip, as_path, UNIQUE(peer_ip, prefix))
next_hop, last_updated) live_routes_v6(id INTEGER PRIMARY KEY, prefix, mask_length, origin_asn,
peer_ip, as_path, next_hop, last_updated,
UNIQUE(peer_ip, prefix))
-- Relationship tracking -- Relationship tracking
peerings(id, as_a, as_b, first_seen, last_seen) peerings(id PRIMARY KEY, as_a, as_b, first_seen, last_seen,
bgp_peers(id, peer_ip, peer_asn, last_message_type, last_seen) UNIQUE(as_a, as_b))
bgp_peers(id PRIMARY KEY, peer_ip UNIQUE, peer_asn, first_seen, last_seen,
last_message_type)
-- Nothing in the daemon writes this table
announcements(id PRIMARY KEY, prefix_id, peer_asn, origin_asn, path,
next_hop, timestamp, is_withdrawal)
``` ```
The indexes are in `internal/database/schema.sql`.
## Configuration ## Configuration
Configuration is handled via environment variables and OS-specific paths: Configuration is handled via environment variables and OS-specific paths:
@@ -245,8 +258,12 @@ data it tracks. The image sets ceilings that keep it inside a 5 GiB container.
Budget: Budget:
- Go heap: a 1.5 GiB soft limit (`GOMEMLIMIT=1536MiB`, set in the image). - Go heap: a 1.5 GiB soft limit (`GOMEMLIMIT=1536MiB`, set in the image).
- SQLite: at most 640 MiB of page cache across the connection pool (64 MiB per - SQLite: 1 GiB of page cache for the one connection every write uses, which
connection, 10 connections) and a 1.5 GiB hard heap limit for the C library. keeps the pages the route writes touch from one batch to the next, and 16 MiB
for each of the 9 reading connections, whose cache SQLite empties whenever
another connection has written: about 1.15 GiB in all. A 1.5 GiB soft heap
limit sits above the caches, so it does not shrink them, and a 2 GiB hard heap
limit caps the C library.
- glibc allocator: the SQLite C library runs on glibc `malloc`, which frees - glibc allocator: the SQLite C library runs on glibc `malloc`, which frees
page-cache chunks back to per-arena free lists rather than to the kernel, so page-cache chunks back to per-arena free lists rather than to the kernel, so
process RSS tracks the high-water mark of those arenas, not SQLite's live process RSS tracks the high-water mark of those arenas, not SQLite's live
@@ -256,6 +273,9 @@ Budget:
database writes are already serialized. database writes are already serialized.
- About 0.2 GiB for everything else in the runtime. - About 0.2 GiB for everything else in the runtime.
The ceilings add up to 3.7 GiB: the Go heap's 1.5 GiB, SQLite's 2 GiB and the
0.2 GiB for the rest.
Run the container with a memory limit of 5 GiB and swap disabled: Run the container with a memory limit of 5 GiB and swap disabled:
```bash ```bash
@@ -274,8 +294,8 @@ What happens at each limit:
- Go soft limit: as the heap approaches `GOMEMLIMIT`, the runtime runs garbage - Go soft limit: as the heap approaches `GOMEMLIMIT`, the runtime runs garbage
collection more aggressively rather than growing further. collection more aggressively rather than growing further.
- SQLite: at a 1 GiB soft heap limit it recycles its page cache instead of - SQLite: at the 1.5 GiB soft heap limit it recycles its page cache instead of
allocating more; at the 1.5 GiB hard heap limit a statement fails with an allocating more; at the 2 GiB hard heap limit a statement fails with an
out-of-memory error, and the handler logs it and drops that batch. The process out-of-memory error, and the handler logs it and drops that batch. The process
keeps running. keeps running.
- Handler queues: each of the four handler queues holds at most 20,000 messages. - Handler queues: each of the four handler queues holds at most 20,000 messages.
+11 -7
View File
@@ -18,22 +18,26 @@ every push. The image sets memory ceilings for a 5 GiB container (README
"Memory") and the README says how to run it under upaas (README "Running under "Memory") and the README says how to run it under upaas (README "Running under
upaas"). A 25-hour run of `a9b6860` on the live feed peaked at about 0.6 GiB upaas"). A 25-hour run of `a9b6860` on the live feed peaked at about 0.6 GiB
(https://git.eeqj.de/sneak/routewatch/issues/3). A run of `825afa2` on the live (https://git.eeqj.de/sneak/routewatch/issues/3). A run of `825afa2` on the live
feed found that the prefix handler drops about a quarter of its updates, because feed found the prefix handler dropping about a quarter of its updates because
its route writes are slower than the feed its route writes were slower than the feed
(https://git.eeqj.de/sneak/routewatch/issues/96). (https://git.eeqj.de/sneak/routewatch/issues/96); the cheaper writes since then
are measured by a benchmark, not yet on the live feed.
# Next Step # Next Step
Make route writes faster than the feed `next` waits for sneak to merge it to `main` through
(https://git.eeqj.de/sneak/routewatch/issues/96), in rework in https://git.eeqj.de/sneak/routewatch/pulls/87.
https://git.eeqj.de/sneak/routewatch/pulls/98. Merging `next` to `main` through
https://git.eeqj.de/sneak/routewatch/pulls/87 is sneak's.
# Completed Steps # Completed Steps
- 2026-10-19: README "HTTP Endpoints" lists every route with what it answers, - 2026-10-19: README "HTTP Endpoints" lists every route with what it answers,
"Database Schema" matches `schema.sql`, and the rest of the README matches the "Database Schema" matches `schema.sql`, and the rest of the README matches the
code; this file brought up to date (closes #72) code; this file brought up to date (closes #72)
- 2026-10-19: route writes are cheaper: the live route and prefix tables lose
their random UUID keys and the prefix tables their duplicate index, a route
update sets only the columns that change, and every write goes through one
connection with a page cache of 1 GiB (README "Memory"). A benchmark writes
the prefix handler's largest batch faster than the feed sends it (closes #96)
- 2026-10-19: the WAL checkpoint, the incremental vacuum and ANALYZE take the - 2026-10-19: the WAL checkpoint, the incremental vacuum and ANALYZE take the
database lock every batch write holds, so a vacuum no longer fails with database lock every batch write holds, so a vacuum no longer fails with
"database is locked" and an ANALYZE no longer makes a batch write lose its "database is locked" and an ANALYZE no longer makes a batch write lose its
@@ -29,7 +29,6 @@ func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute {
} }
return &LiveRoute{ return &LiveRoute{
ID: uuid.New(),
Prefix: prefix, Prefix: prefix,
MaskLength: maskLength, MaskLength: maskLength,
IPVersion: ipVersionV4, IPVersion: ipVersionV4,
@@ -44,7 +43,6 @@ func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute {
// mkV6Route builds an IPv6 live route. // mkV6Route builds an IPv6 live route.
func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute { func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute {
return &LiveRoute{ return &LiveRoute{
ID: uuid.New(),
Prefix: prefix, Prefix: prefix,
MaskLength: 32, MaskLength: 32,
IPVersion: ipVersionV6, IPVersion: ipVersionV6,
+206 -172
View File
@@ -1,25 +1,27 @@
package database package database
import ( import (
"cmp"
"context" "context"
"database/sql" "database/sql"
_ "embed" _ "embed"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"maps"
"math/rand/v2" "math/rand/v2"
"net" "net"
"net/netip" "net/netip"
"os" "os"
"path/filepath" "path/filepath"
"runtime" "runtime"
"slices"
"sync" "sync"
"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"
"git.eeqj.de/sneak/routewatch/pkg/asinfo" "git.eeqj.de/sneak/routewatch/pkg/asinfo"
"github.com/google/uuid"
_ "github.com/mattn/go-sqlite3" // CGO SQLite driver _ "github.com/mattn/go-sqlite3" // CGO SQLite driver
) )
@@ -52,20 +54,31 @@ const (
minLookupMaskLengthV6 = 16 minLookupMaskLengthV6 = 16
) )
// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every // SQLite memory tuning (README "Memory"). cache_size and busy_timeout go in the
// pooled connection gets them; the heap limits are process-wide and set once. // DSN so every pooled connection gets them; the connection kept for writing
// then sets its own cache size. The heap limits are process-wide and set once.
const ( const (
// sqliteCacheSizeKiB is the per-connection page cache; negative means KiB. // maxConns is the size of the connection pool: the connection kept for
// -65536 = 64 MiB, so at most 640 MiB across the 10-connection pool. // writing and 9 reading connections.
sqliteCacheSizeKiB = -65536 maxConns = 10
// sqliteReadCacheSizeKiB is the page cache of each reading connection;
// negative means KiB. -16384 = 16 MiB, at most 144 MiB for the 9. A reading
// connection's cache is emptied whenever another connection has written
// since its last read, many times a minute, so a large one is wasted.
sqliteReadCacheSizeKiB = -16384
// sqliteWriteCacheSizeKiB is the page cache of the connection kept for
// writing, -1048576 = 1 GiB. It keeps its pages from one write to the next,
// so the index pages the route writes touch stay in memory.
sqliteWriteCacheSizeKiB = -1048576
// sqliteBusyTimeoutMs is how long a connection waits on a locked database. // sqliteBusyTimeoutMs is how long a connection waits on a locked database.
sqliteBusyTimeoutMs = 5000 sqliteBusyTimeoutMs = 5000
// sqliteSoftHeapLimitBytes (1 GiB) makes SQLite recycle its cache rather // sqliteSoftHeapLimitBytes (1.5 GiB) makes SQLite recycle its cache rather
// than allocate once its C heap passes this size. // than allocate once its C heap passes this size. It is above the caches
sqliteSoftHeapLimitBytes = 1073741824 // together, so it does not shrink them.
// sqliteHardHeapLimitBytes (1.5 GiB) fails a statement with SQLITE_NOMEM sqliteSoftHeapLimitBytes = 1610612736
// sqliteHardHeapLimitBytes (2 GiB) fails a statement with SQLITE_NOMEM
// instead of growing the C heap without bound. // instead of growing the C heap without bound.
sqliteHardHeapLimitBytes = 1610612736 sqliteHardHeapLimitBytes = 2147483648
) )
// Settings of the ANALYZE that Analyze runs. // Settings of the ANALYZE that Analyze runs.
@@ -106,7 +119,10 @@ var (
// Database manages the SQLite database connection and operations. // Database manages the SQLite database connection and operations.
type Database struct { type Database struct {
db *sql.DB db *sql.DB
// writer is the connection kept for writing, taken from db when the
// database opens. Every write uses it while holding mu.
writer *sql.Conn
logger *logger.Logger logger *logger.Logger
path string path string
mu sync.Mutex mu sync.Mutex
@@ -144,7 +160,7 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
"file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d"+ "file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d"+
"&_auto_vacuum=incremental&_journal_mode=WAL&_txlock=immediate", "&_auto_vacuum=incremental&_journal_mode=WAL&_txlock=immediate",
dbPath, dbPath,
sqliteCacheSizeKiB, sqliteReadCacheSizeKiB,
sqliteBusyTimeoutMs, sqliteBusyTimeoutMs,
) )
@@ -153,20 +169,34 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
return nil, fmt.Errorf("failed to open database: %w", err) return nil, fmt.Errorf("failed to open database: %w", err)
} }
err = db.PingContext(context.Background()) ctx := context.Background()
err = db.PingContext(ctx)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to ping database: %w", err) return nil, fmt.Errorf("failed to ping database: %w", err)
} }
// Set connection pool parameters // Set connection pool parameters
// Multiple connections allow concurrent reads while writes are serialized // Multiple connections allow concurrent reads while writes are serialized
const maxConns = 10
db.SetMaxOpenConns(maxConns) db.SetMaxOpenConns(maxConns)
db.SetMaxIdleConns(maxConns) db.SetMaxIdleConns(maxConns)
db.SetConnMaxLifetime(0) db.SetConnMaxLifetime(0)
writer, err := db.Conn(ctx)
if err != nil {
return nil, fmt.Errorf("failed to open the writing connection: %w", err)
}
_, err = writer.ExecContext(ctx,
fmt.Sprintf("PRAGMA cache_size=%d", sqliteWriteCacheSizeKiB))
if err != nil {
return nil, fmt.Errorf("failed to set the writing connection's cache: %w",
err)
}
database := &Database{ database := &Database{
db: db, logger: logger, path: dbPath, counts: &liveCounts{}, db: db, writer: writer, logger: logger, path: dbPath,
counts: &liveCounts{},
} }
err = database.Initialize() err = database.Initialize()
@@ -188,10 +218,11 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
// Initialize creates the database schema if it doesn't exist. // Initialize creates the database schema if it doesn't exist.
func (d *Database) Initialize() error { func (d *Database) Initialize() error {
// Set SQLite pragmas for performance. Per-connection settings (cache_size, // Set SQLite pragmas for performance. Per-connection settings (cache_size,
// synchronous, busy_timeout, journal_mode) live in the DSN; temp_store is // synchronous, busy_timeout, journal_mode) live in the DSN, and New sets the
// left at its default so DISTINCT temp B-trees spill to disk instead of C // writing connection's cache_size; temp_store is left at its default so
// heap. The heap limits below are process-wide, so setting them once here is // DISTINCT temp B-trees spill to disk instead of C heap. The heap limits
// enough for the whole pool. // below are process-wide, so setting them once here is enough for the whole
// pool.
pragmas := []string{ pragmas := []string{
"PRAGMA journal_mode=WAL", // Write-Ahead Logging "PRAGMA journal_mode=WAL", // Write-Ahead Logging
fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes), fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes),
@@ -208,7 +239,7 @@ func (d *Database) Initialize() error {
// Run WAL checkpoint on startup to consolidate any existing WAL // Run WAL checkpoint on startup to consolidate any existing WAL
var walPages, checkpointedPages, movedPages int var walPages, checkpointedPages, movedPages int
err := d.db.QueryRowContext(context.Background(), err := d.writer.QueryRowContext(context.Background(),
"PRAGMA wal_checkpoint(TRUNCATE)", "PRAGMA wal_checkpoint(TRUNCATE)",
).Scan(&walPages, &checkpointedPages, &movedPages) ).Scan(&walPages, &checkpointedPages, &movedPages)
if err != nil { if err != nil {
@@ -229,35 +260,47 @@ func (d *Database) Initialize() error {
return nil return nil
} }
// Close closes the database connection. // Close closes the writing connection and then the connection pool.
func (d *Database) Close() error { func (d *Database) Close() error {
return d.db.Close() return errors.Join(d.writer.Close(), d.db.Close())
} }
// A live-route upsert is an UPDATE followed, only when no row matched, by an // A live-route upsert sets the origin AS of the peer's route for the prefix
// INSERT. A live route is keyed by prefix and peer, so an announcement replaces // when the announcement changes it, then updates the rest of the route, and
// the peer's route for the prefix whatever its origin AS. The UPDATE's // inserts the route only when that update matched no row. A live route is keyed
// rows-affected count (1 for an existing key, 0 for a new one) is what lets the // by prefix and peer, so an announcement replaces the peer's route for the
// in-memory route counters stay exact without a COUNT(*). Callers hold the // prefix whatever its origin AS. The update's rows-affected count (1 for an
// database write lock, so no other writer can insert the same key between the // existing key, 0 for a new one) is what lets the in-memory route counters stay
// two statements. The id column is set only on INSERT, so an updated route // exact without a COUNT(*). Callers hold the database write lock, so no other
// keeps its original id, exactly as the previous ON CONFLICT upsert did. // writer can insert the same key between the statements.
//
// SQLite rewrites a row's entry in every index on a column an UPDATE sets, even
// when the value stays the same, so the update sets only the columns that
// change: mask_length follows from the prefix, and origin_asn has a statement
// of its own that changes nothing when the origin AS stays the same, as it
// mostly does.
const ( const (
updateLiveRouteOriginV4SQL = `UPDATE live_routes_v4 SET origin_asn = ?
WHERE prefix = ? AND peer_ip = ? AND origin_asn != ?`
updateLiveRouteV4SQL = `UPDATE live_routes_v4 updateLiveRouteV4SQL = `UPDATE live_routes_v4
SET mask_length = ?, origin_asn = ?, as_path = ?, next_hop = ?, SET as_path = ?, next_hop = ?, last_updated = ?
last_updated = ? WHERE prefix = ? AND peer_ip = ?` WHERE prefix = ? AND peer_ip = ?`
insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 insertLiveRouteV4SQL = `INSERT INTO live_routes_v4
(id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, (prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` last_updated) VALUES (?, ?, ?, ?, ?, ?, ?)`
updateLiveRouteOriginV6SQL = `UPDATE live_routes_v6 SET origin_asn = ?
WHERE prefix = ? AND peer_ip = ? AND origin_asn != ?`
updateLiveRouteV6SQL = `UPDATE live_routes_v6 updateLiveRouteV6SQL = `UPDATE live_routes_v6
SET mask_length = ?, origin_asn = ?, as_path = ?, next_hop = ?, SET as_path = ?, next_hop = ?, last_updated = ?
last_updated = ? WHERE prefix = ? AND peer_ip = ?` WHERE prefix = ? AND peer_ip = ?`
insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 insertLiveRouteV6SQL = `INSERT INTO live_routes_v6
(id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, (prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` last_updated) VALUES (?, ?, ?, ?, ?, ?, ?)`
) )
// Before a new route is inserted and after a route is deleted, the write looks // Before a new route is inserted and after a route is deleted, the write looks
@@ -278,15 +321,22 @@ const (
WHERE prefix = ? AND peer_ip = ?` WHERE prefix = ? AND peer_ip = ?`
) )
// upsertRouteRow updates a live route with the update statement of its IP // upsertRouteRow updates a live route with the origin and update statements of
// version's table, inserting it when no row matched. It reports whether a new // its IP version's table, inserting it when no row matched. It reports whether
// row was inserted and whether that row is the first live route for its prefix. // a new row was inserted and whether that row is the first live route for its
// prefix.
func upsertRouteRow( func upsertRouteRow(
ctx context.Context, upd, ins, has *sql.Stmt, route *LiveRoute, ctx context.Context, setOrigin, upd, ins, has *sql.Stmt, route *LiveRoute,
pathJSON string, pathJSON string,
) (bool, bool, error) { ) (bool, bool, error) {
res, err := upd.ExecContext(ctx, route.MaskLength, route.OriginASN, pathJSON, _, err := setOrigin.ExecContext(ctx, route.OriginASN, route.Prefix,
route.NextHop, route.LastUpdated, route.Prefix, route.PeerIP) route.PeerIP, route.OriginASN)
if err != nil {
return false, false, err
}
res, err := upd.ExecContext(ctx, pathJSON, route.NextHop, route.LastUpdated,
route.Prefix, route.PeerIP)
if err != nil { if err != nil {
return false, false, err return false, false, err
} }
@@ -307,9 +357,9 @@ func upsertRouteRow(
return false, false, err return false, false, err
} }
_, err = ins.ExecContext(ctx, route.ID.String(), route.Prefix, _, err = ins.ExecContext(ctx, route.Prefix, route.MaskLength,
route.MaskLength, route.OriginASN, route.PeerIP, pathJSON, route.OriginASN, route.PeerIP, pathJSON, route.NextHop,
route.NextHop, route.LastUpdated) route.LastUpdated)
if err != nil { if err != nil {
return false, false, err return false, false, err
} }
@@ -327,13 +377,23 @@ func upsertRoutesInTable(
return 0, nil, nil return 0, nil, nil
} }
family, updateSQL, insertSQL, lookupSQL := ipv4Family, updateLiveRouteV4SQL, family, originSQL, updateSQL := ipv4Family, updateLiveRouteOriginV4SQL,
insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL updateLiveRouteV4SQL
insertSQL, lookupSQL := insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
if ipVersion == ipVersionV6 { if ipVersion == ipVersionV6 {
family, updateSQL, insertSQL, lookupSQL = ipv6Family, family, originSQL, updateSQL = ipv6Family, updateLiveRouteOriginV6SQL,
updateLiveRouteV6SQL, insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL updateLiveRouteV6SQL
insertSQL, lookupSQL = insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
} }
setOrigin, err := tx.PrepareContext(ctx, originSQL)
if err != nil {
return 0, nil, fmt.Errorf("failed to prepare %s origin statement: %w",
family, err)
}
defer func() { _ = setOrigin.Close() }()
upd, err := tx.PrepareContext(ctx, updateSQL) upd, err := tx.PrepareContext(ctx, updateSQL)
if err != nil { if err != nil {
return 0, nil, fmt.Errorf("failed to prepare %s update statement: %w", return 0, nil, fmt.Errorf("failed to prepare %s update statement: %w",
@@ -365,8 +425,8 @@ func upsertRoutesInTable(
return 0, nil, fmt.Errorf("failed to encode AS path: %w", err) return 0, nil, fmt.Errorf("failed to encode AS path: %w", err)
} }
inserted, newPrefix, err := upsertRouteRow(ctx, upd, ins, has, route, inserted, newPrefix, err := upsertRouteRow(ctx, setOrigin, upd, ins, has,
string(pathJSON)) route, string(pathJSON))
if err != nil { if err != nil {
return 0, nil, fmt.Errorf("failed to upsert route %s: %w", return 0, nil, fmt.Errorf("failed to upsert route %s: %w",
route.Prefix, err) route.Prefix, err)
@@ -400,9 +460,18 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
} }
defer d.rollback(tx) defer d.rollback(tx)
// The rows are written in key order, peer then prefix, so consecutive writes
// land on the same index pages. The sort is stable, so of two updates of
// one route the later one is still written last.
sorted := slices.SortedStableFunc(slices.Values(routes),
func(a, b *LiveRoute) int {
return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP),
cmp.Compare(a.Prefix, b.Prefix))
})
var routesV4, routesV6 []*LiveRoute var routesV4, routesV6 []*LiveRoute
for _, route := range routes { for _, route := range sorted {
if route.IPVersion == ipVersionV4 { if route.IPVersion == ipVersionV4 {
routesV4 = append(routesV4, route) routesV4 = append(routesV4, route)
} else { } else {
@@ -546,9 +615,17 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
} }
defer d.rollback(tx) defer d.rollback(tx)
// The rows are deleted in key order, peer then prefix, as
// UpsertLiveRouteBatch writes them.
sorted := slices.SortedFunc(slices.Values(deletions),
func(a, b LiveRouteDeletion) int {
return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP),
cmp.Compare(a.Prefix, b.Prefix))
})
var deletionsV4, deletionsV6 []LiveRouteDeletion var deletionsV4, deletionsV6 []LiveRouteDeletion
for _, del := range deletions { for _, del := range sorted {
if del.IPVersion == ipVersionV4 { if del.IPVersion == ipVersionV4 {
deletionsV4 = append(deletionsV4, del) deletionsV4 = append(deletionsV4, del)
} else { } else {
@@ -683,46 +760,11 @@ func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
return int(deletedV4 + deletedV6), nil return int(deletedV4 + deletedV6), nil
} }
// upsertPrefix updates the last_seen time of prefix with the statements of its // insertPrefixesInTable inserts the prefixes, all of one IP version, that the
// IP version's table, inserting the prefix when it is new. It reports whether // prefix table of that version does not hold yet, each first seen at its time
// it inserted it. // in prefixes. A prefix already stored is left as it is. It returns how many it
func upsertPrefix( // inserted.
ctx context.Context, selectStmt, updateStmt, insertStmt *sql.Stmt, func insertPrefixesInTable(
prefix string, timestamp time.Time,
) (bool, error) {
var id string
err := selectStmt.QueryRowContext(ctx, prefix).Scan(&id)
switch {
case err == nil:
// Prefix exists, update last_seen
_, err = updateStmt.ExecContext(ctx, timestamp, prefix)
if err != nil {
return false, fmt.Errorf("failed to update prefix %s: %w",
prefix, err)
}
return false, nil
case errors.Is(err, sql.ErrNoRows):
// Prefix doesn't exist, create it
_, err = insertStmt.ExecContext(ctx, generateUUID().String(), prefix,
timestamp, timestamp)
if err != nil {
return false, fmt.Errorf("failed to insert prefix %s: %w",
prefix, err)
}
return true, nil
default:
return false, fmt.Errorf("failed to query prefix %s: %w", prefix, err)
}
}
// upsertPrefixesInTable updates the last_seen time of prefixes, all of one IP
// version, in the prefix table of that version, inserting the ones that are
// new. It returns how many it inserted.
func upsertPrefixesInTable(
ctx context.Context, tx *loggingTx, prefixes map[string]time.Time, ctx context.Context, tx *loggingTx, prefixes map[string]time.Time,
ipVersion int, ipVersion int,
) (int, error) { ) (int, error) {
@@ -731,33 +773,15 @@ func upsertPrefixesInTable(
} }
family := ipv4Family family := ipv4Family
selectSQL := "SELECT id FROM prefixes_v4 WHERE prefix = ?" insertSQL := "INSERT INTO prefixes_v4 (prefix, first_seen) VALUES (?, ?) " +
updateSQL := "UPDATE prefixes_v4 SET last_seen = ? WHERE prefix = ?" "ON CONFLICT (prefix) DO NOTHING"
insertSQL := "INSERT INTO prefixes_v4 (id, prefix, first_seen, last_seen) " +
"VALUES (?, ?, ?, ?)"
if ipVersion == ipVersionV6 { if ipVersion == ipVersionV6 {
family = ipv6Family family = ipv6Family
selectSQL = "SELECT id FROM prefixes_v6 WHERE prefix = ?" insertSQL = "INSERT INTO prefixes_v6 (prefix, first_seen) VALUES (?, ?) " +
updateSQL = "UPDATE prefixes_v6 SET last_seen = ? WHERE prefix = ?" "ON CONFLICT (prefix) DO NOTHING"
insertSQL = "INSERT INTO prefixes_v6 (id, prefix, first_seen, last_seen) " +
"VALUES (?, ?, ?, ?)"
} }
selectStmt, err := tx.PrepareContext(ctx, selectSQL)
if err != nil {
return 0, fmt.Errorf("failed to prepare %s select statement: %w",
family, err)
}
defer func() { _ = selectStmt.Close() }()
updateStmt, err := tx.PrepareContext(ctx, updateSQL)
if err != nil {
return 0, fmt.Errorf("failed to prepare %s update statement: %w",
family, err)
}
defer func() { _ = updateStmt.Close() }()
insertStmt, err := tx.PrepareContext(ctx, insertSQL) insertStmt, err := tx.PrepareContext(ctx, insertSQL)
if err != nil { if err != nil {
return 0, fmt.Errorf("failed to prepare %s insert statement: %w", return 0, fmt.Errorf("failed to prepare %s insert statement: %w",
@@ -767,14 +791,21 @@ func upsertPrefixesInTable(
var newPrefixes int var newPrefixes int
for prefix, timestamp := range prefixes { // The rows are written in key order, as UpsertLiveRouteBatch writes routes.
inserted, err := upsertPrefix(ctx, selectStmt, updateStmt, insertStmt, for _, prefix := range slices.Sorted(maps.Keys(prefixes)) {
prefix, timestamp) res, err := insertStmt.ExecContext(ctx, prefix, prefixes[prefix])
if err != nil { if err != nil {
return 0, err return 0, fmt.Errorf("failed to insert prefix %s: %w", prefix, err)
} }
if inserted { // One row for a new prefix, none for one already stored.
inserted, err := res.RowsAffected()
if err != nil {
return 0, fmt.Errorf("failed to count inserted prefix %s: %w",
prefix, err)
}
if inserted > 0 {
newPrefixes++ newPrefixes++
} }
} }
@@ -782,8 +813,9 @@ func upsertPrefixesInTable(
return newPrefixes, nil return newPrefixes, nil
} }
// UpdatePrefixesBatch updates the last_seen time for multiple prefixes in a // UpdatePrefixesBatch stores, in a single transaction, the prefixes that are
// single transaction, the IPv4 prefixes first and then the IPv6 prefixes. // not stored yet, each first seen at its time in prefixes: the IPv4 prefixes
// first and then the IPv6 prefixes.
func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
if len(prefixes) == 0 { if len(prefixes) == 0 {
return nil return nil
@@ -811,12 +843,12 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
ctx := context.Background() ctx := context.Background()
newV4, err := upsertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4) newV4, err := insertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4)
if err != nil { if err != nil {
return err return err
} }
newV6, err := upsertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6) newV6, err := insertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6)
if err != nil { if err != nil {
return err return err
} }
@@ -1047,35 +1079,15 @@ func (d *Database) GetOrCreatePrefix(
tableName = "prefixes_v6" tableName = "prefixes_v6"
} }
var ( var p Prefix
p Prefix
idStr string
)
query := fmt.Sprintf("SELECT id, prefix, first_seen, last_seen "+ query := fmt.Sprintf("SELECT prefix, first_seen FROM %s WHERE prefix = ?",
"FROM %s WHERE prefix = ?", tableName) tableName)
err = tx.QueryRow(query, prefix).Scan(&idStr, &p.Prefix, &p.FirstSeen, &p.LastSeen) err = tx.QueryRow(query, prefix).Scan(&p.Prefix, &p.FirstSeen)
if err == nil { if err == nil {
// Prefix exists, update last_seen // Prefix exists; the deferred rollback ends the read.
p.ID, _ = uuid.Parse(idStr)
p.IPVersion = ipVersion p.IPVersion = ipVersion
updateQuery := fmt.Sprintf("UPDATE %s SET last_seen = ? WHERE id = ?", tableName)
_, err = tx.Exec(updateQuery, timestamp, p.ID.String())
if err != nil {
return nil, err
}
p.LastSeen = timestamp
err = tx.Commit()
if err != nil {
d.logger.Error("Failed to commit transaction for prefix update",
"prefix", prefix, "error", err)
return nil, err
}
return &p, nil return &p, nil
} }
@@ -1085,12 +1097,11 @@ func (d *Database) GetOrCreatePrefix(
} }
// Prefix doesn't exist, create it // Prefix doesn't exist, create it
p = Prefix{ID: generateUUID(), Prefix: prefix, IPVersion: ipVersion, p = Prefix{Prefix: prefix, IPVersion: ipVersion, FirstSeen: timestamp}
FirstSeen: timestamp, LastSeen: timestamp}
insertQuery := fmt.Sprintf("INSERT INTO %s "+ insertQuery := fmt.Sprintf("INSERT INTO %s "+
"(id, prefix, first_seen, last_seen) VALUES (?, ?, ?, ?)", tableName) "(prefix, first_seen) VALUES (?, ?)", tableName)
_, err = tx.Exec(insertQuery, p.ID.String(), p.Prefix, p.FirstSeen, p.LastSeen) _, err = tx.Exec(insertQuery, p.Prefix, p.FirstSeen)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -1389,37 +1400,44 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
return fmt.Errorf("failed to encode AS path: %w", err) return fmt.Errorf("failed to encode AS path: %w", err)
} }
updateSQL, insertSQL, lookupSQL := updateLiveRouteV4SQL, originSQL, updateSQL := updateLiveRouteOriginV4SQL, updateLiveRouteV4SQL
insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL insertSQL, lookupSQL := insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
if route.IPVersion == ipVersionV6 { if route.IPVersion == ipVersionV6 {
updateSQL, insertSQL, lookupSQL = updateLiveRouteV6SQL, originSQL, updateSQL = updateLiveRouteOriginV6SQL, updateLiveRouteV6SQL
insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL insertSQL, lookupSQL = insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
} }
ctx := context.Background() ctx := context.Background()
// The write lock is held, so no other writer can insert this key between the // The write lock is held, so no other writer can insert this key between the
// update and the insert even though they are separate autocommit statements. // update and the insert even though they are separate autocommit statements.
upd, err := d.db.PrepareContext(ctx, updateSQL) setOrigin, err := d.writer.PrepareContext(ctx, originSQL)
if err != nil {
return fmt.Errorf("failed to prepare origin statement: %w", err)
}
defer func() { _ = setOrigin.Close() }()
upd, err := d.writer.PrepareContext(ctx, updateSQL)
if err != nil { if err != nil {
return fmt.Errorf("failed to prepare update statement: %w", err) return fmt.Errorf("failed to prepare update statement: %w", err)
} }
defer func() { _ = upd.Close() }() defer func() { _ = upd.Close() }()
ins, err := d.db.PrepareContext(ctx, insertSQL) ins, err := d.writer.PrepareContext(ctx, insertSQL)
if err != nil { if err != nil {
return fmt.Errorf("failed to prepare insert statement: %w", err) return fmt.Errorf("failed to prepare insert statement: %w", err)
} }
defer func() { _ = ins.Close() }() defer func() { _ = ins.Close() }()
has, err := d.db.PrepareContext(ctx, lookupSQL) has, err := d.writer.PrepareContext(ctx, lookupSQL)
if err != nil { if err != nil {
return fmt.Errorf("failed to prepare prefix lookup statement: %w", err) return fmt.Errorf("failed to prepare prefix lookup statement: %w", err)
} }
defer func() { _ = has.Close() }() defer func() { _ = has.Close() }()
inserted, newPrefix, err := upsertRouteRow(ctx, upd, ins, has, route, inserted, newPrefix, err := upsertRouteRow(ctx, setOrigin, upd, ins, has,
string(pathJSON)) route, string(pathJSON))
if err != nil { if err != nil {
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
} }
@@ -2008,7 +2026,7 @@ func (d *Database) UpdateASNWHOIS(
WHERE asn = ? WHERE asn = ?
` `
_, err := d.db.ExecContext(ctx, query, _, err := d.writer.ExecContext(ctx, query,
update.ASName, update.ASName,
update.OrgName, update.OrgName,
update.OrgID, update.OrgID,
@@ -2165,7 +2183,8 @@ func (d *Database) GetIPInfoContext(
// Vacuum runs incremental vacuum to return up to 1000 unused pages to the // Vacuum runs incremental vacuum to return up to 1000 unused pages to the
// filesystem. Like Analyze and Checkpoint, it holds the database lock, so it // filesystem. Like Analyze and Checkpoint, it holds the database lock, so it
// and the batch writes wait for each other instead of failing on SQLite's // and the batch writes wait for each other instead of failing on SQLite's
// write lock. // write lock, and returns ctx's error without running if ctx ended while it
// waited: the writing connection it uses does not check ctx itself.
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 // This keeps each vacuum operation quick
@@ -2174,7 +2193,12 @@ func (d *Database) Vacuum(ctx context.Context) error {
d.lock("Vacuum") d.lock("Vacuum")
defer d.unlock() defer d.unlock()
rows, err := d.db.QueryContext(ctx, err := ctx.Err()
if err != nil {
return fmt.Errorf("failed to run incremental vacuum: %w", err)
}
rows, err := d.writer.QueryContext(ctx,
fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree)) 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)
@@ -2204,9 +2228,14 @@ func (d *Database) Analyze(ctx context.Context) error {
d.lock("Analyze") d.lock("Analyze")
defer d.unlock() defer d.unlock()
err := ctx.Err()
if err != nil {
return fmt.Errorf("failed to analyze database: %w", err)
}
// analysis_limit is a setting of the connection, so it is set in the same // analysis_limit is a setting of the connection, so it is set in the same
// call as PRAGMA optimize, which runs both statements on one connection. // call as PRAGMA optimize, which runs both statements on one connection.
_, err := d.db.ExecContext(ctx, fmt.Sprintf( _, err = d.writer.ExecContext(ctx, fmt.Sprintf(
"PRAGMA analysis_limit=%d; PRAGMA optimize=%#x", "PRAGMA analysis_limit=%d; PRAGMA optimize=%#x",
analysisLimit, optimizeAllTables)) analysisLimit, optimizeAllTables))
if err != nil { if err != nil {
@@ -2223,7 +2252,12 @@ func (d *Database) Checkpoint(ctx context.Context) error {
d.lock("Checkpoint") d.lock("Checkpoint")
defer d.unlock() defer d.unlock()
_, err := d.db.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)") err := ctx.Err()
if err != nil {
return fmt.Errorf("failed to checkpoint WAL: %w", err)
}
_, err = d.writer.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)")
if err != nil { if err != nil {
return fmt.Errorf("failed to checkpoint WAL: %w", err) return fmt.Errorf("failed to checkpoint WAL: %w", err)
} }
@@ -2279,9 +2313,9 @@ func (d *Database) unlock() {
"held_by", lockedBy, "duration_ms", holdDuration.Milliseconds()) "held_by", lockedBy, "duration_ms", holdDuration.Milliseconds())
} }
// beginTx starts a new transaction with logging // beginTx starts a new transaction on the writing connection, with logging
func (d *Database) beginTx() (*loggingTx, error) { func (d *Database) beginTx() (*loggingTx, error) {
tx, err := d.db.BeginTx(context.Background(), nil) tx, err := d.writer.BeginTx(context.Background(), nil)
if err != nil { if err != nil {
return nil, err return nil, err
} }
+437 -6
View File
@@ -6,15 +6,16 @@ import (
"errors" "errors"
"fmt" "fmt"
"log/slog" "log/slog"
"math/rand/v2"
"net/netip" "net/netip"
"slices" "slices"
"strconv"
"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"
"github.com/mattn/go-sqlite3" "github.com/mattn/go-sqlite3"
) )
@@ -33,6 +34,9 @@ const vacuumTestRoutes = 2000
// 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
// bytesPerKiB turns a cache_size in KiB into bytes.
const bytesPerKiB = 1024
// testPeerIP is the peer every route in the IP lookup test is learned from. // testPeerIP is the peer every route in the IP lookup test is learned from.
const testPeerIP = "192.0.2.254" const testPeerIP = "192.0.2.254"
@@ -293,7 +297,7 @@ func TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer(t *testing.T) {
t.Fatalf("UpsertLiveRoute(%s): %v", tt.route.Prefix, err) t.Fatalf("UpsertLiveRoute(%s): %v", tt.route.Prefix, err)
} }
_, err = db.db.ExecContext(t.Context(), tt.insertSQL, uuid.NewString(), _, err = db.db.ExecContext(t.Context(), tt.insertSQL,
tt.route.Prefix, tt.route.MaskLength, otherOriginASN, tt.route.Prefix, tt.route.MaskLength, otherOriginASN,
tt.route.PeerIP, "[]", tt.route.NextHop, ts) tt.route.PeerIP, "[]", tt.route.NextHop, ts)
@@ -317,7 +321,6 @@ func writeOriginRoutes(t *testing.T, db *Database, origins map[string]int) {
for prefix, asn := range origins { for prefix, asn := range origins {
routes = append(routes, &LiveRoute{ routes = append(routes, &LiveRoute{
ID: uuid.New(),
Prefix: prefix, Prefix: prefix,
MaskLength: netip.MustParsePrefix(prefix).Bits(), MaskLength: netip.MustParsePrefix(prefix).Bits(),
IPVersion: detectIPVersion(prefix), IPVersion: detectIPVersion(prefix),
@@ -337,7 +340,7 @@ func writeOriginRoutes(t *testing.T, db *Database, origins map[string]int) {
// 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 heap limits, and sees the new file with auto_vacuum
// incremental. // incremental.
func TestConnectionPoolPragmas(t *testing.T) { func TestConnectionPoolPragmas(t *testing.T) {
t.Parallel() t.Parallel()
@@ -372,9 +375,9 @@ func TestConnectionPoolPragmas(t *testing.T) {
for i, c := range conns { for i, c := range conns {
cacheSize := readPragma(ctx, t, c, i, "cache_size") cacheSize := readPragma(ctx, t, c, i, "cache_size")
if cacheSize != sqliteCacheSizeKiB { if cacheSize != sqliteReadCacheSizeKiB {
t.Errorf("conn %d: cache_size = %d, want %d", t.Errorf("conn %d: cache_size = %d, want %d",
i, cacheSize, sqliteCacheSizeKiB) i, cacheSize, sqliteReadCacheSizeKiB)
} }
busyTimeout := readPragma(ctx, t, c, i, "busy_timeout") busyTimeout := readPragma(ctx, t, c, i, "busy_timeout")
@@ -389,6 +392,12 @@ func TestConnectionPoolPragmas(t *testing.T) {
i, tempStore, tempStoreMemory) i, tempStore, tempStoreMemory)
} }
softHeapLimit := readPragma(ctx, t, c, i, "soft_heap_limit")
if softHeapLimit != sqliteSoftHeapLimitBytes {
t.Errorf("conn %d: soft_heap_limit = %d, want %d",
i, softHeapLimit, sqliteSoftHeapLimitBytes)
}
hardHeapLimit := readPragma(ctx, t, c, i, "hard_heap_limit") hardHeapLimit := readPragma(ctx, t, c, i, "hard_heap_limit")
if hardHeapLimit != sqliteHardHeapLimitBytes { if hardHeapLimit != sqliteHardHeapLimitBytes {
t.Errorf("conn %d: hard_heap_limit = %d, want %d", t.Errorf("conn %d: hard_heap_limit = %d, want %d",
@@ -403,6 +412,51 @@ func TestConnectionPoolPragmas(t *testing.T) {
} }
} }
// TestWritingConnectionCache checks that the connection kept for writing has
// the large cache, and that the soft heap limit is above the caches of all the
// connections together: SQLite shrinks the caches to stay under it.
func TestWritingConnectionCache(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() }()
ctx := t.Context()
var writeCacheSize, readCacheSize, softHeapLimit int64
err = db.writer.QueryRowContext(ctx, "PRAGMA cache_size").
Scan(&writeCacheSize)
if err != nil {
t.Fatalf("failed to read the writing connection's cache_size: %v", err)
}
if writeCacheSize != sqliteWriteCacheSizeKiB {
t.Errorf("writing connection: cache_size = %d, want %d",
writeCacheSize, sqliteWriteCacheSizeKiB)
}
err = db.db.QueryRowContext(ctx, "PRAGMA cache_size").Scan(&readCacheSize)
if err != nil {
t.Fatalf("failed to read a reading connection's cache_size: %v", err)
}
err = db.db.QueryRowContext(ctx, "PRAGMA soft_heap_limit").
Scan(&softHeapLimit)
if err != nil {
t.Fatalf("failed to read soft_heap_limit: %v", err)
}
cachesBytes := -bytesPerKiB * (writeCacheSize + (maxConns-1)*readCacheSize)
if softHeapLimit <= cachesBytes {
t.Errorf("soft_heap_limit = %d, not above the %d bytes of the caches",
softHeapLimit, cachesBytes)
}
}
// readPragma reads the integer value of PRAGMA name on connection i, c. // readPragma reads the integer value of PRAGMA name on connection i, c.
func readPragma( func readPragma(
ctx context.Context, t *testing.T, c *sql.Conn, i int, name string, ctx context.Context, t *testing.T, c *sql.Conn, i int, name string,
@@ -1090,3 +1144,380 @@ func TestGetRandomPrefixesByLengthSkipsWithdrawnRoutes(t *testing.T) {
routes, kept.Prefix) routes, kept.Prefix)
} }
} }
// TestLiveRouteUpdateSetsOriginWhenItChanges announces a peer's route for a
// prefix, then again with the same origin AS and a new path, then with a new
// origin AS. Each time the route must carry the announced origin AS, path and
// time, and the AS details must list the prefix under that origin AS only. The
// update sets the origin AS with a statement of its own, which changes nothing
// while the origin AS stays the same.
func TestLiveRouteUpdateSetsOriginWhenItChanges(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 firstOrigin, secondOrigin = 64500, 64501
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
for _, asn := range []int{firstOrigin, secondOrigin} {
_, err := db.GetOrCreateASN(asn, ts)
if err != nil {
t.Fatalf("GetOrCreateASN(%d): %v", asn, err)
}
}
route := mkV4Route(t, "198.51.100.0/24", firstOrigin, ts)
for _, announced := range []struct {
origin int
path []int
}{
{firstOrigin, []int{firstOrigin}},
{firstOrigin, []int{secondOrigin, firstOrigin}},
{secondOrigin, []int{firstOrigin, secondOrigin}},
} {
ts = ts.Add(time.Minute)
route.OriginASN, route.ASPath, route.LastUpdated =
announced.origin, announced.path, ts
err := db.UpsertLiveRouteBatch([]*LiveRoute{route})
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
routes, _, err := db.GetPrefixDetails(route.Prefix)
if err != nil {
t.Fatalf("GetPrefixDetails: %v", err)
}
if len(routes) != 1 || routes[0].OriginASN != announced.origin ||
!slices.Equal(routes[0].ASPath, announced.path) ||
!routes[0].LastUpdated.Equal(ts) {
t.Errorf("after announcing origin AS%d, path %v at %s: routes %+v",
announced.origin, announced.path, ts, routes)
}
for _, asn := range []int{firstOrigin, secondOrigin} {
_, prefixes, err := db.GetASDetails(asn)
if err != nil {
t.Fatalf("GetASDetails(%d): %v", asn, err)
}
listed := len(prefixes) == 1 && prefixes[0].Prefix == route.Prefix
if listed != (asn == announced.origin) {
t.Errorf("after announcing origin AS%d, AS%d lists %+v",
announced.origin, asn, prefixes)
}
}
}
}
// 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.
func TestUpdatePrefixesBatchKeepsFirstSeen(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 known = "198.51.100.0/24"
first := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
later := first.Add(time.Hour)
err = db.UpdatePrefixesBatch(map[string]time.Time{known: first})
if err != nil {
t.Fatalf("UpdatePrefixesBatch: %v", err)
}
err = db.UpdatePrefixesBatch(map[string]time.Time{
known: later, testV6Prefix: later,
})
if err != nil {
t.Fatalf("UpdatePrefixesBatch: %v", err)
}
stats, err := db.GetStats()
if err != nil {
t.Fatalf("GetStats: %v", err)
}
if stats.IPv4Prefixes != 1 || stats.IPv6Prefixes != 1 {
t.Errorf("prefix counts = (v4 %d, v6 %d), want (1, 1)",
stats.IPv4Prefixes, stats.IPv6Prefixes)
}
for _, tt := range []struct {
query, prefix string
firstSeen time.Time
}{
{
query: "SELECT first_seen FROM prefixes_v4 WHERE prefix = ?",
prefix: known, firstSeen: first,
},
{
query: "SELECT first_seen FROM prefixes_v6 WHERE prefix = ?",
prefix: testV6Prefix, firstSeen: later,
},
} {
var firstSeen time.Time
err := db.db.QueryRowContext(t.Context(), tt.query, tt.prefix).
Scan(&firstSeen)
if err != nil {
t.Fatalf("reading %s: %v", tt.prefix, err)
}
if !firstSeen.Equal(tt.firstSeen) {
t.Errorf("%s: first seen %s, want %s", tt.prefix, firstSeen,
tt.firstSeen)
}
}
}
// TestUpsertLiveRouteBatchWritesInKeyOrder writes new routes of two peers out
// of key order, the last one a second announcement of the first, and checks
// that the rows were inserted in key order, peer then prefix (each new row's id
// is one above the last), and that the route announced twice carries the later
// announcement.
func TestUpsertLiveRouteBatchWritesInKeyOrder(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 (
firstOrigin, laterOrigin = 64500, 64501
// Three prefixes, in text order.
firstPrefix = "192.0.2.0/24"
secondPrefix = "198.51.100.0/24"
thirdPrefix = "203.0.113.0/24"
)
type row struct {
peer, prefix string
origin int
}
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
announcements := []row{
{secondV4PeerIP, secondPrefix, firstOrigin},
{testV4PeerIP, thirdPrefix, firstOrigin},
{secondV4PeerIP, firstPrefix, firstOrigin},
{testV4PeerIP, secondPrefix, firstOrigin},
{secondV4PeerIP, secondPrefix, laterOrigin},
}
routes := make([]*LiveRoute, 0, len(announcements))
for _, announced := range announcements {
route := mkV4Route(t, announced.prefix, announced.origin, ts)
route.PeerIP = announced.peer
routes = append(routes, route)
}
err = db.UpsertLiveRouteBatch(routes)
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
rows, err := db.db.QueryContext(t.Context(),
"SELECT peer_ip, prefix, origin_asn FROM live_routes_v4 ORDER BY id")
if err != nil {
t.Fatalf("reading the routes: %v", err)
}
defer func() { _ = rows.Close() }()
var stored []row
for rows.Next() {
var r row
err := rows.Scan(&r.peer, &r.prefix, &r.origin)
if err != nil {
t.Fatalf("reading a route: %v", err)
}
stored = append(stored, r)
}
err = rows.Err()
if err != nil {
t.Fatalf("reading the routes: %v", err)
}
want := []row{
{testV4PeerIP, secondPrefix, firstOrigin},
{testV4PeerIP, thirdPrefix, firstOrigin},
{secondV4PeerIP, firstPrefix, firstOrigin},
{secondV4PeerIP, secondPrefix, laterOrigin},
}
if !slices.Equal(stored, want) {
t.Errorf("routes in insertion order = %+v, want %+v", stored, want)
}
}
// Parameters for BenchmarkPrefixBatchOnFiveMillionRoutes.
const (
// benchPeers and benchPrefixes size the table the batches are written to:
// every peer has a route for every prefix, 5 million routes in all.
benchPeers = 10
benchPrefixes = 500000
// benchBatchSize is how many updates each batch holds, as many as a full
// prefix handler batch (prefixBatchSize in internal/routewatch).
benchBatchSize = 25000
// benchMaxBatchSize is the largest batch the prefix handler writes, the
// one it holds while a slow write runs (prefixBatchMaxSize in
// internal/routewatch).
benchMaxBatchSize = 250000
// benchWithdrawalEvery makes every fifth update of a timed batch a
// withdrawal; the others are announcements.
benchWithdrawalEvery = 5
// benchOrigins is how many origin ASes the prefixes are spread over.
benchOrigins = 50000
// benchFirstASN is the lowest AS number in the benchmark's AS paths.
benchFirstASN = 64500
// benchMaskLength is the mask length of every prefix in the benchmark.
benchMaskLength = 24
)
// BenchmarkPrefixBatchOnFiveMillionRoutes times the writes the prefix handler
// makes for one batch of announcements and withdrawals of routes picked at
// random, to a database that already holds 5 million live routes: 500,000
// IPv4 prefixes, each from 10 peers. It times a full batch of 25,000 updates,
// then the largest batch, 250,000. It builds that database first, one batch at
// a time, which takes some minutes and a few GB of disk under TMPDIR. It is
// run by hand, not by make test, for example in the test image:
//
// go test -run '^$' -bench FiveMillion -benchtime 10x ./internal/database/
func BenchmarkPrefixBatchOnFiveMillionRoutes(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() }()
//nolint:gosec // The benchmark needs the same batches every run.
rng := rand.New(rand.NewPCG(1, 2))
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
// Each peer sends its whole table, as when its session comes up.
for peer := range benchPeers {
for first := 0; first < benchPrefixes; first += benchBatchSize {
routes := make([]*LiveRoute, 0, benchBatchSize)
for prefix := first; prefix < first+benchBatchSize; prefix++ {
routes = append(routes, benchRoute(rng, prefix, peer, ts))
}
writeBenchBatch(b, db, routes, nil, ts)
}
}
for _, batchSize := range []int{benchBatchSize, benchMaxBatchSize} {
b.Run(strconv.Itoa(batchSize), func(b *testing.B) {
for b.Loop() {
ts = ts.Add(time.Second)
var (
routes []*LiveRoute
withdrawals []LiveRouteDeletion
)
for i := range batchSize {
route := benchRoute(rng, rng.IntN(benchPrefixes),
rng.IntN(benchPeers), ts)
if i%benchWithdrawalEvery == 0 {
withdrawals = append(withdrawals, LiveRouteDeletion{
Prefix: route.Prefix, PeerIP: route.PeerIP,
IPVersion: ipVersionV4,
})
} else {
routes = append(routes, route)
}
}
writeBenchBatch(b, db, routes, withdrawals, ts)
}
})
}
}
// benchRoute returns the route peer has for the prefix numbered prefix, a /24
// from 1.0.0.0/24 on, with a new AS path of three to six ASes from the peer's
// AS to the prefix's origin AS.
func benchRoute(rng *rand.Rand, prefix, peer int, ts time.Time) *LiveRoute {
const minPathLength, extraPathLength = 3, 4
origin := benchFirstASN + prefix%benchOrigins
pathLength := minPathLength + rng.IntN(extraPathLength)
path := make([]int, 0, pathLength)
path = append(path, benchFirstASN+peer)
for len(path) < pathLength-1 {
path = append(path, benchFirstASN+rng.IntN(benchOrigins))
}
path = append(path, origin)
peerIP := fmt.Sprintf("198.51.100.%d", 1+peer)
return &LiveRoute{
Prefix: fmt.Sprintf("%d.%d.%d.0/%d",
1+prefix>>16, (prefix>>8)%256, prefix%256, benchMaskLength),
MaskLength: benchMaskLength,
IPVersion: ipVersionV4,
OriginASN: origin,
PeerIP: peerIP,
ASPath: path,
NextHop: peerIP,
LastUpdated: ts,
}
}
// writeBenchBatch writes what the prefix handler writes for a batch: the
// announced routes, the withdrawals, and every prefix the batch names, of
// which only the ones not stored yet are inserted.
func writeBenchBatch(
b *testing.B, db *Database, routes []*LiveRoute,
withdrawals []LiveRouteDeletion, ts time.Time,
) {
b.Helper()
prefixes := make(map[string]time.Time, len(routes)+len(withdrawals))
for _, route := range routes {
prefixes[route.Prefix] = ts
}
for _, withdrawal := range withdrawals {
prefixes[withdrawal.Prefix] = ts
}
err := db.UpsertLiveRouteBatch(routes)
if err != nil {
b.Fatalf("UpsertLiveRouteBatch: %v", err)
}
err = db.DeleteLiveRouteBatch(withdrawals)
if err != nil {
b.Fatalf("DeleteLiveRouteBatch: %v", err)
}
err = db.UpdatePrefixesBatch(prefixes)
if err != nil {
b.Fatalf("UpdatePrefixesBatch: %v", err)
}
}
+1 -4
View File
@@ -35,13 +35,11 @@ type ASN struct {
} }
// Prefix represents an IP prefix (CIDR block) with its IP version (4 or 6) // Prefix represents an IP prefix (CIDR block) with its IP version (4 or 6)
// and first/last seen timestamps. // and the time it was first seen.
type Prefix struct { type Prefix struct {
ID uuid.UUID
Prefix string Prefix string
IPVersion int // 4 or 6 IPVersion int // 4 or 6
FirstSeen time.Time FirstSeen time.Time
LastSeen time.Time
} }
// Announcement represents a BGP announcement or withdrawal event, // Announcement represents a BGP announcement or withdrawal event,
@@ -71,7 +69,6 @@ type ASNPeering struct {
// //
//nolint:tagliatelle // field names are the existing JSON API //nolint:tagliatelle // field names are the existing JSON API
type LiveRoute struct { type LiveRoute struct {
ID uuid.UUID `json:"id"`
Prefix string `json:"prefix"` Prefix string `json:"prefix"`
MaskLength int `json:"mask_length"` MaskLength int `json:"mask_length"`
IPVersion int `json:"ip_version"` IPVersion int `json:"ip_version"`
+40 -38
View File
@@ -27,21 +27,18 @@ CREATE TABLE IF NOT EXISTS asns (
whois_updated_at DATETIME -- when we last fetched WHOIS data whois_updated_at DATETIME -- when we last fetched WHOIS data
); );
-- IPv4 prefixes table -- IPv4 prefixes table. Every read and write finds a row by its prefix, so the
-- rows are stored by prefix (WITHOUT ROWID) and need no index of their own.
CREATE TABLE IF NOT EXISTS prefixes_v4 ( CREATE TABLE IF NOT EXISTS prefixes_v4 (
id TEXT PRIMARY KEY, prefix TEXT PRIMARY KEY,
prefix TEXT UNIQUE NOT NULL, first_seen DATETIME NOT NULL
first_seen DATETIME NOT NULL, ) WITHOUT ROWID;
last_seen DATETIME NOT NULL
);
-- IPv6 prefixes table -- IPv6 prefixes table, stored as prefixes_v4 is.
CREATE TABLE IF NOT EXISTS prefixes_v6 ( CREATE TABLE IF NOT EXISTS prefixes_v6 (
id TEXT PRIMARY KEY, prefix TEXT PRIMARY KEY,
prefix TEXT UNIQUE NOT NULL, first_seen DATETIME NOT NULL
first_seen DATETIME NOT NULL, ) WITHOUT ROWID;
last_seen DATETIME NOT NULL
);
CREATE TABLE IF NOT EXISTS announcements ( CREATE TABLE IF NOT EXISTS announcements (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
@@ -75,11 +72,6 @@ CREATE TABLE IF NOT EXISTS bgp_peers (
last_message_type TEXT last_message_type TEXT
); );
-- Indexes for prefixes_v4 table
CREATE INDEX IF NOT EXISTS idx_prefixes_v4_prefix ON prefixes_v4(prefix);
-- Indexes for prefixes_v6 table
CREATE INDEX IF NOT EXISTS idx_prefixes_v6_prefix ON prefixes_v6(prefix);
CREATE INDEX IF NOT EXISTS idx_announcements_timestamp ON announcements(timestamp); CREATE INDEX IF NOT EXISTS idx_announcements_timestamp ON announcements(timestamp);
CREATE INDEX IF NOT EXISTS idx_announcements_prefix_id ON announcements(prefix_id); CREATE INDEX IF NOT EXISTS idx_announcements_prefix_id ON announcements(prefix_id);
CREATE INDEX IF NOT EXISTS idx_announcements_peer_asn ON announcements(peer_asn); CREATE INDEX IF NOT EXISTS idx_announcements_peer_asn ON announcements(peer_asn);
@@ -97,9 +89,14 @@ CREATE INDEX IF NOT EXISTS idx_bgp_peers_asn ON bgp_peers(peer_asn);
CREATE INDEX IF NOT EXISTS idx_bgp_peers_last_seen ON bgp_peers(last_seen); CREATE INDEX IF NOT EXISTS idx_bgp_peers_last_seen ON bgp_peers(last_seen);
CREATE INDEX IF NOT EXISTS idx_bgp_peers_ip ON bgp_peers(peer_ip); CREATE INDEX IF NOT EXISTS idx_bgp_peers_ip ON bgp_peers(peer_ip);
-- IPv4 routing table maintained by PrefixHandler -- IPv4 routing table maintained by PrefixHandler. A route write changes the
-- table and its entry in every index on a column the write sets, so the table
-- has only the indexes that the queries named with them use.
CREATE TABLE IF NOT EXISTS live_routes_v4 ( CREATE TABLE IF NOT EXISTS live_routes_v4 (
id TEXT PRIMARY KEY, -- The row id, which the prefix-length pages' sample draws through the
-- mask_length index. Declared as a column, it stays the same for as long
-- as the route is live, also through a VACUUM.
id INTEGER PRIMARY KEY,
prefix TEXT NOT NULL, prefix TEXT NOT NULL,
mask_length INTEGER NOT NULL, -- CIDR mask length (0-32) mask_length INTEGER NOT NULL, -- CIDR mask length (0-32)
origin_asn INTEGER NOT NULL, origin_asn INTEGER NOT NULL,
@@ -107,15 +104,15 @@ 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,
-- A peer has at most one route for a prefix. peer_ip comes first so that -- A peer has at most one route for a prefix. This key finds the route a
-- this key also finds all of a peer's routes, when its BGP session goes -- write updates or deletes and, peer_ip coming first, all of a peer's
-- down; lookups by prefix use the prefix index below. -- routes when its BGP session goes down.
UNIQUE(peer_ip, prefix) UNIQUE(peer_ip, prefix)
); );
-- IPv6 routing table maintained by PrefixHandler -- IPv6 routing table maintained by PrefixHandler, as live_routes_v4.
CREATE TABLE IF NOT EXISTS live_routes_v6 ( CREATE TABLE IF NOT EXISTS live_routes_v6 (
id TEXT PRIMARY KEY, id INTEGER PRIMARY KEY,
prefix TEXT NOT NULL, prefix TEXT NOT NULL,
mask_length INTEGER NOT NULL, -- CIDR mask length (0-128) mask_length INTEGER NOT NULL, -- CIDR mask length (0-128)
origin_asn INTEGER NOT NULL, origin_asn INTEGER NOT NULL,
@@ -123,24 +120,29 @@ 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,
-- As in live_routes_v4: peer_ip first, to find all of a peer's routes.
UNIQUE(peer_ip, prefix) UNIQUE(peer_ip, prefix)
); );
-- Indexes for live_routes_v4 table -- Indexes for both live route tables, and the queries that use each.
-- Finds a prefix's newest route in one search, whatever the number of peers. --
-- 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.
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_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_last_updated ON live_routes_v4(last_updated);
-- Index to optimize prefix distribution queries
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_prefix ON live_routes_v4(mask_length, prefix);
-- Indexes for live_routes_v6 table
-- Finds a prefix's newest route in one search, whatever the number of peers.
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 and the first route of it at or after a row id.
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_v6_mask_length ON live_routes_v6(mask_length); CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_length ON live_routes_v6(mask_length);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_origin_asn ON live_routes_v6(origin_asn); -- mask_length, prefix: the prefix-length pages' list of a mask length's
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_last_updated ON live_routes_v6(last_updated); -- prefixes in text order, and the prefix distribution read at startup.
-- 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_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.
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);
-- 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);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_last_updated ON live_routes_v6(last_updated);
+2 -2
View File
@@ -18,12 +18,12 @@ func logSlowQuery(logger *logger.Logger, query string, start time.Time) {
} }
} }
// exec wraps Exec with slow query logging // exec runs a statement on the writing connection, with slow query logging
func (d *Database) exec(query string, args ...any) error { func (d *Database) exec(query string, args ...any) error {
start := time.Now() start := time.Now()
defer logSlowQuery(d.logger, query, start) defer logSlowQuery(d.logger, query, start)
_, err := d.db.ExecContext(context.Background(), query, args...) _, err := d.writer.ExecContext(context.Background(), query, args...)
return err return err
} }
@@ -18,7 +18,6 @@ import (
"git.eeqj.de/sneak/routewatch/internal/ristypes" "git.eeqj.de/sneak/routewatch/internal/ristypes"
"git.eeqj.de/sneak/routewatch/internal/server" "git.eeqj.de/sneak/routewatch/internal/server"
"git.eeqj.de/sneak/routewatch/internal/streamer" "git.eeqj.de/sneak/routewatch/internal/streamer"
"github.com/google/uuid"
) )
// mockStore is a mock implementation of database.Store for testing // mockStore is a mock implementation of database.Store for testing
@@ -83,9 +82,7 @@ func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
defer m.mu.Unlock() defer m.mu.Unlock()
for prefix, timestamp := range prefixes { for prefix, timestamp := range prefixes {
if p, exists := m.Prefixes[prefix]; exists { if _, exists := m.Prefixes[prefix]; !exists {
p.LastSeen = timestamp
} else {
const ( const (
ipVersionV4 = 4 ipVersionV4 = 4
ipVersionV6 = 6 ipVersionV6 = 6
@@ -97,11 +94,9 @@ func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
} }
m.Prefixes[prefix] = &database.Prefix{ m.Prefixes[prefix] = &database.Prefix{
ID: uuid.New(),
Prefix: prefix, Prefix: prefix,
IPVersion: ipVersion, IPVersion: ipVersion,
FirstSeen: timestamp, FirstSeen: timestamp,
LastSeen: timestamp,
} }
} }
} }
@@ -117,8 +112,6 @@ func (m *mockStore) GetOrCreatePrefix(
defer m.mu.Unlock() defer m.mu.Unlock()
if p, exists := m.Prefixes[prefix]; exists { if p, exists := m.Prefixes[prefix]; exists {
p.LastSeen = timestamp
return p, nil return p, nil
} }
@@ -133,11 +126,9 @@ func (m *mockStore) GetOrCreatePrefix(
} }
p := &database.Prefix{ p := &database.Prefix{
ID: uuid.New(),
Prefix: prefix, Prefix: prefix,
IPVersion: ipVersion, IPVersion: ipVersion,
FirstSeen: timestamp, FirstSeen: timestamp,
LastSeen: timestamp,
} }
m.Prefixes[prefix] = p m.Prefixes[prefix] = p
m.PrefixCount++ m.PrefixCount++
@@ -370,11 +361,9 @@ func (m *mockStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
// Track prefix // Track prefix
if _, exists := m.Prefixes[route.Prefix]; !exists { if _, exists := m.Prefixes[route.Prefix]; !exists {
m.Prefixes[route.Prefix] = &database.Prefix{ m.Prefixes[route.Prefix] = &database.Prefix{
ID: uuid.New(),
Prefix: route.Prefix, Prefix: route.Prefix,
IPVersion: route.IPVersion, IPVersion: route.IPVersion,
FirstSeen: route.LastUpdated, FirstSeen: route.LastUpdated,
LastSeen: route.LastUpdated,
} }
m.PrefixCount++ m.PrefixCount++
-2
View File
@@ -12,7 +12,6 @@ import (
"git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/logger"
"git.eeqj.de/sneak/routewatch/internal/metrics" "git.eeqj.de/sneak/routewatch/internal/metrics"
"git.eeqj.de/sneak/routewatch/internal/ristypes" "git.eeqj.de/sneak/routewatch/internal/ristypes"
"github.com/google/uuid"
) )
const ( const (
@@ -461,7 +460,6 @@ func (h *PrefixHandler) createLiveRoute(update prefixUpdate) *database.LiveRoute
// Create live route record // Create live route record
liveRoute := &database.LiveRoute{ liveRoute := &database.LiveRoute{
ID: uuid.New(),
Prefix: update.prefix, Prefix: update.prefix,
MaskLength: maskLength, MaskLength: maskLength,
IPVersion: ipVersion, IPVersion: ipVersion,
+7 -8
View File
@@ -19,7 +19,6 @@ import (
"git.eeqj.de/sneak/routewatch/internal/metrics" "git.eeqj.de/sneak/routewatch/internal/metrics"
"git.eeqj.de/sneak/routewatch/internal/streamer" "git.eeqj.de/sneak/routewatch/internal/streamer"
"git.eeqj.de/sneak/routewatch/pkg/asinfo" "git.eeqj.de/sneak/routewatch/pkg/asinfo"
"github.com/google/uuid"
) )
// The prefix and peer addresses the prefix page tests store routes for. // The prefix and peer addresses the prefix page tests store routes for.
@@ -114,12 +113,12 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) {
err = db.UpsertLiveRouteBatch([]*database.LiveRoute{ err = db.UpsertLiveRouteBatch([]*database.LiveRoute{
{ {
ID: uuid.New(), Prefix: "198.51.100.0/24", MaskLength: 24, IPVersion: 4, Prefix: "198.51.100.0/24", MaskLength: 24, IPVersion: 4,
OriginASN: 64500, PeerIP: "192.0.2.1", ASPath: []int{64500}, OriginASN: 64500, PeerIP: "192.0.2.1", ASPath: []int{64500},
NextHop: "192.0.2.1", LastUpdated: ts, NextHop: "192.0.2.1", LastUpdated: ts,
}, },
{ {
ID: uuid.New(), Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6, Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6,
OriginASN: 64501, PeerIP: testPeerV6, ASPath: []int{64501}, OriginASN: 64501, PeerIP: testPeerV6, ASPath: []int{64501},
NextHop: testPeerV6, LastUpdated: ts, NextHop: testPeerV6, LastUpdated: ts,
}, },
@@ -204,12 +203,12 @@ func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
err = db.UpsertLiveRouteBatch([]*database.LiveRoute{ err = db.UpsertLiveRouteBatch([]*database.LiveRoute{
{ {
ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4,
OriginASN: 64500, PeerIP: testPeerV4, ASPath: []int{64500}, OriginASN: 64500, PeerIP: testPeerV4, ASPath: []int{64500},
NextHop: testPeerV4, LastUpdated: ts, NextHop: testPeerV4, LastUpdated: ts,
}, },
{ {
ID: uuid.New(), Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6, Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6,
OriginASN: 64501, PeerIP: testPeerV6, ASPath: []int{64501}, OriginASN: 64501, PeerIP: testPeerV6, ASPath: []int{64501},
NextHop: testPeerV6, LastUpdated: ts, NextHop: testPeerV6, LastUpdated: ts,
}, },
@@ -287,12 +286,12 @@ func TestPrefixDetailShowsOriginNames(t *testing.T) {
err = db.UpsertLiveRouteBatch([]*database.LiveRoute{ err = db.UpsertLiveRouteBatch([]*database.LiveRoute{
{ {
ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4,
OriginASN: origins[0], PeerIP: testPeerV4, OriginASN: origins[0], PeerIP: testPeerV4,
ASPath: []int{origins[0]}, NextHop: testPeerV4, LastUpdated: ts, ASPath: []int{origins[0]}, NextHop: testPeerV4, LastUpdated: ts,
}, },
{ {
ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4,
OriginASN: origins[1], PeerIP: "198.51.100.2", OriginASN: origins[1], PeerIP: "198.51.100.2",
ASPath: []int{origins[1]}, NextHop: "198.51.100.2", LastUpdated: ts, ASPath: []int{origins[1]}, NextHop: "198.51.100.2", LastUpdated: ts,
}, },
@@ -347,7 +346,7 @@ func TestPagesSayOriginZeroIsNoKnownOrigin(t *testing.T) {
t.Cleanup(func() { _ = db.Close() }) t.Cleanup(func() { _ = db.Close() })
err = db.UpsertLiveRouteBatch([]*database.LiveRoute{{ err = db.UpsertLiveRouteBatch([]*database.LiveRoute{{
ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4,
OriginASN: 0, PeerIP: testPeerV4, ASPath: []int{64500, 64501, 64502}, OriginASN: 0, PeerIP: testPeerV4, ASPath: []int{64500, 64501, 64502},
NextHop: testPeerV4, LastUpdated: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC), NextHop: testPeerV4, LastUpdated: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
}}) }})