Files
routewatch/internal/database/database_internal_test.go
T
clawbot 71ee2aeda7
check / check (push) Waiting to run
Log the start-up checkpoint's figures under their own names (closes #117)
Initialize scanned the row of PRAGMA wal_checkpoint(TRUNCATE), which is
the busy flag, the frames in the WAL and the frames copied, into
wal_pages, checkpointed and moved; and a completed TRUNCATE reads 0 for
both counts. The line now logs busy, wal_frames and checkpointed_frames,
plus wal_bytes, the WAL file's size, which New reads before it opens the
database because the open writes a frame to the WAL. The line is a
warning when busy is 1. Tests open a database a second time while the
first stays open, with and without a read open on the first, and check
the line's level and figures.

Model: opus-5-5
2026-10-20 04:32:08 +02:00

2183 lines
65 KiB
Go

package database
import (
"bytes"
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"log/slog"
"math/rand/v2"
"net/netip"
"os"
"slices"
"strconv"
"strings"
"sync"
"testing"
"time"
"git.eeqj.de/sneak/routewatch/internal/config"
"git.eeqj.de/sneak/routewatch/internal/logger"
"github.com/mattn/go-sqlite3"
)
// tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in
// memory"; the DSN change must leave temp_store below this so they spill to disk.
const tempStoreMemory = 2
// autoVacuumIncremental is the PRAGMA auto_vacuum value meaning "incremental".
const autoVacuumIncremental = 2
// vacuumTestRoutes is how many routes the vacuum test writes and then deletes,
// enough to leave many free pages in the file.
const vacuumTestRoutes = 2000
// heldConnections is how many pooled connections the pragma test holds open at
// 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"
// Parameters for the checkpoint-contention regression test.
const (
// contentionIterations is how many batch writes race the checkpoint loop.
contentionIterations = 400
// contendedASNCount is the small set of ASNs the batches reuse, so most
// batches update existing rows and exercise the read-before-write path.
contendedASNCount = 16
// asnSecondBand offsets a second ASN per batch so each batch writes more
// than one row.
asnSecondBand = 100
)
// Parameters for the tests of maintenance and the database lock.
const (
// lockMessageBuffer is how many lock messages a test database's logger
// holds before logging waits for the test to read them.
lockMessageBuffer = 64
// operationKey is the log attribute that names the operation taking the
// lock.
operationKey = "operation"
// testBatchWrite is the operation name the tests take the lock under.
testBatchWrite = "test batch write"
// lockWaitTimeout is the context timeout of a maintenance call that the
// tests let run out while it waits for the lock.
lockWaitTimeout = time.Millisecond
// lockTestASN is the AS the batch write in the lock tests stores.
lockTestASN = 64500
// readerCheckpointLimit is how long a checkpoint, and a batch write
// waiting for it, may take while a reader is open: well under the busy
// timeout, which a checkpoint that waits for the reader takes.
readerCheckpointLimit = time.Second
// truncateCheckpointLimit is how long a checkpoint of a WAL file over
// sqliteWALSizeLimitBytes may take to give up while a reader is open: its
// wait for the reader plus a second, well under the busy timeout.
truncateCheckpointLimit = truncateBusyTimeoutMs*time.Millisecond +
time.Second
)
// Parameters for the prefix-length sample tests.
const (
// sampleLimit is how many prefixes the tests ask the sample for.
sampleLimit = 10
// sampledPrefixCount is how many prefixes the sampling test stores, more
// than sampleLimit so the sample has to choose among them.
sampledPrefixCount = 30
// secondV4PeerIP and secondV6PeerIP are the second peer each prefix is
// learned from, besides the one mkV4Route and mkV6Route set.
secondV4PeerIP = "192.0.2.2"
secondV6PeerIP = "2001:db8::2"
// largeSamplePrefixCount is how many prefixes the test of a mask length
// with more than maxPrefixesToShuffle prefixes stores: twice that, so about
// half of them come after the first maxPrefixesToShuffle+1 in text order.
largeSamplePrefixCount = 2 * maxPrefixesToShuffle
// largeSampleLimit is how many prefixes that test asks for. Each random
// lookup lands after the first maxPrefixesToShuffle+1 in text order about
// half the time, so none of them doing so is about as likely as
// largeSampleLimit coin tosses all coming up heads.
largeSampleLimit = 30
// liveRoutesV4Table and liveRoutesV6Table are the live route tables the
// tests pass to the sampling steps.
liveRoutesV4Table = "live_routes_v4"
liveRoutesV6Table = "live_routes_v6"
)
// TestGetIPInfoFindsMostSpecificLiveRoute stores nested live prefixes for both
// families and checks that a lookup returns the most specific one covering the
// address, ErrNoRoute when none covers it, and the next less specific prefix
// once the only route of the most specific one is withdrawn.
func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
// Nested live prefixes, each originated by its own AS.
origins := map[string]int{
"10.0.0.0/8": 64500,
"10.1.0.0/16": 64501,
"10.1.2.0/24": 64502,
"2001:db8::/32": 64500,
"2001:db8:1::/48": 64501,
"2001:db8:1:2::/64": 64502,
}
writeOriginRoutes(t, db, origins)
// lookup checks that ip resolves to the live prefix want, or to ErrNoRoute
// when want is empty.
lookup := func(ip, want string) {
t.Helper()
info, err := db.GetIPInfo(ip)
if want == "" {
if !errors.Is(err, ErrNoRoute) {
t.Errorf("GetIPInfo(%s) = %+v, %v; want ErrNoRoute", ip, info, err)
}
return
}
if err != nil {
t.Errorf("GetIPInfo(%s): %v", ip, err)
return
}
if info.Netblock != want || info.MaskLength != netip.MustParsePrefix(want).Bits() ||
info.ASN != origins[want] {
t.Errorf("GetIPInfo(%s) = %s (mask %d) AS%d, want %s AS%d",
ip, info.Netblock, info.MaskLength, info.ASN, want, origins[want])
}
}
lookup("10.1.2.3", "10.1.2.0/24")
lookup("::ffff:10.1.2.3", "10.1.2.0/24")
lookup("10.1.3.4", "10.1.0.0/16")
lookup("10.2.0.1", "10.0.0.0/8")
lookup("192.0.2.1", "")
lookup("2001:db8:1:2::3", "2001:db8:1:2::/64")
lookup("2001:db8:1:3::4", "2001:db8:1::/48")
lookup("2001:db8:2::1", "2001:db8::/32")
lookup("2001:db9::1", "")
err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
{Prefix: "10.1.2.0/24", PeerIP: testPeerIP, IPVersion: ipVersionV4},
{Prefix: "2001:db8:1:2::/64", PeerIP: testPeerIP, IPVersion: ipVersionV6},
})
if err != nil {
t.Fatalf("DeleteLiveRouteBatch: %v", err)
}
lookup("10.1.2.3", "10.1.0.0/16")
lookup("2001:db8:1:2::3", "2001:db8:1::/48")
}
// TestGetIPInfoIgnoresShortRoutes checks that the IP lookup does not report a
// default route, or a route shorter than /8 for IPv4 or /16 for IPv6, as the
// route for an address, and that a route of exactly /8 or /16, or longer, still
// is. The prefixes are written out so that a wrong limit fails the test.
func TestGetIPInfoIgnoresShortRoutes(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
// announce writes one live route for prefix, originated by asn.
announce := func(prefix string, asn int) {
t.Helper()
writeOriginRoutes(t, db, map[string]int{prefix: asn})
}
// lookup checks that ip resolves to the live prefix netblock from asn.
lookup := func(ip, netblock string, asn int) {
t.Helper()
info, err := db.GetIPInfo(ip)
if err != nil {
t.Errorf("GetIPInfo(%s): %v", ip, err)
return
}
if info.Netblock != netblock || info.ASN != asn {
t.Errorf("GetIPInfo(%s) = %s AS%d, want %s AS%d",
ip, info.Netblock, info.ASN, netblock, asn)
}
}
// Default and very short routes, and routes one bit shorter than each limit.
announce("2000::/3", 64500)
announce("0.0.0.0/0", 64500)
announce("2000::/15", 64500)
announce("192.0.0.0/7", 64500)
for _, ip := range []string{"2001:db8::1", "192.0.2.1"} {
info, err := db.GetIPInfo(ip)
if !errors.Is(err, ErrNoRoute) {
t.Errorf("GetIPInfo(%s) with only short routes = %+v, %v; "+
"want ErrNoRoute", ip, info, err)
}
}
// Routes of exactly the minimum length are used.
announce("2001::/16", 64501)
announce("192.0.0.0/8", 64501)
lookup("2001:db8::1", "2001::/16", 64501)
lookup("192.0.2.1", "192.0.0.0/8", 64501)
announce("2001:db8::/32", 64502)
lookup("2001:db8::1", "2001:db8::/32", 64502)
}
// TestGetIPInfoNeedsNoWHOISRefreshForOriginZero checks that the IP lookup asks
// for no WHOIS lookup when the address's route has origin 0, no known origin,
// and still asks for one when its origin AS has no WHOIS data.
func TestGetIPInfoNeedsNoWHOISRefreshForOriginZero(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() }()
writeOriginRoutes(t, db, map[string]int{"10.0.0.0/8": 0, "11.0.0.0/8": 64500})
for ip, want := range map[string]bool{"10.0.0.1": false, "11.0.0.1": true} {
info, err := db.GetIPInfo(ip)
if err != nil {
t.Fatalf("GetIPInfo(%s): %v", ip, err)
}
if info.NeedsWHOISRefresh != want {
t.Errorf("GetIPInfo(%s) asks for a WHOIS lookup of AS%d: %t, want %t",
ip, info.ASN, info.NeedsWHOISRefresh, want)
}
}
}
// TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer checks that each live route
// table refuses a second route from a peer for a prefix, even with another
// origin AS. An upsert updates the peer's route before it would insert one, so
// only an insert straight into the table shows what the table allows.
func TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer(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 otherOriginASN = 64501
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
for _, tt := range []struct {
route *LiveRoute
insertSQL string
}{
{
route: mkV4Route(t, "198.51.100.0/24", 64500, ts),
insertSQL: insertLiveRouteV4SQL,
},
{
route: mkV6Route(testV6Prefix, 64500, ts),
insertSQL: insertLiveRouteV6SQL,
},
} {
err := db.UpsertLiveRoute(tt.route)
if err != nil {
t.Fatalf("UpsertLiveRoute(%s): %v", tt.route.Prefix, err)
}
_, err = db.db.ExecContext(t.Context(), tt.insertSQL,
tt.route.Prefix, tt.route.MaskLength, otherOriginASN,
tt.route.PeerIP, "[]", tt.route.NextHop, ts)
var sqliteErr sqlite3.Error
if !errors.As(err, &sqliteErr) ||
sqliteErr.ExtendedCode != sqlite3.ErrConstraintUnique {
t.Errorf("second route from peer %s for %s: error %v, "+
"want a UNIQUE constraint error",
tt.route.PeerIP, tt.route.Prefix, err)
}
}
}
// writeOriginRoutes writes one live route from testPeerIP for each prefix in
// origins, originated by the AS it maps to.
func writeOriginRoutes(t *testing.T, db *Database, origins map[string]int) {
t.Helper()
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
routes := make([]*LiveRoute, 0, len(origins))
for prefix, asn := range origins {
routes = append(routes, &LiveRoute{
Prefix: prefix,
MaskLength: netip.MustParsePrefix(prefix).Bits(),
IPVersion: detectIPVersion(prefix),
OriginASN: asn,
PeerIP: testPeerIP,
ASPath: []int{asn},
NextHop: testPeerIP,
LastUpdated: ts,
})
}
err := db.UpsertLiveRouteBatch(routes)
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
}
// TestConnectionPoolPragmas holds several pooled connections open at once and
// checks each one carries the per-connection settings from the DSN, plus the
// process-wide heap limits, and sees the new file with auto_vacuum
// incremental.
func TestConnectionPoolPragmas(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ctx := context.Background()
// Hold distinct connections open simultaneously so the pool must open a new
// one (each parsing the DSN) rather than hand back the same connection.
conns := make([]*sql.Conn, 0, heldConnections)
defer func() {
for _, c := range conns {
_ = c.Close()
}
}()
for i := range heldConnections {
c, err := db.db.Conn(ctx)
if err != nil {
t.Fatalf("failed to open connection %d: %v", i, err)
}
conns = append(conns, c)
}
for i, c := range conns {
cacheSize := readPragma(ctx, t, c, i, "cache_size")
if cacheSize != sqliteReadCacheSizeKiB {
t.Errorf("conn %d: cache_size = %d, want %d",
i, cacheSize, sqliteReadCacheSizeKiB)
}
busyTimeout := readPragma(ctx, t, c, i, "busy_timeout")
if busyTimeout != sqliteBusyTimeoutMs {
t.Errorf("conn %d: busy_timeout = %d, want %d",
i, busyTimeout, sqliteBusyTimeoutMs)
}
tempStore := readPragma(ctx, t, c, i, "temp_store")
if tempStore == tempStoreMemory {
t.Errorf("conn %d: temp_store = %d, want anything but %d (MEMORY)",
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",
i, hardHeapLimit, sqliteHardHeapLimitBytes)
}
autoVacuum := readPragma(ctx, t, c, i, "auto_vacuum")
if autoVacuum != autoVacuumIncremental {
t.Errorf("conn %d: auto_vacuum = %d, want %d (incremental)",
i, autoVacuum, autoVacuumIncremental)
}
}
}
// 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)
}
}
// TestWritingConnectionWALSizeLimit checks that the connection kept for
// writing, the one that writes the WAL, has the WAL size limit, which cuts the
// WAL file back when SQLite starts the WAL again from the beginning.
func TestWritingConnectionWALSizeLimit(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() }()
var limit int64
err = db.writer.QueryRowContext(t.Context(), "PRAGMA journal_size_limit").
Scan(&limit)
if err != nil {
t.Fatalf("failed to read the writing connection's journal_size_limit: %v",
err)
}
if limit != sqliteWALSizeLimitBytes {
t.Errorf("writing connection: journal_size_limit = %d, want %d",
limit, sqliteWALSizeLimitBytes)
}
}
// 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,
) int64 {
t.Helper()
var value int64
err := c.QueryRowContext(ctx, "PRAGMA "+name).Scan(&value)
if err != nil {
t.Fatalf("conn %d: failed to read %s: %v", i, name, err)
}
return value
}
// TestVacuumReturnsFreePages checks that after routes are deleted from a new
// database, one Vacuum call returns every page they used. The deletes leave
// fewer free pages than the 1000 Vacuum frees per call, so none may remain.
// With auto_vacuum off (issue https://git.eeqj.de/sneak/routewatch/issues/43)
// the free pages stayed in the file, and with the PRAGMA run by ExecContext
// Vacuum freed only one page per call.
func TestVacuumReturnsFreePages(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ctx := context.Background()
ts := time.Now().UTC()
const asn = 64500
routes := make([]*LiveRoute, 0, vacuumTestRoutes)
deletions := make([]LiveRouteDeletion, 0, vacuumTestRoutes)
for i := range vacuumTestRoutes {
route := mkV4Route(t, fmt.Sprintf("10.%d.%d.0/24", i/256, i%256), asn, ts)
routes = append(routes, route)
deletions = append(deletions, LiveRouteDeletion{
Prefix: route.Prefix,
PeerIP: route.PeerIP,
IPVersion: ipVersionV4,
})
}
err = db.UpsertLiveRouteBatch(routes)
if err != nil {
t.Fatalf("failed to write routes: %v", err)
}
err = db.DeleteLiveRouteBatch(deletions)
if err != nil {
t.Fatalf("failed to delete routes: %v", err)
}
var before int
err = db.db.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&before)
if err != nil {
t.Fatalf("failed to read freelist_count: %v", err)
}
if before == 0 {
t.Fatalf("no free pages after deleting %d routes", vacuumTestRoutes)
}
err = db.Vacuum(ctx)
if err != nil {
t.Fatalf("Vacuum failed: %v", err)
}
var after int
err = db.db.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&after)
if err != nil {
t.Fatalf("failed to read freelist_count: %v", err)
}
if after != 0 {
t.Errorf("free pages after Vacuum = %d of %d, want 0", after, before)
}
}
// TestBatchWriteDuringCheckpoint runs checkpoints back to back while batch
// writes run and checks that every batch write succeeds: a checkpoint holds the
// database lock as a batch write does, and checkpoints must not keep batch
// writes from getting it.
func TestBatchWriteDuringCheckpoint(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ctx, cancel := context.WithCancel(context.Background())
var wg sync.WaitGroup
wg.Go(func() {
for {
select {
case <-ctx.Done():
return
default:
_ = db.Checkpoint(ctx) // errors are the checkpoint's own to absorb
}
}
})
ts := time.Now().UTC()
seen := SeenTimes{First: ts, Last: ts}
for i := range contentionIterations {
asns := map[int]SeenTimes{
i % contendedASNCount: seen,
(i % contendedASNCount) + asnSecondBand: seen,
}
err := db.GetOrCreateASNBatch(asns)
if err != nil {
cancel()
wg.Wait()
t.Fatalf("batch write failed under checkpoint contention: %v", err)
}
}
cancel()
wg.Wait()
}
// lockMessage is a debug message the database lock logged, and the name of the
// operation that logged it.
type lockMessage struct {
text string
operation string
}
// lockMessages is a log handler that passes on each message the database lock
// logs. The lock tests read it to know that an operation waits for the lock or
// holds it, instead of guessing how long that takes.
type lockMessages chan lockMessage
// Enabled lets every message through, so the lock's debug messages arrive.
func (m lockMessages) Enabled(context.Context, slog.Level) bool { return true }
// Handle passes on a message that names an operation and drops the others.
func (m lockMessages) Handle(_ context.Context, record slog.Record) error {
record.Attrs(func(attr slog.Attr) bool {
if attr.Key != operationKey {
return true
}
m <- lockMessage{text: record.Message, operation: attr.Value.String()}
return false
})
return nil
}
// WithAttrs returns m: the tests need no attributes added.
func (m lockMessages) WithAttrs([]slog.Attr) slog.Handler { return m }
// WithGroup returns m: the tests need no groups.
func (m lockMessages) WithGroup(string) slog.Handler { return m }
// newDatabaseWithLockMessages opens a new database whose logger passes the lock's
// messages to the returned channel.
func newDatabaseWithLockMessages(t *testing.T) (*Database, lockMessages) {
t.Helper()
messages := make(lockMessages, lockMessageBuffer)
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, &logger.Logger{Logger: slog.New(messages)})
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
return db, messages
}
// waitForLockMessage waits until operation logs text, and fails the test if
// returned delivers first: the call running operation returned without logging
// text.
func waitForLockMessage(
t *testing.T,
messages lockMessages,
text, operation string,
returned <-chan error,
) {
t.Helper()
for {
select {
case message := <-messages:
if message.text == text && message.operation == operation {
return
}
case err := <-returned:
t.Fatalf("%s returned (error: %v) without logging %q",
operation, err, text)
}
}
}
// TestVacuumWaitsForBatchWrite holds the database lock as a batch write does and
// checks that Vacuum waits for it and then succeeds. Vacuum ran without the
// lock, met the batch writes at SQLite's write lock and failed with "database
// is locked" (https://git.eeqj.de/sneak/routewatch/issues/94).
func TestVacuumWaitsForBatchWrite(t *testing.T) {
t.Parallel()
db, messages := newDatabaseWithLockMessages(t)
db.lock(testBatchWrite)
vacuumed := make(chan error, 1)
go func() { vacuumed <- db.Vacuum(context.Background()) }()
waitForLockMessage(t, messages, lockWaitMessage, "Vacuum", vacuumed)
db.unlock()
err := <-vacuumed
if err != nil {
t.Fatalf("Vacuum after the batch write: %v", err)
}
}
// TestBatchWriteWaitsForAnalyze keeps Analyze waiting for SQLite's write lock,
// which another connection holds, and checks that a batch write started
// meanwhile waits for Analyze to give up the database lock, that both then
// succeed, and that Analyze stored statistics. ANALYZE ran without the database
// lock, and a batch write that waited longer than the busy timeout for it failed
// with "database is locked".
func TestBatchWriteWaitsForAnalyze(t *testing.T) {
t.Parallel()
db, messages := newDatabaseWithLockMessages(t)
ctx := context.Background()
now := time.Now().UTC()
asns := map[int]SeenTimes{lockTestASN: {First: now, Last: now}}
// A stored AS gives ANALYZE a table to analyze and statistics to write.
err := db.GetOrCreateASNBatch(asns)
if err != nil {
t.Fatalf("failed to store an AS: %v", err)
}
// With _txlock=immediate this transaction takes SQLite's write lock at
// BEGIN, so ANALYZE has to wait for it while Analyze holds the database lock.
writer, err := db.db.BeginTx(ctx, nil)
if err != nil {
t.Fatalf("failed to begin the transaction ANALYZE waits for: %v", err)
}
defer func() { _ = writer.Rollback() }()
analyzed := make(chan error, 1)
go func() { analyzed <- db.Analyze(ctx) }()
waitForLockMessage(t, messages, lockHeldMessage, "Analyze", analyzed)
written := make(chan error, 1)
go func() { written <- db.GetOrCreateASNBatch(asns) }()
waitForLockMessage(t, messages, lockWaitMessage, "GetOrCreateASNBatch", written)
err = writer.Rollback()
if err != nil {
t.Fatalf("failed to end the transaction ANALYZE waits for: %v", err)
}
err = <-analyzed
if err != nil {
t.Errorf("Analyze: %v", err)
}
err = <-written
if err != nil {
t.Errorf("batch write after Analyze: %v", err)
}
var statistics int
err = db.db.QueryRowContext(ctx,
"SELECT count(*) FROM sqlite_stat1 WHERE tbl = 'asns'").Scan(&statistics)
if err != nil {
t.Fatalf("failed to read the statistics: %v", err)
}
if statistics == 0 {
t.Error("Analyze stored no statistics for the asns table")
}
}
// TestCheckpointLockWaitEndsWithContext holds the database lock while the
// context of a waiting Checkpoint times out, and checks that Checkpoint then
// returns the context's error and leaves the lock free.
func TestCheckpointLockWaitEndsWithContext(t *testing.T) {
t.Parallel()
db, messages := newDatabaseWithLockMessages(t)
db.lock(testBatchWrite)
ctx, cancel := context.WithTimeout(context.Background(), lockWaitTimeout)
defer cancel()
checkpointed := make(chan error, 1)
go func() { checkpointed <- db.Checkpoint(ctx) }()
waitForLockMessage(t, messages, lockWaitMessage, "Checkpoint", checkpointed)
<-ctx.Done()
db.unlock()
err := <-checkpointed
if !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("Checkpoint = %v, want %v", err, context.DeadlineExceeded)
}
if !db.mu.TryLock() {
t.Fatal("Checkpoint kept the database lock after its context ended")
}
db.mu.Unlock()
}
// TestCheckpointDoesNotWaitForReader keeps a read transaction on
// live_routes_v4 open while a route is committed, then starts Checkpoint and a
// batch write together. It checks that both end within readerCheckpointLimit,
// whichever of them takes the database lock first, and that Checkpoint reports
// the part of the WAL the reader still needs as left uncopied. A TRUNCATE
// checkpoint waited the busy timeout for the reader while holding the lock
// (https://git.eeqj.de/sneak/routewatch/issues/102).
func TestCheckpointDoesNotWaitForReader(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 := context.Background()
ts := time.Now().UTC()
err = db.UpsertLiveRouteBatch(
[]*LiveRoute{mkV4Route(t, "198.51.100.0/24", lockTestASN, ts)})
if err != nil {
t.Fatalf("failed to store the route the reader reads: %v", err)
}
reader, err := db.db.Conn(ctx)
if err != nil {
t.Fatalf("failed to open the reading connection: %v", err)
}
defer func() { _ = reader.Close() }()
// A plain BEGIN starts a transaction that only reads, unlike the BEGIN
// IMMEDIATE of _txlock. It keeps the snapshot its first read takes until
// it ends.
_, err = reader.ExecContext(ctx, "BEGIN")
if err != nil {
t.Fatalf("failed to begin the read transaction: %v", err)
}
defer func() { _, _ = reader.ExecContext(ctx, "ROLLBACK") }()
var routes int
err = reader.QueryRowContext(ctx,
"SELECT count(*) FROM live_routes_v4").Scan(&routes)
if err != nil {
t.Fatalf("failed to read the live routes: %v", err)
}
err = db.UpsertLiveRouteBatch(
[]*LiveRoute{mkV4Route(t, "203.0.113.0/24", lockTestASN, ts)})
if err != nil {
t.Fatalf("failed to store a route after the read: %v", err)
}
started := time.Now()
checkpointed := make(chan error, 1)
go func() { checkpointed <- db.Checkpoint(ctx) }()
// A route write, not an AS write: the first new AS loads the embedded AS
// names, which takes seconds under -race.
route := mkV4Route(t, "192.0.2.0/24", lockTestASN, ts)
written := make(chan error, 1)
go func() { written <- db.UpsertLiveRouteBatch([]*LiveRoute{route}) }()
err = <-checkpointed
if !errors.Is(err, ErrCheckpointIncomplete) {
t.Errorf("Checkpoint = %v with a reader open, want %v",
err, ErrCheckpointIncomplete)
}
took := time.Since(started)
if took >= readerCheckpointLimit {
t.Errorf("Checkpoint took %v with a reader open, want under %v",
took, readerCheckpointLimit)
}
err = <-written
if err != nil {
t.Errorf("batch write during Checkpoint: %v", err)
}
took = time.Since(started)
if took >= readerCheckpointLimit {
t.Errorf("batch write during Checkpoint ended after %v, want under %v",
took, readerCheckpointLimit)
}
}
// TestCheckpointEmptiesWALOnceNoReadIsOpen fills the WAL file past
// sqliteWALSizeLimitBytes while the test holds a read open. With the read
// open, it checks that Checkpoint gives up within truncateCheckpointLimit and
// reports the checkpoint incomplete, and that a batch write still succeeds.
// After the test closes the read, it checks that the next Checkpoint brings
// the WAL file under the limit and gives the writing connection its busy
// timeout back (https://git.eeqj.de/sneak/routewatch/issues/102).
func TestCheckpointEmptiesWALOnceNoReadIsOpen(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()
reader, err := db.db.Conn(ctx)
if err != nil {
t.Fatalf("failed to open the reading connection: %v", err)
}
defer func() { _ = reader.Close() }()
// A plain BEGIN starts a transaction that only reads, unlike the BEGIN
// IMMEDIATE of _txlock. It keeps the snapshot its first read takes until
// it ends, so no checkpoint can copy what is written after that read.
_, err = reader.ExecContext(ctx, "BEGIN")
if err != nil {
t.Fatalf("failed to begin the read transaction: %v", err)
}
var routes int
err = reader.QueryRowContext(ctx,
"SELECT count(*) FROM live_routes_v4").Scan(&routes)
if err != nil {
t.Fatalf("failed to read the live routes: %v", err)
}
fillWALPastLimit(ctx, t, db)
started := time.Now()
err = db.Checkpoint(ctx)
if !errors.Is(err, ErrCheckpointIncomplete) {
t.Errorf("Checkpoint = %v with a read open, want %v",
err, ErrCheckpointIncomplete)
}
took := time.Since(started)
if took >= truncateCheckpointLimit {
t.Errorf("Checkpoint took %v with a read open, want under %v",
took, truncateCheckpointLimit)
}
err = db.UpsertLiveRouteBatch([]*LiveRoute{
mkV4Route(t, "192.0.2.0/24", lockTestASN, time.Now().UTC()),
})
if err != nil {
t.Errorf("batch write after Checkpoint gave up: %v", err)
}
_, err = reader.ExecContext(ctx, "ROLLBACK")
if err != nil {
t.Fatalf("failed to end the read transaction: %v", err)
}
err = db.Checkpoint(ctx)
if err != nil {
t.Errorf("Checkpoint with no read open: %v", err)
}
size := walFileSize(t, db)
if size >= sqliteWALSizeLimitBytes {
t.Errorf("WAL file of %d bytes after Checkpoint with no read open, "+
"want under %d", size, sqliteWALSizeLimitBytes)
}
busyTimeout := readPragma(ctx, t, db.writer, 0, "busy_timeout")
if busyTimeout != sqliteBusyTimeoutMs {
t.Errorf("writing connection: busy_timeout = %d after the checkpoints, "+
"want %d", busyTimeout, sqliteBusyTimeoutMs)
}
}
// fillWALPastLimit writes sqliteWALSizeLimitBytes of random bytes to a table
// of the test's own, which with SQLite's page and frame headers leaves the WAL
// file of db over the limit, and checks that it does.
func fillWALPastLimit(ctx context.Context, t *testing.T, db *Database) {
t.Helper()
_, err := db.writer.ExecContext(ctx, "CREATE TABLE wal_fill (data BLOB)")
if err != nil {
t.Fatalf("failed to create the table the test writes to: %v", err)
}
_, err = db.writer.ExecContext(ctx,
"INSERT INTO wal_fill VALUES (randomblob(?))", sqliteWALSizeLimitBytes)
if err != nil {
t.Fatalf("failed to fill the WAL: %v", err)
}
size := walFileSize(t, db)
if size <= sqliteWALSizeLimitBytes {
t.Fatalf("WAL file of %d bytes after the test's write, want over %d",
size, sqliteWALSizeLimitBytes)
}
}
// walFileSize returns the size of the WAL file of db.
func walFileSize(t *testing.T, db *Database) int64 {
t.Helper()
wal, err := os.Stat(db.path + walFileSuffix)
if err != nil {
t.Fatalf("failed to read the WAL file's size: %v", err)
}
return wal.Size()
}
// TestStartupCheckpointLogsItsFigures opens a new database, writes a route,
// and opens the same file again while the first stays open, so that its WAL
// file stays. The first open finds no WAL file: it logs a size of 0 and no
// warning. The second logs the size of the WAL file the first left, and the
// checkpoint's busy flag and frame counts, each under its own name; the
// counts read 0 once the checkpoint has emptied the file.
// Initialize logged the busy flag as wal_pages and the counts under each
// other's names (https://git.eeqj.de/sneak/routewatch/issues/117).
func TestStartupCheckpointLogsItsFigures(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, lines := openLogged(t, cfg)
for _, line := range lines {
if line[slog.LevelKey] == slog.LevelWarn.String() {
t.Errorf("opening a new database logged a warning: %v", line)
}
}
line := findLine(t, lines, startupCheckpointMessage)
if line["wal_bytes"] != 0.0 {
t.Errorf("new database: start-up checkpoint logged wal_bytes = %v, "+
"want 0", line["wal_bytes"])
}
err := db.UpsertLiveRouteBatch([]*LiveRoute{
mkV4Route(t, "192.0.2.0/24", lockTestASN, time.Now().UTC()),
})
if err != nil {
t.Fatalf("failed to store a route: %v", err)
}
walBytes := walFileSize(t, db)
if walBytes == 0 {
t.Fatal("WAL file empty after the route write, want frames in it")
}
_, lines = openLogged(t, cfg)
line = findLine(t, lines, startupCheckpointMessage)
for _, key := range []string{"busy", "wal_frames", "checkpointed_frames"} {
if line[key] != 0.0 {
t.Errorf("start-up checkpoint logged %s = %v, want 0", key, line[key])
}
}
if line["wal_bytes"] != float64(walBytes) {
t.Errorf("start-up checkpoint logged wal_bytes = %v, want the %d "+
"bytes the first database left", line["wal_bytes"], walBytes)
}
}
// TestStartupCheckpointWarnsWhenIncomplete keeps a read open on a database
// while a route is written after the read began, then opens the same file
// again. The second open's start-up checkpoint cannot copy the frames written
// after the read began, and gives up once the busy timeout has passed. It
// checks that the checkpoint is logged as a warning with busy 1 and fewer
// frames copied than the WAL holds.
func TestStartupCheckpointWarnsWhenIncomplete(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, _ := openLogged(t, cfg)
ctx := context.Background()
reader, err := db.db.Conn(ctx)
if err != nil {
t.Fatalf("failed to open the reading connection: %v", err)
}
defer func() { _ = reader.Close() }()
// A plain BEGIN starts a transaction that only reads, unlike the BEGIN
// IMMEDIATE of _txlock. It keeps the snapshot its first read takes until
// it ends.
_, err = reader.ExecContext(ctx, "BEGIN")
if err != nil {
t.Fatalf("failed to begin the read transaction: %v", err)
}
defer func() { _, _ = reader.ExecContext(ctx, "ROLLBACK") }()
var routes int
err = reader.QueryRowContext(ctx,
"SELECT count(*) FROM live_routes_v4").Scan(&routes)
if err != nil {
t.Fatalf("failed to read the live routes: %v", err)
}
err = db.UpsertLiveRouteBatch([]*LiveRoute{
mkV4Route(t, "192.0.2.0/24", lockTestASN, time.Now().UTC()),
})
if err != nil {
t.Fatalf("failed to store a route after the read: %v", err)
}
_, lines := openLogged(t, cfg)
line := findLine(t, lines, startupCheckpointIncompleteMessage)
if line[slog.LevelKey] != slog.LevelWarn.String() {
t.Errorf("start-up checkpoint logged at level %v, want %s",
line[slog.LevelKey], slog.LevelWarn)
}
if line["busy"] != 1.0 {
t.Errorf("start-up checkpoint logged busy = %v, want 1", line["busy"])
}
walFrames, _ := line["wal_frames"].(float64)
checkpointedFrames, _ := line["checkpointed_frames"].(float64)
if checkpointedFrames >= walFrames {
t.Errorf("start-up checkpoint logged checkpointed_frames = %v and "+
"wal_frames = %v with a read open, want fewer copied than the WAL "+
"holds", line["checkpointed_frames"], line["wal_frames"])
}
}
// openLogged opens the database of cfg with a logger that writes JSON, and
// returns it with the lines it logged while it opened, decoded.
func openLogged(t *testing.T, cfg *config.Config) (*Database, []map[string]any) {
t.Helper()
var logged bytes.Buffer
db, err := New(cfg,
&logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logged, nil))})
if err != nil {
t.Fatalf("failed to open the database: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
var lines []map[string]any
decoder := json.NewDecoder(&logged)
for decoder.More() {
var line map[string]any
err := decoder.Decode(&line)
if err != nil {
t.Fatalf("failed to decode the log: %v", err)
}
lines = append(lines, line)
}
return db, lines
}
// findLine returns the first of lines logged with message.
func findLine(
t *testing.T,
lines []map[string]any,
message string,
) map[string]any {
t.Helper()
for _, line := range lines {
if line[slog.MessageKey] == message {
return line
}
}
t.Fatalf("no line %q in the log", message)
return nil
}
// TestGetRandomPrefixesByLengthListsPrefixOnce stores one prefix of each
// family learned from two peers in the same second and checks that the
// prefix-length sample lists each of them once.
func TestGetRandomPrefixesByLengthListsPrefixOnce(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
v4First := mkV4Route(t, "198.51.100.0/24", 64500, ts)
v4Second := mkV4Route(t, "198.51.100.0/24", 64500, ts)
v4Second.PeerIP = secondV4PeerIP
v6First := mkV6Route(testV6Prefix, 64500, ts)
v6Second := mkV6Route(testV6Prefix, 64500, ts)
v6Second.PeerIP = secondV6PeerIP
err = db.UpsertLiveRouteBatch(
[]*LiveRoute{v4First, v4Second, v6First, v6Second})
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
for _, want := range []*LiveRoute{v4First, v6First} {
routes, err := db.GetRandomPrefixesByLength(want.MaskLength,
want.IPVersion, sampleLimit)
if err != nil {
t.Fatalf("GetRandomPrefixesByLength(%d, %d): %v",
want.MaskLength, want.IPVersion, err)
}
if len(routes) != 1 || routes[0].Prefix != want.Prefix {
t.Errorf("GetRandomPrefixesByLength(%d, %d) = %+v, want %s listed once",
want.MaskLength, want.IPVersion, routes, want.Prefix)
}
}
}
// TestGetRandomPrefixesByLengthListsNewestRoute stores a prefix learned from
// two peers at different times, the older route first, and checks that the
// prefix-length sample lists the newer route.
func TestGetRandomPrefixesByLengthListsNewestRoute(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
older := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
olderRoute := mkV4Route(t, "198.51.100.0/24", 64500, older)
newerRoute := mkV4Route(t, "198.51.100.0/24", 64500, older.Add(time.Minute))
newerRoute.PeerIP = secondV4PeerIP
err = db.UpsertLiveRouteBatch([]*LiveRoute{olderRoute, newerRoute})
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
routes, err := db.GetRandomPrefixesByLength(newerRoute.MaskLength,
ipVersionV4, sampleLimit)
if err != nil {
t.Fatalf("GetRandomPrefixesByLength: %v", err)
}
if len(routes) != 1 || routes[0].PeerIP != newerRoute.PeerIP {
t.Errorf("GetRandomPrefixesByLength = %+v, want only the newer route, "+
"from %s", routes, newerRoute.PeerIP)
}
}
// TestGetRandomPrefixesByLengthSamplesManyPrefixes stores more /24 prefixes
// than the sample asks for, each followed by a /16 so that routes of another
// mask length sit between the /24 rows, and each prefix learned from two peers
// in the same second. It checks both the sample the pages get, picked from the
// shuffled prefixes, and the random lookups used for a mask length with more
// than maxPrefixesToShuffle prefixes.
func TestGetRandomPrefixesByLengthSamplesManyPrefixes(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
var routes []*LiveRoute
stored := make([]string, 0, sampledPrefixCount)
for i := range sampledPrefixCount {
prefix := fmt.Sprintf("10.0.%d.0/24", i)
for _, p := range []string{prefix, fmt.Sprintf("172.%d.0.0/16", i)} {
first := mkV4Route(t, p, 64500, ts)
second := mkV4Route(t, p, 64500, ts)
second.PeerIP = secondV4PeerIP
routes = append(routes, first, second)
}
stored = append(stored, prefix)
}
err = db.UpsertLiveRouteBatch(routes)
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
slices.Sort(stored)
maskLength := routes[0].MaskLength
sample, err := db.GetRandomPrefixesByLength(maskLength, ipVersionV4, sampleLimit)
if err != nil {
t.Fatalf("GetRandomPrefixesByLength: %v", err)
}
listed := make([]string, 0, len(sample))
for _, route := range sample {
listed = append(listed, route.Prefix)
}
checkPrefixSample(t, "GetRandomPrefixesByLength", listed, stored)
looked, err := db.samplePrefixesByLength(
context.Background(), liveRoutesV4Table, maskLength, sampleLimit, nil)
if err != nil {
t.Fatalf("samplePrefixesByLength: %v", err)
}
checkPrefixSample(t, "samplePrefixesByLength", looked, stored)
}
// checkPrefixSample checks that sample lists sampleLimit prefixes, each once
// and each one of stored, and that it is not just the first sampleLimit of
// stored, which is in text order.
func checkPrefixSample(t *testing.T, name string, sample, stored []string) {
t.Helper()
if len(sample) != sampleLimit {
t.Errorf("%s listed %d prefixes, want %d", name, len(sample), sampleLimit)
}
listed := make(map[string]bool)
for _, prefix := range sample {
if listed[prefix] || !slices.Contains(stored, prefix) {
t.Errorf("%s lists %s twice or it is not one of the /24s", name, prefix)
}
listed[prefix] = true
}
if slices.Equal(slices.Sorted(slices.Values(sample)), stored[:sampleLimit]) {
t.Errorf("%s lists just the first %d prefixes in text order",
name, sampleLimit)
}
}
// TestGetRandomPrefixesByLengthSamplesPastFirstPrefixes stores more prefixes
// of one mask length than maxPrefixesToShuffle. It checks that
// firstPrefixesByLength reads just the first maxPrefixesToShuffle+1 in text
// order, and that the sample lists the prefixes asked for, each once, some of
// them after those first ones, which only the random lookups reach.
func TestGetRandomPrefixesByLengthSamplesPastFirstPrefixes(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
routes := make([]*LiveRoute, 0, largeSamplePrefixCount)
stored := make([]string, 0, largeSamplePrefixCount)
for i := range largeSamplePrefixCount {
route := mkV6Route(fmt.Sprintf("2001:%x::/32", i), 64500, ts)
routes = append(routes, route)
stored = append(stored, route.Prefix)
}
err = db.UpsertLiveRouteBatch(routes)
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
slices.Sort(stored)
firstPrefixes := stored[:maxPrefixesToShuffle+1]
maskLength := routes[0].MaskLength
read, err := db.firstPrefixesByLength(context.Background(),
liveRoutesV6Table, maskLength, maxPrefixesToShuffle+1)
if err != nil {
t.Fatalf("firstPrefixesByLength: %v", err)
}
if !slices.Equal(read, firstPrefixes) {
t.Errorf("firstPrefixesByLength read %d prefixes, want the first %d "+
"in text order", len(read), len(firstPrefixes))
}
sample, err := db.GetRandomPrefixesByLength(maskLength, ipVersionV6,
largeSampleLimit)
if err != nil {
t.Fatalf("GetRandomPrefixesByLength: %v", err)
}
if len(sample) != largeSampleLimit {
t.Errorf("GetRandomPrefixesByLength listed %d prefixes, want %d",
len(sample), largeSampleLimit)
}
listed := make(map[string]bool)
pastFirstPrefixes := 0
for _, route := range sample {
if listed[route.Prefix] {
t.Errorf("GetRandomPrefixesByLength lists %s twice", route.Prefix)
}
listed[route.Prefix] = true
if !slices.Contains(firstPrefixes, route.Prefix) {
pastFirstPrefixes++
}
}
if pastFirstPrefixes == 0 {
t.Errorf("GetRandomPrefixesByLength lists only prefixes among the first "+
"%d in text order", len(firstPrefixes))
}
}
// TestSamplePrefixesByLengthFillsUpFromShuffled stores a single prefix, so
// every random lookup lands on it, and checks that the sample still lists
// sampleLimit prefixes: the stored one, then the shuffled ones in order. The
// stored prefix is the second shuffled one, so the fill-up must skip it.
func TestSamplePrefixesByLengthFillsUpFromShuffled(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
shuffled := make([]string, 0, sampledPrefixCount)
for i := range sampledPrefixCount {
shuffled = append(shuffled, fmt.Sprintf("10.0.%d.0/24", i))
}
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
route := mkV4Route(t, shuffled[1], 64500, ts)
err = db.UpsertLiveRouteBatch([]*LiveRoute{route})
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
sample, err := db.samplePrefixesByLength(context.Background(),
liveRoutesV4Table, route.MaskLength, sampleLimit, shuffled)
if err != nil {
t.Fatalf("samplePrefixesByLength: %v", err)
}
want := append([]string{shuffled[1], shuffled[0]}, shuffled[2:sampleLimit]...)
if !slices.Equal(sample, want) {
t.Errorf("samplePrefixesByLength = %v, want %v", sample, want)
}
}
// TestGetRandomPrefixesByLengthSkipsWithdrawnRoutes withdraws a route after
// storing it, as can happen between the steps of GetRandomPrefixesByLength. It
// checks that the random lookups pick nothing for a mask length with no route
// left, and that only the prefixes that still have a route get their newest
// route listed.
func TestGetRandomPrefixesByLengthSkipsWithdrawnRoutes(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
kept := mkV4Route(t, "198.51.100.0/24", 64500, ts)
withdrawn := mkV4Route(t, "172.16.0.0/16", 64500, ts)
err = db.UpsertLiveRouteBatch([]*LiveRoute{kept, withdrawn})
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
err = db.DeleteLiveRoute(withdrawn.Prefix, withdrawn.PeerIP)
if err != nil {
t.Fatalf("DeleteLiveRoute: %v", err)
}
ctx := context.Background()
sample, err := db.samplePrefixesByLength(ctx, liveRoutesV4Table,
withdrawn.MaskLength, sampleLimit, []string{withdrawn.Prefix})
if err != nil {
t.Fatalf("samplePrefixesByLength: %v", err)
}
if len(sample) != 0 {
t.Errorf("samplePrefixesByLength = %v, want none for a mask length "+
"with no route left", sample)
}
routes, err := db.newestRouteOfEachPrefix(ctx, liveRoutesV4Table,
ipVersionV4, []string{withdrawn.Prefix, kept.Prefix})
if err != nil {
t.Fatalf("newestRouteOfEachPrefix: %v", err)
}
if len(routes) != 1 || routes[0].Prefix != kept.Prefix {
t.Errorf("newestRouteOfEachPrefix = %+v, want only the route of %s",
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)
}
}
}
}
// TestGetASDetailsListsEachPrefixOnce stores two IPv4 prefixes and an IPv6
// prefix an AS originates, two of them from two peers at different times, and
// a prefix another AS originates. The AS details list each of the AS's
// prefixes once, IPv4 first and in text order, with its mask length and the
// time of its newest route.
func TestGetASDetailsListsEachPrefixOnce(t *testing.T) {
t.Parallel()
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
const asn, otherASN = 64500, 64501
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
_, err = db.GetOrCreateASN(asn, ts)
if err != nil {
t.Fatalf("GetOrCreateASN: %v", err)
}
v4Older := mkV4Route(t, "192.0.2.0/24", asn, ts)
v4Newer := mkV4Route(t, "192.0.2.0/24", asn, ts.Add(time.Minute))
v4Newer.PeerIP = secondV4PeerIP
v4Single := mkV4Route(t, "198.51.100.128/25", asn, ts)
v6Newer := mkV6Route(testV6Prefix, asn, ts.Add(time.Minute))
v6Older := mkV6Route(testV6Prefix, asn, ts)
v6Older.PeerIP = secondV6PeerIP
other := mkV4Route(t, "203.0.113.0/24", otherASN, ts)
err = db.UpsertLiveRouteBatch([]*LiveRoute{
v4Older, v4Newer, v4Single, v6Newer, v6Older, other,
})
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
_, prefixes, err := db.GetASDetails(asn)
if err != nil {
t.Fatalf("GetASDetails: %v", err)
}
want := []*LiveRoute{v4Newer, v4Single, v6Newer}
listedOnce := slices.EqualFunc(prefixes, want,
func(got LiveRoute, want *LiveRoute) bool {
return got.Prefix == want.Prefix &&
got.MaskLength == want.MaskLength &&
got.IPVersion == want.IPVersion &&
got.OriginASN == asn &&
got.LastUpdated.Equal(want.LastUpdated)
})
if !listedOnce {
t.Errorf("GetASDetails listed %+v, want the prefixes and times of %+v",
prefixes, want)
}
}
// TestASPrefixQueriesUseCoveringIndexes checks the plan of each query that
// lists the prefixes an AS originates: a search of a covering index and no
// temporary B-tree, so the list reads one index entry per prefix, not one
// table row per route. Grouping the AS's routes by prefix read every route's
// row and sorted them in a temporary B-tree
// (https://git.eeqj.de/sneak/routewatch/issues/104).
func TestASPrefixQueriesUseCoveringIndexes(t *testing.T) {
t.Parallel()
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
tests := []struct {
query string
args []any
index string
}{
{nextOriginPrefixV4SQL, []any{0, ""}, "idx_live_routes_v4_origin_prefix"},
{newestRouteTimeV4SQL, []any{""}, "idx_live_routes_v4_prefix_last_updated"},
{nextOriginPrefixV6SQL, []any{0, ""}, "idx_live_routes_v6_origin_prefix"},
{newestRouteTimeV6SQL, []any{""}, "idx_live_routes_v6_prefix_last_updated"},
}
for _, tt := range tests {
plan := queryPlan(t, db, tt.query, tt.args...)
if !strings.Contains(plan, "USING COVERING INDEX "+tt.index+" ") ||
strings.Contains(plan, "TEMP B-TREE") {
t.Errorf("plan of %s:\n%s\nwant a search of covering index %s "+
"and no temporary B-tree", tt.query, plan, tt.index)
}
}
}
// queryPlan returns the steps of EXPLAIN QUERY PLAN for query, one per line.
func queryPlan(t *testing.T, db *Database, query string, args ...any) string {
t.Helper()
rows, err := db.db.QueryContext(t.Context(), "EXPLAIN QUERY PLAN "+query,
args...)
if err != nil {
t.Fatalf("EXPLAIN QUERY PLAN %s: %v", query, err)
}
defer func() { _ = rows.Close() }()
var steps []string
for rows.Next() {
var (
id, parent, unused int
detail string
)
err := rows.Scan(&id, &parent, &unused, &detail)
if err != nil {
t.Fatalf("reading the plan of %s: %v", query, err)
}
steps = append(steps, detail)
}
err = rows.Err()
if err != nil {
t.Fatalf("reading the plan of %s: %v", query, err)
}
return strings.Join(steps, "\n")
}
// TestUpdatePrefixesBatchKeepsFirstSeen writes a prefix, then a batch with it
// 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)
}
}
}
// TestUpdatePeerBatchStoresTheEarliestAndLatestTimesOfANewPeer writes a new
// peer whose updates span 5 seconds. Its row is first seen at the earliest time
// and last seen at the latest.
func TestUpdatePeerBatchStoresTheEarliestAndLatestTimesOfANewPeer(
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() }()
first := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
last := first.Add(5 * time.Second)
err = db.UpdatePeerBatch(map[string]PeerUpdate{
testPeerIP: {
PeerIP: testPeerIP, PeerASN: 64500, MessageType: "UPDATE",
Seen: SeenTimes{First: first, Last: last},
},
})
if err != nil {
t.Fatalf("UpdatePeerBatch: %v", err)
}
var firstSeen, lastSeen time.Time
err = db.db.QueryRowContext(t.Context(),
"SELECT first_seen, last_seen FROM bgp_peers WHERE peer_ip = ?",
testPeerIP).Scan(&firstSeen, &lastSeen)
if err != nil {
t.Fatalf("reading peer %s: %v", testPeerIP, err)
}
if !firstSeen.Equal(first) || !lastSeen.Equal(last) {
t.Errorf("peer %s: first seen %s, last seen %s; want %s, %s",
testPeerIP, firstSeen, lastSeen, first, last)
}
}
// 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)
}
}
// Parameters for BenchmarkGetASDetailsWithFiveHundredPeers: the AS originates
// benchASPrefixes prefixes and every one of benchASPeers peers has a route for
// each, 2.5 million routes in all.
const (
benchASPrefixes = 5000
benchASPeers = 500
)
// BenchmarkGetASDetailsWithFiveHundredPeers times the read behind the AS page
// for an AS that originates 5,000 IPv4 prefixes, each from 500 peers. It
// builds that database first, one peer at a time, which takes some minutes and
// about a GB of disk under TMPDIR. It is run by hand, not by make test, for
// example in the test image:
//
// go test -run '^$' -bench GetASDetails -benchtime 10x ./internal/database/
func BenchmarkGetASDetailsWithFiveHundredPeers(b *testing.B) {
db, err := New(&config.Config{StateDir: b.TempDir()}, logger.New())
if err != nil {
b.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
_, err = db.GetOrCreateASN(benchFirstASN, ts)
if err != nil {
b.Fatalf("GetOrCreateASN: %v", err)
}
for peer := range benchASPeers {
peerIP := fmt.Sprintf("198.18.%d.%d", peer/256, peer%256)
routes := make([]*LiveRoute, 0, benchASPrefixes)
for prefix := range benchASPrefixes {
routes = append(routes, &LiveRoute{
Prefix: fmt.Sprintf("%d.%d.%d.0/%d",
1+prefix>>16, (prefix>>8)%256, prefix%256, benchMaskLength),
MaskLength: benchMaskLength,
IPVersion: ipVersionV4,
OriginASN: benchFirstASN,
PeerIP: peerIP,
ASPath: []int{benchFirstASN + 1 + peer, benchFirstASN},
NextHop: peerIP,
LastUpdated: ts,
})
}
writeBenchBatch(b, db, routes, nil, ts)
}
for b.Loop() {
_, prefixes, err := db.GetASDetails(benchFirstASN)
if err != nil {
b.Fatalf("GetASDetails: %v", err)
}
if len(prefixes) != benchASPrefixes {
b.Fatalf("GetASDetails listed %d prefixes, want %d",
len(prefixes), benchASPrefixes)
}
}
}