From 29394860787a48f63415b6d03cec377e9bfde5b8 Mon Sep 17 00:00:00 2001 From: sneak Date: Mon, 19 Oct 2026 13:21:42 +0000 Subject: [PATCH] Make live route writes cheaper so the prefix handler keeps up (closes #96) 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 --- README.md | 21 +- internal/database/counts_internal_test.go | 2 - internal/database/database.go | 378 ++++++++------- internal/database/database_internal_test.go | 443 +++++++++++++++++- internal/database/models.go | 5 +- internal/database/schema.sql | 78 +-- internal/database/slowquery.go | 4 +- .../app_integration_internal_test.go | 13 +- internal/routewatch/prefixhandler.go | 2 - internal/server/handlers_internal_test.go | 15 +- 10 files changed, 708 insertions(+), 253 deletions(-) diff --git a/README.md b/README.md index 84a7bce..3526890 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/internal/database/counts_internal_test.go b/internal/database/counts_internal_test.go index c5decf7..bbe5577 100644 --- a/internal/database/counts_internal_test.go +++ b/internal/database/counts_internal_test.go @@ -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, diff --git a/internal/database/database.go b/internal/database/database.go index a536e77..c4832bb 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -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 } diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index d93b188..92768c9 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -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) + } +} diff --git a/internal/database/models.go b/internal/database/models.go index 3872f95..5392a4c 100644 --- a/internal/database/models.go +++ b/internal/database/models.go @@ -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"` diff --git a/internal/database/schema.sql b/internal/database/schema.sql index ef2f5dd..6bd65a6 100644 --- a/internal/database/schema.sql +++ b/internal/database/schema.sql @@ -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); \ No newline at end of file +-- 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); \ No newline at end of file diff --git a/internal/database/slowquery.go b/internal/database/slowquery.go index d027211..13a0ac8 100644 --- a/internal/database/slowquery.go +++ b/internal/database/slowquery.go @@ -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 } diff --git a/internal/routewatch/app_integration_internal_test.go b/internal/routewatch/app_integration_internal_test.go index bdab0a4..8f094ca 100644 --- a/internal/routewatch/app_integration_internal_test.go +++ b/internal/routewatch/app_integration_internal_test.go @@ -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++ diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 4858e76..c3ae80e 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -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, diff --git a/internal/server/handlers_internal_test.go b/internal/server/handlers_internal_test.go index ff68f08..8ea16a3 100644 --- a/internal/server/handlers_internal_test.go +++ b/internal/server/handlers_internal_test.go @@ -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), }}) -- 2.54.0