next into main #87

Open
clawbot wants to merge 45 commits from next into main
10 changed files with 708 additions and 253 deletions
Showing only changes of commit d84dd56460 - Show all commits
+14 -7
View File
@@ -117,7 +117,7 @@ routewatch/
- Signal handlers registered for graceful shutdown
2. **Initialization** (`internal/routewatch/app.go`)
- Database created with SQLite schema (WAL mode, 3GB cache)
- Database created with SQLite schema (WAL mode, page caches as in "Memory")
- Message handlers registered with the streamer
- HTTP server started on configured port
@@ -170,8 +170,8 @@ routewatch/
```sql
-- Core tables
asns(id, number, handle, description, first_seen, last_seen)
prefixes_v4(id, prefix, mask_length, first_seen, last_seen)
prefixes_v6(id, prefix, mask_length, first_seen, last_seen)
prefixes_v4(prefix, first_seen)
prefixes_v6(prefix, first_seen)
-- Live routing tables (one per IP version)
live_routes_v4(id, prefix, mask_length, origin_asn, peer_ip, as_path,
@@ -215,8 +215,12 @@ data it tracks. The image sets ceilings that keep it inside a 5 GiB container.
Budget:
- 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
connection, 10 connections) and a 1.5 GiB hard heap limit for the C library.
- SQLite: 1 GiB of page cache for the one connection every write uses, which
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
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
@@ -226,6 +230,9 @@ Budget:
database writes are already serialized.
- 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:
```bash
@@ -244,8 +251,8 @@ What happens at each limit:
- Go soft limit: as the heap approaches `GOMEMLIMIT`, the runtime runs garbage
collection more aggressively rather than growing further.
- SQLite: at a 1 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
- SQLite: at the 1.5 GiB soft heap limit it recycles its page cache instead of
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
keeps running.
- Handler queues: each of the four handler queues holds at most 20,000 messages.
@@ -29,7 +29,6 @@ func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute {
}
return &LiveRoute{
ID: uuid.New(),
Prefix: prefix,
MaskLength: maskLength,
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.
func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute {
return &LiveRoute{
ID: uuid.New(),
Prefix: prefix,
MaskLength: 32,
IPVersion: ipVersionV6,
+206 -172
View File
@@ -1,25 +1,27 @@
package database
import (
"cmp"
"context"
"database/sql"
_ "embed"
"encoding/json"
"errors"
"fmt"
"maps"
"math/rand/v2"
"net"
"net/netip"
"os"
"path/filepath"
"runtime"
"slices"
"sync"
"time"
"git.eeqj.de/sneak/routewatch/internal/config"
"git.eeqj.de/sneak/routewatch/internal/logger"
"git.eeqj.de/sneak/routewatch/pkg/asinfo"
"github.com/google/uuid"
_ "github.com/mattn/go-sqlite3" // CGO SQLite driver
)
@@ -52,20 +54,31 @@ const (
minLookupMaskLengthV6 = 16
)
// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every
// pooled connection gets them; the heap limits are process-wide and set once.
// SQLite memory tuning (README "Memory"). cache_size and busy_timeout go in the
// 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 (
// sqliteCacheSizeKiB is the per-connection page cache; negative means KiB.
// -65536 = 64 MiB, so at most 640 MiB across the 10-connection pool.
sqliteCacheSizeKiB = -65536
// maxConns is the size of the connection pool: the connection kept for
// writing and 9 reading connections.
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 = 5000
// sqliteSoftHeapLimitBytes (1 GiB) makes SQLite recycle its cache rather
// than allocate once its C heap passes this size.
sqliteSoftHeapLimitBytes = 1073741824
// sqliteHardHeapLimitBytes (1.5 GiB) fails a statement with SQLITE_NOMEM
// sqliteSoftHeapLimitBytes (1.5 GiB) makes SQLite recycle its cache rather
// than allocate once its C heap passes this size. It is above the caches
// together, so it does not shrink them.
sqliteSoftHeapLimitBytes = 1610612736
// sqliteHardHeapLimitBytes (2 GiB) fails a statement with SQLITE_NOMEM
// instead of growing the C heap without bound.
sqliteHardHeapLimitBytes = 1610612736
sqliteHardHeapLimitBytes = 2147483648
)
// Settings of the ANALYZE that Analyze runs.
@@ -106,7 +119,10 @@ var (
// Database manages the SQLite database connection and operations.
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
path string
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"+
"&_auto_vacuum=incremental&_journal_mode=WAL&_txlock=immediate",
dbPath,
sqliteCacheSizeKiB,
sqliteReadCacheSizeKiB,
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)
}
err = db.PingContext(context.Background())
ctx := context.Background()
err = db.PingContext(ctx)
if err != nil {
return nil, fmt.Errorf("failed to ping database: %w", err)
}
// Set connection pool parameters
// Multiple connections allow concurrent reads while writes are serialized
const maxConns = 10
db.SetMaxOpenConns(maxConns)
db.SetMaxIdleConns(maxConns)
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{
db: db, logger: logger, path: dbPath, counts: &liveCounts{},
db: db, writer: writer, logger: logger, path: dbPath,
counts: &liveCounts{},
}
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.
func (d *Database) Initialize() error {
// Set SQLite pragmas for performance. Per-connection settings (cache_size,
// synchronous, busy_timeout, journal_mode) live in the DSN; temp_store is
// left at its default so DISTINCT temp B-trees spill to disk instead of C
// heap. The heap limits below are process-wide, so setting them once here is
// enough for the whole pool.
// synchronous, busy_timeout, journal_mode) live in the DSN, and New sets the
// writing connection's cache_size; temp_store is left at its default so
// DISTINCT temp B-trees spill to disk instead of C heap. The heap limits
// below are process-wide, so setting them once here is enough for the whole
// pool.
pragmas := []string{
"PRAGMA journal_mode=WAL", // Write-Ahead Logging
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
var walPages, checkpointedPages, movedPages int
err := d.db.QueryRowContext(context.Background(),
err := d.writer.QueryRowContext(context.Background(),
"PRAGMA wal_checkpoint(TRUNCATE)",
).Scan(&walPages, &checkpointedPages, &movedPages)
if err != nil {
@@ -229,35 +260,47 @@ func (d *Database) Initialize() error {
return nil
}
// Close closes the database connection.
// Close closes the writing connection and then the connection pool.
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
// INSERT. A live route is keyed by prefix and peer, so an announcement replaces
// the peer's route for the prefix whatever its origin AS. The UPDATE's
// rows-affected count (1 for an existing key, 0 for a new one) is what lets the
// in-memory route counters stay exact without a COUNT(*). Callers hold the
// database write lock, so no other writer can insert the same key between the
// two statements. The id column is set only on INSERT, so an updated route
// keeps its original id, exactly as the previous ON CONFLICT upsert did.
// A live-route upsert sets the origin AS of the peer's route for the prefix
// when the announcement changes it, then updates the rest of the route, and
// inserts the route only when that update matched no row. A live route is keyed
// by prefix and peer, so an announcement replaces the peer's route for the
// prefix whatever its origin AS. The update's rows-affected count (1 for an
// existing key, 0 for a new one) is what lets the in-memory route counters stay
// exact without a COUNT(*). Callers hold the database write lock, so no other
// 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 (
updateLiveRouteOriginV4SQL = `UPDATE live_routes_v4 SET origin_asn = ?
WHERE prefix = ? AND peer_ip = ? AND origin_asn != ?`
updateLiveRouteV4SQL = `UPDATE live_routes_v4
SET mask_length = ?, origin_asn = ?, as_path = ?, next_hop = ?,
last_updated = ? WHERE prefix = ? AND peer_ip = ?`
SET as_path = ?, next_hop = ?, last_updated = ?
WHERE prefix = ? AND peer_ip = ?`
insertLiveRouteV4SQL = `INSERT INTO live_routes_v4
(id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
(prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated) VALUES (?, ?, ?, ?, ?, ?, ?)`
updateLiveRouteOriginV6SQL = `UPDATE live_routes_v6 SET origin_asn = ?
WHERE prefix = ? AND peer_ip = ? AND origin_asn != ?`
updateLiveRouteV6SQL = `UPDATE live_routes_v6
SET mask_length = ?, origin_asn = ?, as_path = ?, next_hop = ?,
last_updated = ? WHERE prefix = ? AND peer_ip = ?`
SET as_path = ?, next_hop = ?, last_updated = ?
WHERE prefix = ? AND peer_ip = ?`
insertLiveRouteV6SQL = `INSERT INTO live_routes_v6
(id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
(prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated) VALUES (?, ?, ?, ?, ?, ?, ?)`
)
// 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 = ?`
)
// upsertRouteRow updates a live route with the update statement of its IP
// version's table, inserting it when no row matched. It reports whether a new
// row was inserted and whether that row is the first live route for its prefix.
// upsertRouteRow updates a live route with the origin and update statements of
// its IP version's table, inserting it when no row matched. It reports whether
// a new row was inserted and whether that row is the first live route for its
// prefix.
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,
) (bool, bool, error) {
res, err := upd.ExecContext(ctx, route.MaskLength, route.OriginASN, pathJSON,
route.NextHop, route.LastUpdated, route.Prefix, route.PeerIP)
_, err := setOrigin.ExecContext(ctx, route.OriginASN, route.Prefix,
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 {
return false, false, err
}
@@ -307,9 +357,9 @@ func upsertRouteRow(
return false, false, err
}
_, err = ins.ExecContext(ctx, route.ID.String(), route.Prefix,
route.MaskLength, route.OriginASN, route.PeerIP, pathJSON,
route.NextHop, route.LastUpdated)
_, err = ins.ExecContext(ctx, route.Prefix, route.MaskLength,
route.OriginASN, route.PeerIP, pathJSON, route.NextHop,
route.LastUpdated)
if err != nil {
return false, false, err
}
@@ -327,13 +377,23 @@ func upsertRoutesInTable(
return 0, nil, nil
}
family, updateSQL, insertSQL, lookupSQL := ipv4Family, updateLiveRouteV4SQL,
insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
family, originSQL, updateSQL := ipv4Family, updateLiveRouteOriginV4SQL,
updateLiveRouteV4SQL
insertSQL, lookupSQL := insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
if ipVersion == ipVersionV6 {
family, updateSQL, insertSQL, lookupSQL = ipv6Family,
updateLiveRouteV6SQL, insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
family, originSQL, updateSQL = ipv6Family, updateLiveRouteOriginV6SQL,
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)
if err != nil {
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)
}
inserted, newPrefix, err := upsertRouteRow(ctx, upd, ins, has, route,
string(pathJSON))
inserted, newPrefix, err := upsertRouteRow(ctx, setOrigin, upd, ins, has,
route, string(pathJSON))
if err != nil {
return 0, nil, fmt.Errorf("failed to upsert route %s: %w",
route.Prefix, err)
@@ -400,9 +460,18 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
}
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
for _, route := range routes {
for _, route := range sorted {
if route.IPVersion == ipVersionV4 {
routesV4 = append(routesV4, route)
} else {
@@ -546,9 +615,17 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
}
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
for _, del := range deletions {
for _, del := range sorted {
if del.IPVersion == ipVersionV4 {
deletionsV4 = append(deletionsV4, del)
} else {
@@ -683,46 +760,11 @@ func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
return int(deletedV4 + deletedV6), nil
}
// upsertPrefix updates the last_seen time of prefix with the statements of its
// IP version's table, inserting the prefix when it is new. It reports whether
// it inserted it.
func upsertPrefix(
ctx context.Context, selectStmt, updateStmt, insertStmt *sql.Stmt,
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(
// insertPrefixesInTable inserts the prefixes, all of one IP version, that the
// prefix table of that version does not hold yet, each first seen at its time
// in prefixes. A prefix already stored is left as it is. It returns how many it
// inserted.
func insertPrefixesInTable(
ctx context.Context, tx *loggingTx, prefixes map[string]time.Time,
ipVersion int,
) (int, error) {
@@ -731,33 +773,15 @@ func upsertPrefixesInTable(
}
family := ipv4Family
selectSQL := "SELECT id FROM prefixes_v4 WHERE prefix = ?"
updateSQL := "UPDATE prefixes_v4 SET last_seen = ? WHERE prefix = ?"
insertSQL := "INSERT INTO prefixes_v4 (id, prefix, first_seen, last_seen) " +
"VALUES (?, ?, ?, ?)"
insertSQL := "INSERT INTO prefixes_v4 (prefix, first_seen) VALUES (?, ?) " +
"ON CONFLICT (prefix) DO NOTHING"
if ipVersion == ipVersionV6 {
family = ipv6Family
selectSQL = "SELECT id FROM prefixes_v6 WHERE prefix = ?"
updateSQL = "UPDATE prefixes_v6 SET last_seen = ? WHERE prefix = ?"
insertSQL = "INSERT INTO prefixes_v6 (id, prefix, first_seen, last_seen) " +
"VALUES (?, ?, ?, ?)"
insertSQL = "INSERT INTO prefixes_v6 (prefix, first_seen) VALUES (?, ?) " +
"ON CONFLICT (prefix) DO NOTHING"
}
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)
if err != nil {
return 0, fmt.Errorf("failed to prepare %s insert statement: %w",
@@ -767,14 +791,21 @@ func upsertPrefixesInTable(
var newPrefixes int
for prefix, timestamp := range prefixes {
inserted, err := upsertPrefix(ctx, selectStmt, updateStmt, insertStmt,
prefix, timestamp)
// The rows are written in key order, as UpsertLiveRouteBatch writes routes.
for _, prefix := range slices.Sorted(maps.Keys(prefixes)) {
res, err := insertStmt.ExecContext(ctx, prefix, prefixes[prefix])
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++
}
}
@@ -782,8 +813,9 @@ func upsertPrefixesInTable(
return newPrefixes, nil
}
// UpdatePrefixesBatch updates the last_seen time for multiple prefixes in a
// single transaction, the IPv4 prefixes first and then the IPv6 prefixes.
// UpdatePrefixesBatch stores, in a single transaction, the prefixes that are
// 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 {
if len(prefixes) == 0 {
return nil
@@ -811,12 +843,12 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
ctx := context.Background()
newV4, err := upsertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4)
newV4, err := insertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4)
if err != nil {
return err
}
newV6, err := upsertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6)
newV6, err := insertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6)
if err != nil {
return err
}
@@ -1047,35 +1079,15 @@ func (d *Database) GetOrCreatePrefix(
tableName = "prefixes_v6"
}
var (
p Prefix
idStr string
)
var p Prefix
query := fmt.Sprintf("SELECT id, prefix, first_seen, last_seen "+
"FROM %s WHERE prefix = ?", tableName)
query := fmt.Sprintf("SELECT prefix, first_seen FROM %s WHERE prefix = ?",
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 {
// Prefix exists, update last_seen
p.ID, _ = uuid.Parse(idStr)
// Prefix exists; the deferred rollback ends the read.
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
}
@@ -1085,12 +1097,11 @@ func (d *Database) GetOrCreatePrefix(
}
// Prefix doesn't exist, create it
p = Prefix{ID: generateUUID(), Prefix: prefix, IPVersion: ipVersion,
FirstSeen: timestamp, LastSeen: timestamp}
p = Prefix{Prefix: prefix, IPVersion: ipVersion, FirstSeen: timestamp}
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 {
return nil, err
}
@@ -1389,37 +1400,44 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
return fmt.Errorf("failed to encode AS path: %w", err)
}
updateSQL, insertSQL, lookupSQL := updateLiveRouteV4SQL,
insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
originSQL, updateSQL := updateLiveRouteOriginV4SQL, updateLiveRouteV4SQL
insertSQL, lookupSQL := insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
if route.IPVersion == ipVersionV6 {
updateSQL, insertSQL, lookupSQL = updateLiveRouteV6SQL,
insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
originSQL, updateSQL = updateLiveRouteOriginV6SQL, updateLiveRouteV6SQL
insertSQL, lookupSQL = insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
}
ctx := context.Background()
// 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.
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 {
return fmt.Errorf("failed to prepare update statement: %w", err)
}
defer func() { _ = upd.Close() }()
ins, err := d.db.PrepareContext(ctx, insertSQL)
ins, err := d.writer.PrepareContext(ctx, insertSQL)
if err != nil {
return fmt.Errorf("failed to prepare insert statement: %w", err)
}
defer func() { _ = ins.Close() }()
has, err := d.db.PrepareContext(ctx, lookupSQL)
has, err := d.writer.PrepareContext(ctx, lookupSQL)
if err != nil {
return fmt.Errorf("failed to prepare prefix lookup statement: %w", err)
}
defer func() { _ = has.Close() }()
inserted, newPrefix, err := upsertRouteRow(ctx, upd, ins, has, route,
string(pathJSON))
inserted, newPrefix, err := upsertRouteRow(ctx, setOrigin, upd, ins, has,
route, string(pathJSON))
if err != nil {
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
}
@@ -2008,7 +2026,7 @@ func (d *Database) UpdateASNWHOIS(
WHERE asn = ?
`
_, err := d.db.ExecContext(ctx, query,
_, err := d.writer.ExecContext(ctx, query,
update.ASName,
update.OrgName,
update.OrgID,
@@ -2165,7 +2183,8 @@ func (d *Database) GetIPInfoContext(
// 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
// 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 {
// Free up to 1000 pages per call (~4MB with default 4KB page size)
// This keeps each vacuum operation quick
@@ -2174,7 +2193,12 @@ func (d *Database) Vacuum(ctx context.Context) error {
d.lock("Vacuum")
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))
if err != nil {
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")
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
// 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",
analysisLimit, optimizeAllTables))
if err != nil {
@@ -2223,7 +2252,12 @@ func (d *Database) Checkpoint(ctx context.Context) error {
d.lock("Checkpoint")
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 {
return fmt.Errorf("failed to checkpoint WAL: %w", err)
}
@@ -2279,9 +2313,9 @@ func (d *Database) unlock() {
"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) {
tx, err := d.db.BeginTx(context.Background(), nil)
tx, err := d.writer.BeginTx(context.Background(), nil)
if err != nil {
return nil, err
}
+437 -6
View File
@@ -6,15 +6,16 @@ import (
"errors"
"fmt"
"log/slog"
"math/rand/v2"
"net/netip"
"slices"
"strconv"
"sync"
"testing"
"time"
"git.eeqj.de/sneak/routewatch/internal/config"
"git.eeqj.de/sneak/routewatch/internal/logger"
"github.com/google/uuid"
"github.com/mattn/go-sqlite3"
)
@@ -33,6 +34,9 @@ const vacuumTestRoutes = 2000
// once so each is a distinct SQLite connection that parsed the DSN.
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.
const testPeerIP = "192.0.2.254"
@@ -293,7 +297,7 @@ func TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer(t *testing.T) {
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.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 {
routes = append(routes, &LiveRoute{
ID: uuid.New(),
Prefix: prefix,
MaskLength: netip.MustParsePrefix(prefix).Bits(),
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
// 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.
func TestConnectionPoolPragmas(t *testing.T) {
t.Parallel()
@@ -372,9 +375,9 @@ func TestConnectionPoolPragmas(t *testing.T) {
for i, c := range conns {
cacheSize := readPragma(ctx, t, c, i, "cache_size")
if cacheSize != sqliteCacheSizeKiB {
if cacheSize != sqliteReadCacheSizeKiB {
t.Errorf("conn %d: cache_size = %d, want %d",
i, cacheSize, sqliteCacheSizeKiB)
i, cacheSize, sqliteReadCacheSizeKiB)
}
busyTimeout := readPragma(ctx, t, c, i, "busy_timeout")
@@ -389,6 +392,12 @@ func TestConnectionPoolPragmas(t *testing.T) {
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")
if hardHeapLimit != sqliteHardHeapLimitBytes {
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.
func readPragma(
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)
}
}
// 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)
// and first/last seen timestamps.
// and the time it was first seen.
type Prefix struct {
ID uuid.UUID
Prefix string
IPVersion int // 4 or 6
FirstSeen time.Time
LastSeen time.Time
}
// Announcement represents a BGP announcement or withdrawal event,
@@ -71,7 +69,6 @@ type ASNPeering struct {
//
//nolint:tagliatelle // field names are the existing JSON API
type LiveRoute struct {
ID uuid.UUID `json:"id"`
Prefix string `json:"prefix"`
MaskLength int `json:"mask_length"`
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
);
-- 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 (
id TEXT PRIMARY KEY,
prefix TEXT UNIQUE NOT NULL,
first_seen DATETIME NOT NULL,
last_seen DATETIME NOT NULL
);
prefix TEXT PRIMARY KEY,
first_seen DATETIME NOT NULL
) WITHOUT ROWID;
-- IPv6 prefixes table
-- IPv6 prefixes table, stored as prefixes_v4 is.
CREATE TABLE IF NOT EXISTS prefixes_v6 (
id TEXT PRIMARY KEY,
prefix TEXT UNIQUE NOT NULL,
first_seen DATETIME NOT NULL,
last_seen DATETIME NOT NULL
);
prefix TEXT PRIMARY KEY,
first_seen DATETIME NOT NULL
) WITHOUT ROWID;
CREATE TABLE IF NOT EXISTS announcements (
id TEXT PRIMARY KEY,
@@ -75,11 +72,6 @@ CREATE TABLE IF NOT EXISTS bgp_peers (
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_prefix_id ON announcements(prefix_id);
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_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 (
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,
mask_length INTEGER NOT NULL, -- CIDR mask length (0-32)
origin_asn INTEGER NOT NULL,
@@ -107,15 +104,15 @@ CREATE TABLE IF NOT EXISTS live_routes_v4 (
as_path TEXT NOT NULL, -- JSON array
next_hop TEXT NOT NULL,
last_updated DATETIME NOT NULL,
-- A peer has at most one route for a prefix. peer_ip comes first so that
-- this key also finds all of a peer's routes, when its BGP session goes
-- down; lookups by prefix use the prefix index below.
-- A peer has at most one route for a prefix. This key finds the route a
-- write updates or deletes and, peer_ip coming first, all of a peer's
-- routes when its BGP session goes down.
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 (
id TEXT PRIMARY KEY,
id INTEGER PRIMARY KEY,
prefix TEXT NOT NULL,
mask_length INTEGER NOT NULL, -- CIDR mask length (0-128)
origin_asn INTEGER NOT NULL,
@@ -123,24 +120,29 @@ CREATE TABLE IF NOT EXISTS live_routes_v6 (
as_path TEXT NOT NULL, -- JSON array
next_hop TEXT 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)
);
-- Indexes for live_routes_v4 table
-- Finds a prefix's newest route in one search, whatever the number of peers.
-- Indexes for both live route tables, and the queries that use each.
--
-- 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_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);
-- 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);
-- mask_length, prefix: the prefix-length pages' list of a mask length's
-- prefixes in text order, and the prefix distribution read at startup.
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_prefix ON live_routes_v4(mask_length, prefix);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_prefix ON live_routes_v6(mask_length, prefix);
-- origin_asn: the prefixes an AS originates, for the AS page and its API answer.
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_asn ON live_routes_v4(origin_asn);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_origin_asn ON live_routes_v6(origin_asn);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_last_updated ON live_routes_v6(last_updated);
-- Index to optimize prefix distribution queries
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_prefix ON live_routes_v6(mask_length, prefix);
-- last_updated: the oldest and newest route times in /api/v1/stats and
-- /status.json.
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated);
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 {
start := time.Now()
defer logSlowQuery(d.logger, query, start)
_, err := d.db.ExecContext(context.Background(), query, args...)
_, err := d.writer.ExecContext(context.Background(), query, args...)
return err
}
@@ -18,7 +18,6 @@ import (
"git.eeqj.de/sneak/routewatch/internal/ristypes"
"git.eeqj.de/sneak/routewatch/internal/server"
"git.eeqj.de/sneak/routewatch/internal/streamer"
"github.com/google/uuid"
)
// 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()
for prefix, timestamp := range prefixes {
if p, exists := m.Prefixes[prefix]; exists {
p.LastSeen = timestamp
} else {
if _, exists := m.Prefixes[prefix]; !exists {
const (
ipVersionV4 = 4
ipVersionV6 = 6
@@ -97,11 +94,9 @@ func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
}
m.Prefixes[prefix] = &database.Prefix{
ID: uuid.New(),
Prefix: prefix,
IPVersion: ipVersion,
FirstSeen: timestamp,
LastSeen: timestamp,
}
}
}
@@ -117,8 +112,6 @@ func (m *mockStore) GetOrCreatePrefix(
defer m.mu.Unlock()
if p, exists := m.Prefixes[prefix]; exists {
p.LastSeen = timestamp
return p, nil
}
@@ -133,11 +126,9 @@ func (m *mockStore) GetOrCreatePrefix(
}
p := &database.Prefix{
ID: uuid.New(),
Prefix: prefix,
IPVersion: ipVersion,
FirstSeen: timestamp,
LastSeen: timestamp,
}
m.Prefixes[prefix] = p
m.PrefixCount++
@@ -370,11 +361,9 @@ func (m *mockStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
// Track prefix
if _, exists := m.Prefixes[route.Prefix]; !exists {
m.Prefixes[route.Prefix] = &database.Prefix{
ID: uuid.New(),
Prefix: route.Prefix,
IPVersion: route.IPVersion,
FirstSeen: route.LastUpdated,
LastSeen: route.LastUpdated,
}
m.PrefixCount++
-2
View File
@@ -12,7 +12,6 @@ import (
"git.eeqj.de/sneak/routewatch/internal/logger"
"git.eeqj.de/sneak/routewatch/internal/metrics"
"git.eeqj.de/sneak/routewatch/internal/ristypes"
"github.com/google/uuid"
)
const (
@@ -461,7 +460,6 @@ func (h *PrefixHandler) createLiveRoute(update prefixUpdate) *database.LiveRoute
// Create live route record
liveRoute := &database.LiveRoute{
ID: uuid.New(),
Prefix: update.prefix,
MaskLength: maskLength,
IPVersion: ipVersion,
+7 -8
View File
@@ -19,7 +19,6 @@ import (
"git.eeqj.de/sneak/routewatch/internal/metrics"
"git.eeqj.de/sneak/routewatch/internal/streamer"
"git.eeqj.de/sneak/routewatch/pkg/asinfo"
"github.com/google/uuid"
)
// 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{
{
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},
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},
NextHop: testPeerV6, LastUpdated: ts,
},
@@ -204,12 +203,12 @@ func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
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},
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},
NextHop: testPeerV6, LastUpdated: ts,
},
@@ -287,12 +286,12 @@ func TestPrefixDetailShowsOriginNames(t *testing.T) {
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,
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",
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() })
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},
NextHop: testPeerV4, LastUpdated: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
}})