Compare commits

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

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

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

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

No files matched your search

+85 -35
View File
@@ -38,21 +38,44 @@ DEBUG=routewatch ./routewatch
### Web Interface ### Web Interface
- `GET /` - Redirects to /status - `GET /` - Home page (HTML): headline figures, and lookups by AS number and by
- `GET /status` - HTML status dashboard IP address
- `GET /status.json` - JSON statistics - `GET /status` - HTML status dashboard, which reads `/api/v1/stats`
- `GET /as/{asn}` - AS detail page (HTML) - `GET /status.json` - JSON statistics: feed, memory, database and route figures
- `GET /prefix/{prefix}` - Prefix detail page (HTML) - `GET /.well-known/healthcheck.json` - Health check (JSON): 200 when the
- `GET /prefixlength/{length}` - IPv4 prefixes by mask length database answers and the RIS Live feed is connected, otherwise 503;
- `GET /prefixlength6/{length}` - IPv6 prefixes by mask length `data.status` is `ok` or `error` and `data.checks` says which check failed
- `GET /ip/{ip}` - Redirects to prefix containing the IP - `GET /as/{asn}` - AS detail page (HTML): the prefixes the AS originates and
its peers
- `GET /prefix/{prefix}/{length}` - Prefix detail page (HTML), such as
`/prefix/192.0.2.0/24`: every live route for the prefix, one per peer, and its
origin ASes
- `GET /prefixlength/{length}` - Up to 500 IPv4 prefixes with that mask length,
picked at random, each with its newest route (HTML)
- `GET /prefixlength6/{length}` - The same for IPv6
- `GET /ip/{addr}` - The same JSON answer as `/api/v1/ip/{ip}`
- `GET /ip/` - The same, for the caller's own address, or the one a proxy names
in `True-Client-IP`, `X-Real-IP` or `X-Forwarded-For`
### API v1 ### API v1
- `GET /api/v1/stats` - Detailed statistics with handler metrics - `GET /api/v1/stats` - The `/status.json` figures plus connection, garbage
- `GET /api/v1/ip/{ip}` - Look up AS information for an IP address collector, stream and handler queue statistics
- `GET /api/v1/as/{asn}` - Get prefixes announced by an AS - `GET /api/v1/ip/{ip}` - Looks up each address: its most specific live route
- `GET /api/v1/prefix/{prefix}` - Get routes for a specific prefix (`netblock`), that route's origin AS with its WHOIS details, and the address's
PTR names. `{ip}` is an IP address, a host name (each address it resolves to
is looked up), or a comma-separated list of these. `results` has one entry per
address found and `errors` one per address or name that failed; the answer is
404 when no address was found.
- `GET /api/v1/as/{asn}` - The AS and the prefixes it originates, IPv4 and IPv6
apart, each with when its newest route was updated
- `GET /api/v1/prefix/{prefix}/{length}` - Every live route for the prefix, one
per peer, and the same routes grouped by origin AS in `origins`
A JSON answer is `{"status": "ok", "data": ...}`, or for an error
`{"status": "error", "error": {"msg": ..., "code": ...}}`, and carries an
`@meta` object with the time zone, the API version and the time the request
took.
An origin AS of 0 (`origin_asn` in a route, `asn` in an IP lookup, the key `0` An origin AS of 0 (`origin_asn` in a route, `asn` in an IP lookup, the key `0`
in a prefix's `origins`) means the route has no known origin: its AS path ends in a prefix's `origins`) means the route has no known origin: its AS path ends
@@ -76,7 +99,9 @@ routewatch/
│ ├── logger/ # Structured logging wrapper │ ├── logger/ # Structured logging wrapper
│ ├── metrics/ # Performance metrics tracking │ ├── metrics/ # Performance metrics tracking
│ ├── config/ # Configuration management │ ├── config/ # Configuration management
│ └── templates/ # HTML templates │ ├── templates/ # HTML templates
│ ├── version/ # Build version information
│ └── whois/ # WHOIS lookups of AS details
└── pkg/ └── pkg/
└── asinfo/ # AS information lookup (public API) └── asinfo/ # AS information lookup (public API)
``` ```
@@ -95,7 +120,7 @@ routewatch/
│ │ Streamer │───▶│ Handlers │───▶│ Database │ │ │ │ Streamer │───▶│ Handlers │───▶│ Database │ │
│ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │
│ │ RIS Live │ │ - ASHandler │ │ SQLite with │ │ │ │ RIS Live │ │ - ASHandler │ │ SQLite with │ │
│ │ WebSocket │ │ - PeerHandler│ │ WAL mode │ │ │ │ HTTP stream │ │ - PeerHandler│ │ WAL mode │ │
│ │ client │ │ - PrefixHdlr │ │ │ │ │ │ client │ │ - PrefixHdlr │ │ │ │
│ │ │ │ - PeeringHdlr│ │ Tables: │ │ │ │ │ │ - PeeringHdlr│ │ Tables: │ │
│ └──────────────┘ └──────────────┘ │ - asns │ │ │ └──────────────┘ └──────────────┘ │ - asns │ │
@@ -117,7 +142,7 @@ routewatch/
- Signal handlers registered for graceful shutdown - Signal handlers registered for graceful shutdown
2. **Initialization** (`internal/routewatch/app.go`) 2. **Initialization** (`internal/routewatch/app.go`)
- Database created with SQLite schema (WAL mode, 3GB cache) - Database created with SQLite schema (WAL mode, page caches as in "Memory")
- Message handlers registered with the streamer - Message handlers registered with the streamer
- HTTP server started on configured port - HTTP server started on configured port
@@ -131,7 +156,10 @@ routewatch/
- Parses BGP UPDATE messages from JSON stream - Parses BGP UPDATE messages from JSON stream
- Dispatches to registered handlers based on message type - Dispatches to registered handlers based on message type
- Each handler has its own queue with backpressure handling - Each handler has its own queue with backpressure handling
- Handlers batch writes for efficiency (25K-30K ops, 1-2s timeout) - Handlers batch writes for efficiency: the AS, peer and prefix handlers
write 30,000, 10,000 and 25,000 updates at a time, or after 2, 2 and 1
seconds; the peering handler writes the peerings it has collected every 30
seconds
4. **Handler Details** 4. **Handler Details**
- **ASHandler**: Tracks all ASNs seen in AS paths - **ASHandler**: Tracks all ASNs seen in AS paths
@@ -150,7 +178,8 @@ routewatch/
### Key Design Patterns ### Key Design Patterns
- **Batched Writes**: All database operations are batched for performance - **Batched Writes**: The handlers write the feed's updates to the database in
batches for performance
- **Backpressure**: Probabilistic message dropping when queues exceed 50% - **Backpressure**: Probabilistic message dropping when queues exceed 50%
capacity capacity
- **Graceful Shutdown**: 60-second timeout, flushes all pending batches - **Graceful Shutdown**: 60-second timeout, flushes all pending batches
@@ -169,21 +198,35 @@ routewatch/
```sql ```sql
-- Core tables -- Core tables
asns(id, number, handle, description, first_seen, last_seen) asns(asn PRIMARY KEY, handle, description,
prefixes_v4(id, prefix, mask_length, first_seen, last_seen) as_name, org_name, org_id, address, country_code, abuse_email,
prefixes_v6(id, prefix, mask_length, first_seen, last_seen) abuse_phone, tech_email, tech_phone, rir, rir_registration_date,
rir_last_modified, whois_raw,
first_seen, last_seen, whois_updated_at)
prefixes_v4(prefix PRIMARY KEY, first_seen) WITHOUT ROWID
prefixes_v6(prefix PRIMARY KEY, first_seen) WITHOUT ROWID
-- Live routing tables (one per IP version) -- Live routing tables (one per IP version), one route per peer and prefix
live_routes_v4(id, prefix, mask_length, origin_asn, peer_ip, as_path, live_routes_v4(id INTEGER PRIMARY KEY, prefix, mask_length, origin_asn,
next_hop, last_updated) peer_ip, as_path, next_hop, last_updated,
live_routes_v6(id, prefix, mask_length, origin_asn, peer_ip, as_path, UNIQUE(peer_ip, prefix))
next_hop, last_updated) live_routes_v6(id INTEGER PRIMARY KEY, prefix, mask_length, origin_asn,
peer_ip, as_path, next_hop, last_updated,
UNIQUE(peer_ip, prefix))
-- Relationship tracking -- Relationship tracking
peerings(id, as_a, as_b, first_seen, last_seen) peerings(id PRIMARY KEY, as_a, as_b, first_seen, last_seen,
bgp_peers(id, peer_ip, peer_asn, last_message_type, last_seen) UNIQUE(as_a, as_b))
bgp_peers(id PRIMARY KEY, peer_ip UNIQUE, peer_asn, first_seen, last_seen,
last_message_type)
-- Nothing in the daemon writes this table
announcements(id PRIMARY KEY, prefix_id, peer_asn, origin_asn, path,
next_hop, timestamp, is_withdrawal)
``` ```
The indexes are in `internal/database/schema.sql`.
## Configuration ## Configuration
Configuration is handled via environment variables and OS-specific paths: Configuration is handled via environment variables and OS-specific paths:
@@ -201,7 +244,7 @@ non-zero exit. An empty variable counts as unset.
State directory (database location): State directory (database location):
- macOS: `~/Library/Application Support/routewatch/` - macOS: `~/Library/Application Support/berlin.sneak.app.routewatch/`
- Linux: `/var/lib/berlin.sneak.app.routewatch/` when running as root, otherwise - Linux: `/var/lib/berlin.sneak.app.routewatch/` when running as root, otherwise
`$XDG_DATA_HOME/berlin.sneak.app.routewatch/` (with `XDG_DATA_HOME` unset, `$XDG_DATA_HOME/berlin.sneak.app.routewatch/` (with `XDG_DATA_HOME` unset,
`~/.local/share/berlin.sneak.app.routewatch/`). In the Docker image this is `~/.local/share/berlin.sneak.app.routewatch/`). In the Docker image this is
@@ -215,8 +258,12 @@ data it tracks. The image sets ceilings that keep it inside a 5 GiB container.
Budget: Budget:
- Go heap: a 1.5 GiB soft limit (`GOMEMLIMIT=1536MiB`, set in the image). - Go heap: a 1.5 GiB soft limit (`GOMEMLIMIT=1536MiB`, set in the image).
- SQLite: at most 640 MiB of page cache across the connection pool (64 MiB per - SQLite: 1 GiB of page cache for the one connection every write uses, which
connection, 10 connections) and a 1.5 GiB hard heap limit for the C library. keeps the pages the route writes touch from one batch to the next, and 16 MiB
for each of the 9 reading connections, whose cache SQLite empties whenever
another connection has written: about 1.15 GiB in all. A 1.5 GiB soft heap
limit sits above the caches, so it does not shrink them, and a 2 GiB hard heap
limit caps the C library.
- glibc allocator: the SQLite C library runs on glibc `malloc`, which frees - glibc allocator: the SQLite C library runs on glibc `malloc`, which frees
page-cache chunks back to per-arena free lists rather than to the kernel, so page-cache chunks back to per-arena free lists rather than to the kernel, so
process RSS tracks the high-water mark of those arenas, not SQLite's live process RSS tracks the high-water mark of those arenas, not SQLite's live
@@ -226,6 +273,9 @@ Budget:
database writes are already serialized. database writes are already serialized.
- About 0.2 GiB for everything else in the runtime. - About 0.2 GiB for everything else in the runtime.
The ceilings add up to 3.7 GiB: the Go heap's 1.5 GiB, SQLite's 2 GiB and the
0.2 GiB for the rest.
Run the container with a memory limit of 5 GiB and swap disabled: Run the container with a memory limit of 5 GiB and swap disabled:
```bash ```bash
@@ -244,8 +294,8 @@ What happens at each limit:
- Go soft limit: as the heap approaches `GOMEMLIMIT`, the runtime runs garbage - Go soft limit: as the heap approaches `GOMEMLIMIT`, the runtime runs garbage
collection more aggressively rather than growing further. collection more aggressively rather than growing further.
- SQLite: at a 1 GiB soft heap limit it recycles its page cache instead of - SQLite: at the 1.5 GiB soft heap limit it recycles its page cache instead of
allocating more; at the 1.5 GiB hard heap limit a statement fails with an allocating more; at the 2 GiB hard heap limit a statement fails with an
out-of-memory error, and the handler logs it and drops that batch. The process out-of-memory error, and the handler logs it and drops that batch. The process
keeps running. keeps running.
- Handler queues: each of the four handler queues holds at most 20,000 messages. - Handler queues: each of the four handler queues holds at most 20,000 messages.
@@ -294,7 +344,7 @@ make fmt
make lint make lint
# Build # Build
make make build
``` ```
## Entrypoints ## Entrypoints
@@ -323,8 +373,8 @@ provide:
- `script/docker` — build the Docker image tagged via `script/projectname` - `script/docker` — build the Docker image tagged via `script/projectname`
- `script/cibuild` — CI entrypoint: runs `script/bootstrap` and `script/check`, - `script/cibuild` — CI entrypoint: runs `script/bootstrap` and `script/check`,
then builds the image then builds the image
- `script/precommit` — pre-commit gate: `go mod tidy` + `go fmt` (must not - `script/precommit` — pre-commit gate: `go mod tidy` (must not change `go.mod`
change files), then `script/check` or `go.sum`) and `go fmt` (writes), then `script/check`
- `script/install-precommit` — install the git pre-commit hook that runs - `script/install-precommit` — install the git pre-commit hook that runs
`script/precommit` `script/precommit`
+71 -13
View File
@@ -10,28 +10,88 @@
# Status # Status
pre-1.0. No git tags. The linter and the tests are phases of the Docker build, pre-1.0. No git tags. `main` last took `next` on 2026-10-06, through
and the Gitea workflow runs `script/cibuild` (the format check of the Go code https://git.eeqj.de/sneak/routewatch/pulls/6. The linter and the tests are
and the markdown, both phases, then the image) on every push. The image sets phases of the Docker build, and the Gitea workflow runs `script/cibuild` (the
memory ceilings for a 5 GiB container (README "Memory") and the README says how format check of the Go code and the markdown, both phases, then the image) on
to run it under upaas (README "Running under upaas"). A 35-hour run of `3898daa` every push. The image sets memory ceilings for a 5 GiB container (README
on the live feed peaked at about 1 GiB, without a container memory limit. "Memory") and the README says how to run it under upaas (README "Running under
upaas"). A 25-hour run of `a9b6860` on the live feed peaked at about 0.6 GiB
(https://git.eeqj.de/sneak/routewatch/issues/3). A run of `825afa2` on the live
feed found the prefix handler dropping about a quarter of its updates because
its route writes were slower than the feed
(https://git.eeqj.de/sneak/routewatch/issues/96); the cheaper writes since then
are measured by a benchmark, not yet on the live feed.
# Next Step # Next Step
`next` waits for sneak to merge it to `main` through `next` waits for sneak to merge it to `main` through
https://git.eeqj.de/sneak/routewatch/pulls/6. After that, setting routewatch up https://git.eeqj.de/sneak/routewatch/pulls/87.
under upaas on fsn1app1 and deploying it are his
(https://git.eeqj.de/sneak/routewatch/issues/31), and so is the run under a real
5 GiB limit (https://git.eeqj.de/sneak/routewatch/issues/3).
# Completed Steps # Completed Steps
- 2026-10-19: README "HTTP Endpoints" lists every route with what it answers,
"Database Schema" matches `schema.sql`, and the rest of the README matches the
code; this file brought up to date (closes #72)
- 2026-10-19: route writes are cheaper: the live route and prefix tables lose
their random UUID keys and the prefix tables their duplicate index, a route
update sets only the columns that change, and every write goes through one
connection with a page cache of 1 GiB (README "Memory"). A benchmark writes
the prefix handler's largest batch faster than the feed sends it (closes #96)
- 2026-10-19: the WAL checkpoint, the incremental vacuum and ANALYZE take the
database lock every batch write holds, so a vacuum no longer fails with
"database is locked" and an ANALYZE no longer makes a batch write lose its
updates; ANALYZE runs as `PRAGMA optimize` with an analysis limit (closes #94)
- 2026-10-19: shutdown stops the streamer first, and each handler writes
everything left on its queue before it stops, so the updates waiting there are
no longer lost (closes #69)
- 2026-10-19: a path that ends in an AS set of several members has origin AS 0,
no known origin, instead of the set's last member, and the set adds no
peerings; the pages show such a route as having no known origin (closes #75)
- 2026-10-19: `bgp_peer_count` drops to 0 when the last peer goes down (closes
#92)
- 2026-10-19: `bgp_peer_count` counts every peer the feed has heard from, not
only those whose session opened since startup, and drops a peer whose session
goes down (closes #74)
- 2026-10-19: the shared files are re-vendored from `sneak/prompts` at - 2026-10-19: the shared files are re-vendored from `sneak/prompts` at
`0653579`: golangci-lint v2.14.0 and its configuration, lint and test as `0653579`: golangci-lint v2.14.0 and its configuration, lint and test as
`Dockerfile` phases that keep Go's and golangci-lint's caches, markdown `Dockerfile` phases that keep Go's and golangci-lint's caches, markdown
formatted with prettier, agent guidance in `AGENTS.md`; the code is fixed for formatted with prettier, agent guidance in `AGENTS.md`; the code is fixed for
what the new linter configuration finds (closes #52) what the new linter configuration finds (closes #52)
- 2026-10-19: the streamer reads feed lines up to 1 MiB, so a line over 64 KiB
no longer ends the connection (closes #76)
- 2026-10-19: a peer's live routes are removed when its BGP session goes down;
`STATE` messages are never dropped for backpressure, and one that finds its
queue full waits for room ahead of the handler's other messages (closes #56)
- 2026-10-19: the prefix-length pages pick their sample with a bounded number of
index lookups instead of reading every route of that length, and list each
prefix once (closes #66)
- 2026-10-19: a request past its time limit answers 504 instead of 408, and
`/api/v1/stats` answers 504 instead of 500 when its read runs out of time
(closes #70)
- 2026-10-19: the AS, peer and prefix handlers write each batch outside the lock
their message handling needs, so messages keep coming off their queues during
a write; meanwhile the next batch grows to at most ten batch sizes (closes
#65)
- 2026-10-19: a live route is identified by prefix and peer, so an announcement
with a new origin AS replaces the peer's route and a withdrawal removes it
whatever its origin AS (closes #55)
- 2026-10-14: the streamer reconnects when RIS Live ends the stream or sends
nothing for 60 seconds, and the health check reports the feed disconnected
meanwhile (closes #53)
- 2026-10-14: the prefix page takes the names of the origin ASes from the prefix
query instead of reading every route of each origin AS (closes #67)
- 2026-10-14: the daemon exits with status 1 when its web server cannot listen
on its port, instead of running on without it (closes #73)
- 2026-10-14: the prefix pages parse the prefix before looking it up: one that
does not parse answers 400, and any valid way of writing a prefix finds its
routes (closes #71)
- 2026-10-14: the IP lookup ignores routes shorter than /8 for IPv4 and /16 for
IPv6, so a default route is no longer named as an address's owner (closes #57)
- 2026-10-14: a prefix batch keeps each peer's last update for a prefix, instead
of one update per prefix across all peers (closes #54)
- 2026-10-14: `reconnect_count` in `/api/v1/stats` counts every reconnect
(closes #68)
- 2026-10-03: `/api/v1/stats` serves the prefix distribution from memory, seeded - 2026-10-03: `/api/v1/stats` serves the prefix distribution from memory, seeded
at startup and adjusted on every live-route write, so a request no longer at startup and adjusted on every live-route write, so a request no longer
reads every live route (closes #30) reads every live route (closes #30)
@@ -116,6 +176,4 @@ under upaas on fsn1app1 and deploying it are his
# Future Steps # Future Steps
- Production memory under 5 GiB: whether to test under a real 5 GiB container None open.
limit on fsn1app1 is open for sneak
(https://git.eeqj.de/sneak/routewatch/issues/3)
@@ -29,7 +29,6 @@ func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute {
} }
return &LiveRoute{ return &LiveRoute{
ID: uuid.New(),
Prefix: prefix, Prefix: prefix,
MaskLength: maskLength, MaskLength: maskLength,
IPVersion: ipVersionV4, IPVersion: ipVersionV4,
@@ -44,7 +43,6 @@ func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute {
// mkV6Route builds an IPv6 live route. // mkV6Route builds an IPv6 live route.
func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute { func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute {
return &LiveRoute{ return &LiveRoute{
ID: uuid.New(),
Prefix: prefix, Prefix: prefix,
MaskLength: 32, MaskLength: 32,
IPVersion: ipVersionV6, IPVersion: ipVersionV6,
+206 -172
View File
@@ -1,25 +1,27 @@
package database package database
import ( import (
"cmp"
"context" "context"
"database/sql" "database/sql"
_ "embed" _ "embed"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"maps"
"math/rand/v2" "math/rand/v2"
"net" "net"
"net/netip" "net/netip"
"os" "os"
"path/filepath" "path/filepath"
"runtime" "runtime"
"slices"
"sync" "sync"
"time" "time"
"git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/config"
"git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/logger"
"git.eeqj.de/sneak/routewatch/pkg/asinfo" "git.eeqj.de/sneak/routewatch/pkg/asinfo"
"github.com/google/uuid"
_ "github.com/mattn/go-sqlite3" // CGO SQLite driver _ "github.com/mattn/go-sqlite3" // CGO SQLite driver
) )
@@ -52,20 +54,31 @@ const (
minLookupMaskLengthV6 = 16 minLookupMaskLengthV6 = 16
) )
// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every // SQLite memory tuning (README "Memory"). cache_size and busy_timeout go in the
// pooled connection gets them; the heap limits are process-wide and set once. // DSN so every pooled connection gets them; the connection kept for writing
// then sets its own cache size. The heap limits are process-wide and set once.
const ( const (
// sqliteCacheSizeKiB is the per-connection page cache; negative means KiB. // maxConns is the size of the connection pool: the connection kept for
// -65536 = 64 MiB, so at most 640 MiB across the 10-connection pool. // writing and 9 reading connections.
sqliteCacheSizeKiB = -65536 maxConns = 10
// sqliteReadCacheSizeKiB is the page cache of each reading connection;
// negative means KiB. -16384 = 16 MiB, at most 144 MiB for the 9. A reading
// connection's cache is emptied whenever another connection has written
// since its last read, many times a minute, so a large one is wasted.
sqliteReadCacheSizeKiB = -16384
// sqliteWriteCacheSizeKiB is the page cache of the connection kept for
// writing, -1048576 = 1 GiB. It keeps its pages from one write to the next,
// so the index pages the route writes touch stay in memory.
sqliteWriteCacheSizeKiB = -1048576
// sqliteBusyTimeoutMs is how long a connection waits on a locked database. // sqliteBusyTimeoutMs is how long a connection waits on a locked database.
sqliteBusyTimeoutMs = 5000 sqliteBusyTimeoutMs = 5000
// sqliteSoftHeapLimitBytes (1 GiB) makes SQLite recycle its cache rather // sqliteSoftHeapLimitBytes (1.5 GiB) makes SQLite recycle its cache rather
// than allocate once its C heap passes this size. // than allocate once its C heap passes this size. It is above the caches
sqliteSoftHeapLimitBytes = 1073741824 // together, so it does not shrink them.
// sqliteHardHeapLimitBytes (1.5 GiB) fails a statement with SQLITE_NOMEM sqliteSoftHeapLimitBytes = 1610612736
// sqliteHardHeapLimitBytes (2 GiB) fails a statement with SQLITE_NOMEM
// instead of growing the C heap without bound. // instead of growing the C heap without bound.
sqliteHardHeapLimitBytes = 1610612736 sqliteHardHeapLimitBytes = 2147483648
) )
// Settings of the ANALYZE that Analyze runs. // Settings of the ANALYZE that Analyze runs.
@@ -106,7 +119,10 @@ var (
// Database manages the SQLite database connection and operations. // Database manages the SQLite database connection and operations.
type Database struct { type Database struct {
db *sql.DB db *sql.DB
// writer is the connection kept for writing, taken from db when the
// database opens. Every write uses it while holding mu.
writer *sql.Conn
logger *logger.Logger logger *logger.Logger
path string path string
mu sync.Mutex mu sync.Mutex
@@ -144,7 +160,7 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
"file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d"+ "file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d"+
"&_auto_vacuum=incremental&_journal_mode=WAL&_txlock=immediate", "&_auto_vacuum=incremental&_journal_mode=WAL&_txlock=immediate",
dbPath, dbPath,
sqliteCacheSizeKiB, sqliteReadCacheSizeKiB,
sqliteBusyTimeoutMs, sqliteBusyTimeoutMs,
) )
@@ -153,20 +169,34 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
return nil, fmt.Errorf("failed to open database: %w", err) return nil, fmt.Errorf("failed to open database: %w", err)
} }
err = db.PingContext(context.Background()) ctx := context.Background()
err = db.PingContext(ctx)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to ping database: %w", err) return nil, fmt.Errorf("failed to ping database: %w", err)
} }
// Set connection pool parameters // Set connection pool parameters
// Multiple connections allow concurrent reads while writes are serialized // Multiple connections allow concurrent reads while writes are serialized
const maxConns = 10
db.SetMaxOpenConns(maxConns) db.SetMaxOpenConns(maxConns)
db.SetMaxIdleConns(maxConns) db.SetMaxIdleConns(maxConns)
db.SetConnMaxLifetime(0) db.SetConnMaxLifetime(0)
writer, err := db.Conn(ctx)
if err != nil {
return nil, fmt.Errorf("failed to open the writing connection: %w", err)
}
_, err = writer.ExecContext(ctx,
fmt.Sprintf("PRAGMA cache_size=%d", sqliteWriteCacheSizeKiB))
if err != nil {
return nil, fmt.Errorf("failed to set the writing connection's cache: %w",
err)
}
database := &Database{ database := &Database{
db: db, logger: logger, path: dbPath, counts: &liveCounts{}, db: db, writer: writer, logger: logger, path: dbPath,
counts: &liveCounts{},
} }
err = database.Initialize() err = database.Initialize()
@@ -188,10 +218,11 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
// Initialize creates the database schema if it doesn't exist. // Initialize creates the database schema if it doesn't exist.
func (d *Database) Initialize() error { func (d *Database) Initialize() error {
// Set SQLite pragmas for performance. Per-connection settings (cache_size, // Set SQLite pragmas for performance. Per-connection settings (cache_size,
// synchronous, busy_timeout, journal_mode) live in the DSN; temp_store is // synchronous, busy_timeout, journal_mode) live in the DSN, and New sets the
// left at its default so DISTINCT temp B-trees spill to disk instead of C // writing connection's cache_size; temp_store is left at its default so
// heap. The heap limits below are process-wide, so setting them once here is // DISTINCT temp B-trees spill to disk instead of C heap. The heap limits
// enough for the whole pool. // below are process-wide, so setting them once here is enough for the whole
// pool.
pragmas := []string{ pragmas := []string{
"PRAGMA journal_mode=WAL", // Write-Ahead Logging "PRAGMA journal_mode=WAL", // Write-Ahead Logging
fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes), fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes),
@@ -208,7 +239,7 @@ func (d *Database) Initialize() error {
// Run WAL checkpoint on startup to consolidate any existing WAL // Run WAL checkpoint on startup to consolidate any existing WAL
var walPages, checkpointedPages, movedPages int var walPages, checkpointedPages, movedPages int
err := d.db.QueryRowContext(context.Background(), err := d.writer.QueryRowContext(context.Background(),
"PRAGMA wal_checkpoint(TRUNCATE)", "PRAGMA wal_checkpoint(TRUNCATE)",
).Scan(&walPages, &checkpointedPages, &movedPages) ).Scan(&walPages, &checkpointedPages, &movedPages)
if err != nil { if err != nil {
@@ -229,35 +260,47 @@ func (d *Database) Initialize() error {
return nil return nil
} }
// Close closes the database connection. // Close closes the writing connection and then the connection pool.
func (d *Database) Close() error { func (d *Database) Close() error {
return d.db.Close() return errors.Join(d.writer.Close(), d.db.Close())
} }
// A live-route upsert is an UPDATE followed, only when no row matched, by an // A live-route upsert sets the origin AS of the peer's route for the prefix
// INSERT. A live route is keyed by prefix and peer, so an announcement replaces // when the announcement changes it, then updates the rest of the route, and
// the peer's route for the prefix whatever its origin AS. The UPDATE's // inserts the route only when that update matched no row. A live route is keyed
// rows-affected count (1 for an existing key, 0 for a new one) is what lets the // by prefix and peer, so an announcement replaces the peer's route for the
// in-memory route counters stay exact without a COUNT(*). Callers hold the // prefix whatever its origin AS. The update's rows-affected count (1 for an
// database write lock, so no other writer can insert the same key between the // existing key, 0 for a new one) is what lets the in-memory route counters stay
// two statements. The id column is set only on INSERT, so an updated route // exact without a COUNT(*). Callers hold the database write lock, so no other
// keeps its original id, exactly as the previous ON CONFLICT upsert did. // writer can insert the same key between the statements.
//
// SQLite rewrites a row's entry in every index on a column an UPDATE sets, even
// when the value stays the same, so the update sets only the columns that
// change: mask_length follows from the prefix, and origin_asn has a statement
// of its own that changes nothing when the origin AS stays the same, as it
// mostly does.
const ( const (
updateLiveRouteOriginV4SQL = `UPDATE live_routes_v4 SET origin_asn = ?
WHERE prefix = ? AND peer_ip = ? AND origin_asn != ?`
updateLiveRouteV4SQL = `UPDATE live_routes_v4 updateLiveRouteV4SQL = `UPDATE live_routes_v4
SET mask_length = ?, origin_asn = ?, as_path = ?, next_hop = ?, SET as_path = ?, next_hop = ?, last_updated = ?
last_updated = ? WHERE prefix = ? AND peer_ip = ?` WHERE prefix = ? AND peer_ip = ?`
insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 insertLiveRouteV4SQL = `INSERT INTO live_routes_v4
(id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, (prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` last_updated) VALUES (?, ?, ?, ?, ?, ?, ?)`
updateLiveRouteOriginV6SQL = `UPDATE live_routes_v6 SET origin_asn = ?
WHERE prefix = ? AND peer_ip = ? AND origin_asn != ?`
updateLiveRouteV6SQL = `UPDATE live_routes_v6 updateLiveRouteV6SQL = `UPDATE live_routes_v6
SET mask_length = ?, origin_asn = ?, as_path = ?, next_hop = ?, SET as_path = ?, next_hop = ?, last_updated = ?
last_updated = ? WHERE prefix = ? AND peer_ip = ?` WHERE prefix = ? AND peer_ip = ?`
insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 insertLiveRouteV6SQL = `INSERT INTO live_routes_v6
(id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, (prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` last_updated) VALUES (?, ?, ?, ?, ?, ?, ?)`
) )
// Before a new route is inserted and after a route is deleted, the write looks // Before a new route is inserted and after a route is deleted, the write looks
@@ -278,15 +321,22 @@ const (
WHERE prefix = ? AND peer_ip = ?` WHERE prefix = ? AND peer_ip = ?`
) )
// upsertRouteRow updates a live route with the update statement of its IP // upsertRouteRow updates a live route with the origin and update statements of
// version's table, inserting it when no row matched. It reports whether a new // its IP version's table, inserting it when no row matched. It reports whether
// row was inserted and whether that row is the first live route for its prefix. // a new row was inserted and whether that row is the first live route for its
// prefix.
func upsertRouteRow( func upsertRouteRow(
ctx context.Context, upd, ins, has *sql.Stmt, route *LiveRoute, ctx context.Context, setOrigin, upd, ins, has *sql.Stmt, route *LiveRoute,
pathJSON string, pathJSON string,
) (bool, bool, error) { ) (bool, bool, error) {
res, err := upd.ExecContext(ctx, route.MaskLength, route.OriginASN, pathJSON, _, err := setOrigin.ExecContext(ctx, route.OriginASN, route.Prefix,
route.NextHop, route.LastUpdated, route.Prefix, route.PeerIP) route.PeerIP, route.OriginASN)
if err != nil {
return false, false, err
}
res, err := upd.ExecContext(ctx, pathJSON, route.NextHop, route.LastUpdated,
route.Prefix, route.PeerIP)
if err != nil { if err != nil {
return false, false, err return false, false, err
} }
@@ -307,9 +357,9 @@ func upsertRouteRow(
return false, false, err return false, false, err
} }
_, err = ins.ExecContext(ctx, route.ID.String(), route.Prefix, _, err = ins.ExecContext(ctx, route.Prefix, route.MaskLength,
route.MaskLength, route.OriginASN, route.PeerIP, pathJSON, route.OriginASN, route.PeerIP, pathJSON, route.NextHop,
route.NextHop, route.LastUpdated) route.LastUpdated)
if err != nil { if err != nil {
return false, false, err return false, false, err
} }
@@ -327,13 +377,23 @@ func upsertRoutesInTable(
return 0, nil, nil return 0, nil, nil
} }
family, updateSQL, insertSQL, lookupSQL := ipv4Family, updateLiveRouteV4SQL, family, originSQL, updateSQL := ipv4Family, updateLiveRouteOriginV4SQL,
insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL updateLiveRouteV4SQL
insertSQL, lookupSQL := insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
if ipVersion == ipVersionV6 { if ipVersion == ipVersionV6 {
family, updateSQL, insertSQL, lookupSQL = ipv6Family, family, originSQL, updateSQL = ipv6Family, updateLiveRouteOriginV6SQL,
updateLiveRouteV6SQL, insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL updateLiveRouteV6SQL
insertSQL, lookupSQL = insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
} }
setOrigin, err := tx.PrepareContext(ctx, originSQL)
if err != nil {
return 0, nil, fmt.Errorf("failed to prepare %s origin statement: %w",
family, err)
}
defer func() { _ = setOrigin.Close() }()
upd, err := tx.PrepareContext(ctx, updateSQL) upd, err := tx.PrepareContext(ctx, updateSQL)
if err != nil { if err != nil {
return 0, nil, fmt.Errorf("failed to prepare %s update statement: %w", return 0, nil, fmt.Errorf("failed to prepare %s update statement: %w",
@@ -365,8 +425,8 @@ func upsertRoutesInTable(
return 0, nil, fmt.Errorf("failed to encode AS path: %w", err) return 0, nil, fmt.Errorf("failed to encode AS path: %w", err)
} }
inserted, newPrefix, err := upsertRouteRow(ctx, upd, ins, has, route, inserted, newPrefix, err := upsertRouteRow(ctx, setOrigin, upd, ins, has,
string(pathJSON)) route, string(pathJSON))
if err != nil { if err != nil {
return 0, nil, fmt.Errorf("failed to upsert route %s: %w", return 0, nil, fmt.Errorf("failed to upsert route %s: %w",
route.Prefix, err) route.Prefix, err)
@@ -400,9 +460,18 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
} }
defer d.rollback(tx) defer d.rollback(tx)
// The rows are written in key order, peer then prefix, so consecutive writes
// land on the same index pages. The sort is stable, so of two updates of
// one route the later one is still written last.
sorted := slices.SortedStableFunc(slices.Values(routes),
func(a, b *LiveRoute) int {
return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP),
cmp.Compare(a.Prefix, b.Prefix))
})
var routesV4, routesV6 []*LiveRoute var routesV4, routesV6 []*LiveRoute
for _, route := range routes { for _, route := range sorted {
if route.IPVersion == ipVersionV4 { if route.IPVersion == ipVersionV4 {
routesV4 = append(routesV4, route) routesV4 = append(routesV4, route)
} else { } else {
@@ -546,9 +615,17 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
} }
defer d.rollback(tx) defer d.rollback(tx)
// The rows are deleted in key order, peer then prefix, as
// UpsertLiveRouteBatch writes them.
sorted := slices.SortedFunc(slices.Values(deletions),
func(a, b LiveRouteDeletion) int {
return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP),
cmp.Compare(a.Prefix, b.Prefix))
})
var deletionsV4, deletionsV6 []LiveRouteDeletion var deletionsV4, deletionsV6 []LiveRouteDeletion
for _, del := range deletions { for _, del := range sorted {
if del.IPVersion == ipVersionV4 { if del.IPVersion == ipVersionV4 {
deletionsV4 = append(deletionsV4, del) deletionsV4 = append(deletionsV4, del)
} else { } else {
@@ -683,46 +760,11 @@ func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
return int(deletedV4 + deletedV6), nil return int(deletedV4 + deletedV6), nil
} }
// upsertPrefix updates the last_seen time of prefix with the statements of its // insertPrefixesInTable inserts the prefixes, all of one IP version, that the
// IP version's table, inserting the prefix when it is new. It reports whether // prefix table of that version does not hold yet, each first seen at its time
// it inserted it. // in prefixes. A prefix already stored is left as it is. It returns how many it
func upsertPrefix( // inserted.
ctx context.Context, selectStmt, updateStmt, insertStmt *sql.Stmt, func insertPrefixesInTable(
prefix string, timestamp time.Time,
) (bool, error) {
var id string
err := selectStmt.QueryRowContext(ctx, prefix).Scan(&id)
switch {
case err == nil:
// Prefix exists, update last_seen
_, err = updateStmt.ExecContext(ctx, timestamp, prefix)
if err != nil {
return false, fmt.Errorf("failed to update prefix %s: %w",
prefix, err)
}
return false, nil
case errors.Is(err, sql.ErrNoRows):
// Prefix doesn't exist, create it
_, err = insertStmt.ExecContext(ctx, generateUUID().String(), prefix,
timestamp, timestamp)
if err != nil {
return false, fmt.Errorf("failed to insert prefix %s: %w",
prefix, err)
}
return true, nil
default:
return false, fmt.Errorf("failed to query prefix %s: %w", prefix, err)
}
}
// upsertPrefixesInTable updates the last_seen time of prefixes, all of one IP
// version, in the prefix table of that version, inserting the ones that are
// new. It returns how many it inserted.
func upsertPrefixesInTable(
ctx context.Context, tx *loggingTx, prefixes map[string]time.Time, ctx context.Context, tx *loggingTx, prefixes map[string]time.Time,
ipVersion int, ipVersion int,
) (int, error) { ) (int, error) {
@@ -731,33 +773,15 @@ func upsertPrefixesInTable(
} }
family := ipv4Family family := ipv4Family
selectSQL := "SELECT id FROM prefixes_v4 WHERE prefix = ?" insertSQL := "INSERT INTO prefixes_v4 (prefix, first_seen) VALUES (?, ?) " +
updateSQL := "UPDATE prefixes_v4 SET last_seen = ? WHERE prefix = ?" "ON CONFLICT (prefix) DO NOTHING"
insertSQL := "INSERT INTO prefixes_v4 (id, prefix, first_seen, last_seen) " +
"VALUES (?, ?, ?, ?)"
if ipVersion == ipVersionV6 { if ipVersion == ipVersionV6 {
family = ipv6Family family = ipv6Family
selectSQL = "SELECT id FROM prefixes_v6 WHERE prefix = ?" insertSQL = "INSERT INTO prefixes_v6 (prefix, first_seen) VALUES (?, ?) " +
updateSQL = "UPDATE prefixes_v6 SET last_seen = ? WHERE prefix = ?" "ON CONFLICT (prefix) DO NOTHING"
insertSQL = "INSERT INTO prefixes_v6 (id, prefix, first_seen, last_seen) " +
"VALUES (?, ?, ?, ?)"
} }
selectStmt, err := tx.PrepareContext(ctx, selectSQL)
if err != nil {
return 0, fmt.Errorf("failed to prepare %s select statement: %w",
family, err)
}
defer func() { _ = selectStmt.Close() }()
updateStmt, err := tx.PrepareContext(ctx, updateSQL)
if err != nil {
return 0, fmt.Errorf("failed to prepare %s update statement: %w",
family, err)
}
defer func() { _ = updateStmt.Close() }()
insertStmt, err := tx.PrepareContext(ctx, insertSQL) insertStmt, err := tx.PrepareContext(ctx, insertSQL)
if err != nil { if err != nil {
return 0, fmt.Errorf("failed to prepare %s insert statement: %w", return 0, fmt.Errorf("failed to prepare %s insert statement: %w",
@@ -767,14 +791,21 @@ func upsertPrefixesInTable(
var newPrefixes int var newPrefixes int
for prefix, timestamp := range prefixes { // The rows are written in key order, as UpsertLiveRouteBatch writes routes.
inserted, err := upsertPrefix(ctx, selectStmt, updateStmt, insertStmt, for _, prefix := range slices.Sorted(maps.Keys(prefixes)) {
prefix, timestamp) res, err := insertStmt.ExecContext(ctx, prefix, prefixes[prefix])
if err != nil { if err != nil {
return 0, err return 0, fmt.Errorf("failed to insert prefix %s: %w", prefix, err)
} }
if inserted { // One row for a new prefix, none for one already stored.
inserted, err := res.RowsAffected()
if err != nil {
return 0, fmt.Errorf("failed to count inserted prefix %s: %w",
prefix, err)
}
if inserted > 0 {
newPrefixes++ newPrefixes++
} }
} }
@@ -782,8 +813,9 @@ func upsertPrefixesInTable(
return newPrefixes, nil return newPrefixes, nil
} }
// UpdatePrefixesBatch updates the last_seen time for multiple prefixes in a // UpdatePrefixesBatch stores, in a single transaction, the prefixes that are
// single transaction, the IPv4 prefixes first and then the IPv6 prefixes. // not stored yet, each first seen at its time in prefixes: the IPv4 prefixes
// first and then the IPv6 prefixes.
func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
if len(prefixes) == 0 { if len(prefixes) == 0 {
return nil return nil
@@ -811,12 +843,12 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
ctx := context.Background() ctx := context.Background()
newV4, err := upsertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4) newV4, err := insertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4)
if err != nil { if err != nil {
return err return err
} }
newV6, err := upsertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6) newV6, err := insertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6)
if err != nil { if err != nil {
return err return err
} }
@@ -1047,35 +1079,15 @@ func (d *Database) GetOrCreatePrefix(
tableName = "prefixes_v6" tableName = "prefixes_v6"
} }
var ( var p Prefix
p Prefix
idStr string
)
query := fmt.Sprintf("SELECT id, prefix, first_seen, last_seen "+ query := fmt.Sprintf("SELECT prefix, first_seen FROM %s WHERE prefix = ?",
"FROM %s WHERE prefix = ?", tableName) tableName)
err = tx.QueryRow(query, prefix).Scan(&idStr, &p.Prefix, &p.FirstSeen, &p.LastSeen) err = tx.QueryRow(query, prefix).Scan(&p.Prefix, &p.FirstSeen)
if err == nil { if err == nil {
// Prefix exists, update last_seen // Prefix exists; the deferred rollback ends the read.
p.ID, _ = uuid.Parse(idStr)
p.IPVersion = ipVersion p.IPVersion = ipVersion
updateQuery := fmt.Sprintf("UPDATE %s SET last_seen = ? WHERE id = ?", tableName)
_, err = tx.Exec(updateQuery, timestamp, p.ID.String())
if err != nil {
return nil, err
}
p.LastSeen = timestamp
err = tx.Commit()
if err != nil {
d.logger.Error("Failed to commit transaction for prefix update",
"prefix", prefix, "error", err)
return nil, err
}
return &p, nil return &p, nil
} }
@@ -1085,12 +1097,11 @@ func (d *Database) GetOrCreatePrefix(
} }
// Prefix doesn't exist, create it // Prefix doesn't exist, create it
p = Prefix{ID: generateUUID(), Prefix: prefix, IPVersion: ipVersion, p = Prefix{Prefix: prefix, IPVersion: ipVersion, FirstSeen: timestamp}
FirstSeen: timestamp, LastSeen: timestamp}
insertQuery := fmt.Sprintf("INSERT INTO %s "+ insertQuery := fmt.Sprintf("INSERT INTO %s "+
"(id, prefix, first_seen, last_seen) VALUES (?, ?, ?, ?)", tableName) "(prefix, first_seen) VALUES (?, ?)", tableName)
_, err = tx.Exec(insertQuery, p.ID.String(), p.Prefix, p.FirstSeen, p.LastSeen) _, err = tx.Exec(insertQuery, p.Prefix, p.FirstSeen)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -1389,37 +1400,44 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
return fmt.Errorf("failed to encode AS path: %w", err) return fmt.Errorf("failed to encode AS path: %w", err)
} }
updateSQL, insertSQL, lookupSQL := updateLiveRouteV4SQL, originSQL, updateSQL := updateLiveRouteOriginV4SQL, updateLiveRouteV4SQL
insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL insertSQL, lookupSQL := insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
if route.IPVersion == ipVersionV6 { if route.IPVersion == ipVersionV6 {
updateSQL, insertSQL, lookupSQL = updateLiveRouteV6SQL, originSQL, updateSQL = updateLiveRouteOriginV6SQL, updateLiveRouteV6SQL
insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL insertSQL, lookupSQL = insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
} }
ctx := context.Background() ctx := context.Background()
// The write lock is held, so no other writer can insert this key between the // The write lock is held, so no other writer can insert this key between the
// update and the insert even though they are separate autocommit statements. // update and the insert even though they are separate autocommit statements.
upd, err := d.db.PrepareContext(ctx, updateSQL) setOrigin, err := d.writer.PrepareContext(ctx, originSQL)
if err != nil {
return fmt.Errorf("failed to prepare origin statement: %w", err)
}
defer func() { _ = setOrigin.Close() }()
upd, err := d.writer.PrepareContext(ctx, updateSQL)
if err != nil { if err != nil {
return fmt.Errorf("failed to prepare update statement: %w", err) return fmt.Errorf("failed to prepare update statement: %w", err)
} }
defer func() { _ = upd.Close() }() defer func() { _ = upd.Close() }()
ins, err := d.db.PrepareContext(ctx, insertSQL) ins, err := d.writer.PrepareContext(ctx, insertSQL)
if err != nil { if err != nil {
return fmt.Errorf("failed to prepare insert statement: %w", err) return fmt.Errorf("failed to prepare insert statement: %w", err)
} }
defer func() { _ = ins.Close() }() defer func() { _ = ins.Close() }()
has, err := d.db.PrepareContext(ctx, lookupSQL) has, err := d.writer.PrepareContext(ctx, lookupSQL)
if err != nil { if err != nil {
return fmt.Errorf("failed to prepare prefix lookup statement: %w", err) return fmt.Errorf("failed to prepare prefix lookup statement: %w", err)
} }
defer func() { _ = has.Close() }() defer func() { _ = has.Close() }()
inserted, newPrefix, err := upsertRouteRow(ctx, upd, ins, has, route, inserted, newPrefix, err := upsertRouteRow(ctx, setOrigin, upd, ins, has,
string(pathJSON)) route, string(pathJSON))
if err != nil { if err != nil {
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
} }
@@ -2008,7 +2026,7 @@ func (d *Database) UpdateASNWHOIS(
WHERE asn = ? WHERE asn = ?
` `
_, err := d.db.ExecContext(ctx, query, _, err := d.writer.ExecContext(ctx, query,
update.ASName, update.ASName,
update.OrgName, update.OrgName,
update.OrgID, update.OrgID,
@@ -2165,7 +2183,8 @@ func (d *Database) GetIPInfoContext(
// Vacuum runs incremental vacuum to return up to 1000 unused pages to the // Vacuum runs incremental vacuum to return up to 1000 unused pages to the
// filesystem. Like Analyze and Checkpoint, it holds the database lock, so it // filesystem. Like Analyze and Checkpoint, it holds the database lock, so it
// and the batch writes wait for each other instead of failing on SQLite's // and the batch writes wait for each other instead of failing on SQLite's
// write lock. // write lock, and returns ctx's error without running if ctx ended while it
// waited: the writing connection it uses does not check ctx itself.
func (d *Database) Vacuum(ctx context.Context) error { func (d *Database) Vacuum(ctx context.Context) error {
// Free up to 1000 pages per call (~4MB with default 4KB page size) // Free up to 1000 pages per call (~4MB with default 4KB page size)
// This keeps each vacuum operation quick // This keeps each vacuum operation quick
@@ -2174,7 +2193,12 @@ func (d *Database) Vacuum(ctx context.Context) error {
d.lock("Vacuum") d.lock("Vacuum")
defer d.unlock() defer d.unlock()
rows, err := d.db.QueryContext(ctx, err := ctx.Err()
if err != nil {
return fmt.Errorf("failed to run incremental vacuum: %w", err)
}
rows, err := d.writer.QueryContext(ctx,
fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree)) fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree))
if err != nil { if err != nil {
return fmt.Errorf("failed to run incremental vacuum: %w", err) return fmt.Errorf("failed to run incremental vacuum: %w", err)
@@ -2204,9 +2228,14 @@ func (d *Database) Analyze(ctx context.Context) error {
d.lock("Analyze") d.lock("Analyze")
defer d.unlock() defer d.unlock()
err := ctx.Err()
if err != nil {
return fmt.Errorf("failed to analyze database: %w", err)
}
// analysis_limit is a setting of the connection, so it is set in the same // analysis_limit is a setting of the connection, so it is set in the same
// call as PRAGMA optimize, which runs both statements on one connection. // call as PRAGMA optimize, which runs both statements on one connection.
_, err := d.db.ExecContext(ctx, fmt.Sprintf( _, err = d.writer.ExecContext(ctx, fmt.Sprintf(
"PRAGMA analysis_limit=%d; PRAGMA optimize=%#x", "PRAGMA analysis_limit=%d; PRAGMA optimize=%#x",
analysisLimit, optimizeAllTables)) analysisLimit, optimizeAllTables))
if err != nil { if err != nil {
@@ -2223,7 +2252,12 @@ func (d *Database) Checkpoint(ctx context.Context) error {
d.lock("Checkpoint") d.lock("Checkpoint")
defer d.unlock() defer d.unlock()
_, err := d.db.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)") err := ctx.Err()
if err != nil {
return fmt.Errorf("failed to checkpoint WAL: %w", err)
}
_, err = d.writer.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)")
if err != nil { if err != nil {
return fmt.Errorf("failed to checkpoint WAL: %w", err) return fmt.Errorf("failed to checkpoint WAL: %w", err)
} }
@@ -2279,9 +2313,9 @@ func (d *Database) unlock() {
"held_by", lockedBy, "duration_ms", holdDuration.Milliseconds()) "held_by", lockedBy, "duration_ms", holdDuration.Milliseconds())
} }
// beginTx starts a new transaction with logging // beginTx starts a new transaction on the writing connection, with logging
func (d *Database) beginTx() (*loggingTx, error) { func (d *Database) beginTx() (*loggingTx, error) {
tx, err := d.db.BeginTx(context.Background(), nil) tx, err := d.writer.BeginTx(context.Background(), nil)
if err != nil { if err != nil {
return nil, err return nil, err
} }
+437 -6
View File
@@ -6,15 +6,16 @@ import (
"errors" "errors"
"fmt" "fmt"
"log/slog" "log/slog"
"math/rand/v2"
"net/netip" "net/netip"
"slices" "slices"
"strconv"
"sync" "sync"
"testing" "testing"
"time" "time"
"git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/config"
"git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/logger"
"github.com/google/uuid"
"github.com/mattn/go-sqlite3" "github.com/mattn/go-sqlite3"
) )
@@ -33,6 +34,9 @@ const vacuumTestRoutes = 2000
// once so each is a distinct SQLite connection that parsed the DSN. // once so each is a distinct SQLite connection that parsed the DSN.
const heldConnections = 5 const heldConnections = 5
// bytesPerKiB turns a cache_size in KiB into bytes.
const bytesPerKiB = 1024
// testPeerIP is the peer every route in the IP lookup test is learned from. // testPeerIP is the peer every route in the IP lookup test is learned from.
const testPeerIP = "192.0.2.254" const testPeerIP = "192.0.2.254"
@@ -293,7 +297,7 @@ func TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer(t *testing.T) {
t.Fatalf("UpsertLiveRoute(%s): %v", tt.route.Prefix, err) t.Fatalf("UpsertLiveRoute(%s): %v", tt.route.Prefix, err)
} }
_, err = db.db.ExecContext(t.Context(), tt.insertSQL, uuid.NewString(), _, err = db.db.ExecContext(t.Context(), tt.insertSQL,
tt.route.Prefix, tt.route.MaskLength, otherOriginASN, tt.route.Prefix, tt.route.MaskLength, otherOriginASN,
tt.route.PeerIP, "[]", tt.route.NextHop, ts) tt.route.PeerIP, "[]", tt.route.NextHop, ts)
@@ -317,7 +321,6 @@ func writeOriginRoutes(t *testing.T, db *Database, origins map[string]int) {
for prefix, asn := range origins { for prefix, asn := range origins {
routes = append(routes, &LiveRoute{ routes = append(routes, &LiveRoute{
ID: uuid.New(),
Prefix: prefix, Prefix: prefix,
MaskLength: netip.MustParsePrefix(prefix).Bits(), MaskLength: netip.MustParsePrefix(prefix).Bits(),
IPVersion: detectIPVersion(prefix), IPVersion: detectIPVersion(prefix),
@@ -337,7 +340,7 @@ func writeOriginRoutes(t *testing.T, db *Database, origins map[string]int) {
// TestConnectionPoolPragmas holds several pooled connections open at once and // TestConnectionPoolPragmas holds several pooled connections open at once and
// checks each one carries the per-connection settings from the DSN, plus the // checks each one carries the per-connection settings from the DSN, plus the
// process-wide hard heap limit, and sees the new file with auto_vacuum // process-wide heap limits, and sees the new file with auto_vacuum
// incremental. // incremental.
func TestConnectionPoolPragmas(t *testing.T) { func TestConnectionPoolPragmas(t *testing.T) {
t.Parallel() t.Parallel()
@@ -372,9 +375,9 @@ func TestConnectionPoolPragmas(t *testing.T) {
for i, c := range conns { for i, c := range conns {
cacheSize := readPragma(ctx, t, c, i, "cache_size") cacheSize := readPragma(ctx, t, c, i, "cache_size")
if cacheSize != sqliteCacheSizeKiB { if cacheSize != sqliteReadCacheSizeKiB {
t.Errorf("conn %d: cache_size = %d, want %d", t.Errorf("conn %d: cache_size = %d, want %d",
i, cacheSize, sqliteCacheSizeKiB) i, cacheSize, sqliteReadCacheSizeKiB)
} }
busyTimeout := readPragma(ctx, t, c, i, "busy_timeout") busyTimeout := readPragma(ctx, t, c, i, "busy_timeout")
@@ -389,6 +392,12 @@ func TestConnectionPoolPragmas(t *testing.T) {
i, tempStore, tempStoreMemory) i, tempStore, tempStoreMemory)
} }
softHeapLimit := readPragma(ctx, t, c, i, "soft_heap_limit")
if softHeapLimit != sqliteSoftHeapLimitBytes {
t.Errorf("conn %d: soft_heap_limit = %d, want %d",
i, softHeapLimit, sqliteSoftHeapLimitBytes)
}
hardHeapLimit := readPragma(ctx, t, c, i, "hard_heap_limit") hardHeapLimit := readPragma(ctx, t, c, i, "hard_heap_limit")
if hardHeapLimit != sqliteHardHeapLimitBytes { if hardHeapLimit != sqliteHardHeapLimitBytes {
t.Errorf("conn %d: hard_heap_limit = %d, want %d", t.Errorf("conn %d: hard_heap_limit = %d, want %d",
@@ -403,6 +412,51 @@ func TestConnectionPoolPragmas(t *testing.T) {
} }
} }
// TestWritingConnectionCache checks that the connection kept for writing has
// the large cache, and that the soft heap limit is above the caches of all the
// connections together: SQLite shrinks the caches to stay under it.
func TestWritingConnectionCache(t *testing.T) {
t.Parallel()
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ctx := t.Context()
var writeCacheSize, readCacheSize, softHeapLimit int64
err = db.writer.QueryRowContext(ctx, "PRAGMA cache_size").
Scan(&writeCacheSize)
if err != nil {
t.Fatalf("failed to read the writing connection's cache_size: %v", err)
}
if writeCacheSize != sqliteWriteCacheSizeKiB {
t.Errorf("writing connection: cache_size = %d, want %d",
writeCacheSize, sqliteWriteCacheSizeKiB)
}
err = db.db.QueryRowContext(ctx, "PRAGMA cache_size").Scan(&readCacheSize)
if err != nil {
t.Fatalf("failed to read a reading connection's cache_size: %v", err)
}
err = db.db.QueryRowContext(ctx, "PRAGMA soft_heap_limit").
Scan(&softHeapLimit)
if err != nil {
t.Fatalf("failed to read soft_heap_limit: %v", err)
}
cachesBytes := -bytesPerKiB * (writeCacheSize + (maxConns-1)*readCacheSize)
if softHeapLimit <= cachesBytes {
t.Errorf("soft_heap_limit = %d, not above the %d bytes of the caches",
softHeapLimit, cachesBytes)
}
}
// readPragma reads the integer value of PRAGMA name on connection i, c. // readPragma reads the integer value of PRAGMA name on connection i, c.
func readPragma( func readPragma(
ctx context.Context, t *testing.T, c *sql.Conn, i int, name string, ctx context.Context, t *testing.T, c *sql.Conn, i int, name string,
@@ -1090,3 +1144,380 @@ func TestGetRandomPrefixesByLengthSkipsWithdrawnRoutes(t *testing.T) {
routes, kept.Prefix) routes, kept.Prefix)
} }
} }
// TestLiveRouteUpdateSetsOriginWhenItChanges announces a peer's route for a
// prefix, then again with the same origin AS and a new path, then with a new
// origin AS. Each time the route must carry the announced origin AS, path and
// time, and the AS details must list the prefix under that origin AS only. The
// update sets the origin AS with a statement of its own, which changes nothing
// while the origin AS stays the same.
func TestLiveRouteUpdateSetsOriginWhenItChanges(t *testing.T) {
t.Parallel()
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
const firstOrigin, secondOrigin = 64500, 64501
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
for _, asn := range []int{firstOrigin, secondOrigin} {
_, err := db.GetOrCreateASN(asn, ts)
if err != nil {
t.Fatalf("GetOrCreateASN(%d): %v", asn, err)
}
}
route := mkV4Route(t, "198.51.100.0/24", firstOrigin, ts)
for _, announced := range []struct {
origin int
path []int
}{
{firstOrigin, []int{firstOrigin}},
{firstOrigin, []int{secondOrigin, firstOrigin}},
{secondOrigin, []int{firstOrigin, secondOrigin}},
} {
ts = ts.Add(time.Minute)
route.OriginASN, route.ASPath, route.LastUpdated =
announced.origin, announced.path, ts
err := db.UpsertLiveRouteBatch([]*LiveRoute{route})
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
routes, _, err := db.GetPrefixDetails(route.Prefix)
if err != nil {
t.Fatalf("GetPrefixDetails: %v", err)
}
if len(routes) != 1 || routes[0].OriginASN != announced.origin ||
!slices.Equal(routes[0].ASPath, announced.path) ||
!routes[0].LastUpdated.Equal(ts) {
t.Errorf("after announcing origin AS%d, path %v at %s: routes %+v",
announced.origin, announced.path, ts, routes)
}
for _, asn := range []int{firstOrigin, secondOrigin} {
_, prefixes, err := db.GetASDetails(asn)
if err != nil {
t.Fatalf("GetASDetails(%d): %v", asn, err)
}
listed := len(prefixes) == 1 && prefixes[0].Prefix == route.Prefix
if listed != (asn == announced.origin) {
t.Errorf("after announcing origin AS%d, AS%d lists %+v",
announced.origin, asn, prefixes)
}
}
}
}
// TestUpdatePrefixesBatchKeepsFirstSeen writes a prefix, then a batch with it
// and a new prefix. Only the new prefix is inserted and counted, and the known
// one keeps its first_seen time.
func TestUpdatePrefixesBatchKeepsFirstSeen(t *testing.T) {
t.Parallel()
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
const known = "198.51.100.0/24"
first := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
later := first.Add(time.Hour)
err = db.UpdatePrefixesBatch(map[string]time.Time{known: first})
if err != nil {
t.Fatalf("UpdatePrefixesBatch: %v", err)
}
err = db.UpdatePrefixesBatch(map[string]time.Time{
known: later, testV6Prefix: later,
})
if err != nil {
t.Fatalf("UpdatePrefixesBatch: %v", err)
}
stats, err := db.GetStats()
if err != nil {
t.Fatalf("GetStats: %v", err)
}
if stats.IPv4Prefixes != 1 || stats.IPv6Prefixes != 1 {
t.Errorf("prefix counts = (v4 %d, v6 %d), want (1, 1)",
stats.IPv4Prefixes, stats.IPv6Prefixes)
}
for _, tt := range []struct {
query, prefix string
firstSeen time.Time
}{
{
query: "SELECT first_seen FROM prefixes_v4 WHERE prefix = ?",
prefix: known, firstSeen: first,
},
{
query: "SELECT first_seen FROM prefixes_v6 WHERE prefix = ?",
prefix: testV6Prefix, firstSeen: later,
},
} {
var firstSeen time.Time
err := db.db.QueryRowContext(t.Context(), tt.query, tt.prefix).
Scan(&firstSeen)
if err != nil {
t.Fatalf("reading %s: %v", tt.prefix, err)
}
if !firstSeen.Equal(tt.firstSeen) {
t.Errorf("%s: first seen %s, want %s", tt.prefix, firstSeen,
tt.firstSeen)
}
}
}
// TestUpsertLiveRouteBatchWritesInKeyOrder writes new routes of two peers out
// of key order, the last one a second announcement of the first, and checks
// that the rows were inserted in key order, peer then prefix (each new row's id
// is one above the last), and that the route announced twice carries the later
// announcement.
func TestUpsertLiveRouteBatchWritesInKeyOrder(t *testing.T) {
t.Parallel()
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
const (
firstOrigin, laterOrigin = 64500, 64501
// Three prefixes, in text order.
firstPrefix = "192.0.2.0/24"
secondPrefix = "198.51.100.0/24"
thirdPrefix = "203.0.113.0/24"
)
type row struct {
peer, prefix string
origin int
}
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
announcements := []row{
{secondV4PeerIP, secondPrefix, firstOrigin},
{testV4PeerIP, thirdPrefix, firstOrigin},
{secondV4PeerIP, firstPrefix, firstOrigin},
{testV4PeerIP, secondPrefix, firstOrigin},
{secondV4PeerIP, secondPrefix, laterOrigin},
}
routes := make([]*LiveRoute, 0, len(announcements))
for _, announced := range announcements {
route := mkV4Route(t, announced.prefix, announced.origin, ts)
route.PeerIP = announced.peer
routes = append(routes, route)
}
err = db.UpsertLiveRouteBatch(routes)
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
rows, err := db.db.QueryContext(t.Context(),
"SELECT peer_ip, prefix, origin_asn FROM live_routes_v4 ORDER BY id")
if err != nil {
t.Fatalf("reading the routes: %v", err)
}
defer func() { _ = rows.Close() }()
var stored []row
for rows.Next() {
var r row
err := rows.Scan(&r.peer, &r.prefix, &r.origin)
if err != nil {
t.Fatalf("reading a route: %v", err)
}
stored = append(stored, r)
}
err = rows.Err()
if err != nil {
t.Fatalf("reading the routes: %v", err)
}
want := []row{
{testV4PeerIP, secondPrefix, firstOrigin},
{testV4PeerIP, thirdPrefix, firstOrigin},
{secondV4PeerIP, firstPrefix, firstOrigin},
{secondV4PeerIP, secondPrefix, laterOrigin},
}
if !slices.Equal(stored, want) {
t.Errorf("routes in insertion order = %+v, want %+v", stored, want)
}
}
// Parameters for BenchmarkPrefixBatchOnFiveMillionRoutes.
const (
// benchPeers and benchPrefixes size the table the batches are written to:
// every peer has a route for every prefix, 5 million routes in all.
benchPeers = 10
benchPrefixes = 500000
// benchBatchSize is how many updates each batch holds, as many as a full
// prefix handler batch (prefixBatchSize in internal/routewatch).
benchBatchSize = 25000
// benchMaxBatchSize is the largest batch the prefix handler writes, the
// one it holds while a slow write runs (prefixBatchMaxSize in
// internal/routewatch).
benchMaxBatchSize = 250000
// benchWithdrawalEvery makes every fifth update of a timed batch a
// withdrawal; the others are announcements.
benchWithdrawalEvery = 5
// benchOrigins is how many origin ASes the prefixes are spread over.
benchOrigins = 50000
// benchFirstASN is the lowest AS number in the benchmark's AS paths.
benchFirstASN = 64500
// benchMaskLength is the mask length of every prefix in the benchmark.
benchMaskLength = 24
)
// BenchmarkPrefixBatchOnFiveMillionRoutes times the writes the prefix handler
// makes for one batch of announcements and withdrawals of routes picked at
// random, to a database that already holds 5 million live routes: 500,000
// IPv4 prefixes, each from 10 peers. It times a full batch of 25,000 updates,
// then the largest batch, 250,000. It builds that database first, one batch at
// a time, which takes some minutes and a few GB of disk under TMPDIR. It is
// run by hand, not by make test, for example in the test image:
//
// go test -run '^$' -bench FiveMillion -benchtime 10x ./internal/database/
func BenchmarkPrefixBatchOnFiveMillionRoutes(b *testing.B) {
db, err := New(&config.Config{StateDir: b.TempDir()}, logger.New())
if err != nil {
b.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
//nolint:gosec // The benchmark needs the same batches every run.
rng := rand.New(rand.NewPCG(1, 2))
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
// Each peer sends its whole table, as when its session comes up.
for peer := range benchPeers {
for first := 0; first < benchPrefixes; first += benchBatchSize {
routes := make([]*LiveRoute, 0, benchBatchSize)
for prefix := first; prefix < first+benchBatchSize; prefix++ {
routes = append(routes, benchRoute(rng, prefix, peer, ts))
}
writeBenchBatch(b, db, routes, nil, ts)
}
}
for _, batchSize := range []int{benchBatchSize, benchMaxBatchSize} {
b.Run(strconv.Itoa(batchSize), func(b *testing.B) {
for b.Loop() {
ts = ts.Add(time.Second)
var (
routes []*LiveRoute
withdrawals []LiveRouteDeletion
)
for i := range batchSize {
route := benchRoute(rng, rng.IntN(benchPrefixes),
rng.IntN(benchPeers), ts)
if i%benchWithdrawalEvery == 0 {
withdrawals = append(withdrawals, LiveRouteDeletion{
Prefix: route.Prefix, PeerIP: route.PeerIP,
IPVersion: ipVersionV4,
})
} else {
routes = append(routes, route)
}
}
writeBenchBatch(b, db, routes, withdrawals, ts)
}
})
}
}
// benchRoute returns the route peer has for the prefix numbered prefix, a /24
// from 1.0.0.0/24 on, with a new AS path of three to six ASes from the peer's
// AS to the prefix's origin AS.
func benchRoute(rng *rand.Rand, prefix, peer int, ts time.Time) *LiveRoute {
const minPathLength, extraPathLength = 3, 4
origin := benchFirstASN + prefix%benchOrigins
pathLength := minPathLength + rng.IntN(extraPathLength)
path := make([]int, 0, pathLength)
path = append(path, benchFirstASN+peer)
for len(path) < pathLength-1 {
path = append(path, benchFirstASN+rng.IntN(benchOrigins))
}
path = append(path, origin)
peerIP := fmt.Sprintf("198.51.100.%d", 1+peer)
return &LiveRoute{
Prefix: fmt.Sprintf("%d.%d.%d.0/%d",
1+prefix>>16, (prefix>>8)%256, prefix%256, benchMaskLength),
MaskLength: benchMaskLength,
IPVersion: ipVersionV4,
OriginASN: origin,
PeerIP: peerIP,
ASPath: path,
NextHop: peerIP,
LastUpdated: ts,
}
}
// writeBenchBatch writes what the prefix handler writes for a batch: the
// announced routes, the withdrawals, and every prefix the batch names, of
// which only the ones not stored yet are inserted.
func writeBenchBatch(
b *testing.B, db *Database, routes []*LiveRoute,
withdrawals []LiveRouteDeletion, ts time.Time,
) {
b.Helper()
prefixes := make(map[string]time.Time, len(routes)+len(withdrawals))
for _, route := range routes {
prefixes[route.Prefix] = ts
}
for _, withdrawal := range withdrawals {
prefixes[withdrawal.Prefix] = ts
}
err := db.UpsertLiveRouteBatch(routes)
if err != nil {
b.Fatalf("UpsertLiveRouteBatch: %v", err)
}
err = db.DeleteLiveRouteBatch(withdrawals)
if err != nil {
b.Fatalf("DeleteLiveRouteBatch: %v", err)
}
err = db.UpdatePrefixesBatch(prefixes)
if err != nil {
b.Fatalf("UpdatePrefixesBatch: %v", err)
}
}
+1 -4
View File
@@ -35,13 +35,11 @@ type ASN struct {
} }
// Prefix represents an IP prefix (CIDR block) with its IP version (4 or 6) // Prefix represents an IP prefix (CIDR block) with its IP version (4 or 6)
// and first/last seen timestamps. // and the time it was first seen.
type Prefix struct { type Prefix struct {
ID uuid.UUID
Prefix string Prefix string
IPVersion int // 4 or 6 IPVersion int // 4 or 6
FirstSeen time.Time FirstSeen time.Time
LastSeen time.Time
} }
// Announcement represents a BGP announcement or withdrawal event, // Announcement represents a BGP announcement or withdrawal event,
@@ -71,7 +69,6 @@ type ASNPeering struct {
// //
//nolint:tagliatelle // field names are the existing JSON API //nolint:tagliatelle // field names are the existing JSON API
type LiveRoute struct { type LiveRoute struct {
ID uuid.UUID `json:"id"`
Prefix string `json:"prefix"` Prefix string `json:"prefix"`
MaskLength int `json:"mask_length"` MaskLength int `json:"mask_length"`
IPVersion int `json:"ip_version"` IPVersion int `json:"ip_version"`
+40 -38
View File
@@ -27,21 +27,18 @@ CREATE TABLE IF NOT EXISTS asns (
whois_updated_at DATETIME -- when we last fetched WHOIS data whois_updated_at DATETIME -- when we last fetched WHOIS data
); );
-- IPv4 prefixes table -- IPv4 prefixes table. Every read and write finds a row by its prefix, so the
-- rows are stored by prefix (WITHOUT ROWID) and need no index of their own.
CREATE TABLE IF NOT EXISTS prefixes_v4 ( CREATE TABLE IF NOT EXISTS prefixes_v4 (
id TEXT PRIMARY KEY, prefix TEXT PRIMARY KEY,
prefix TEXT UNIQUE NOT NULL, first_seen DATETIME NOT NULL
first_seen DATETIME NOT NULL, ) WITHOUT ROWID;
last_seen DATETIME NOT NULL
);
-- IPv6 prefixes table -- IPv6 prefixes table, stored as prefixes_v4 is.
CREATE TABLE IF NOT EXISTS prefixes_v6 ( CREATE TABLE IF NOT EXISTS prefixes_v6 (
id TEXT PRIMARY KEY, prefix TEXT PRIMARY KEY,
prefix TEXT UNIQUE NOT NULL, first_seen DATETIME NOT NULL
first_seen DATETIME NOT NULL, ) WITHOUT ROWID;
last_seen DATETIME NOT NULL
);
CREATE TABLE IF NOT EXISTS announcements ( CREATE TABLE IF NOT EXISTS announcements (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
@@ -75,11 +72,6 @@ CREATE TABLE IF NOT EXISTS bgp_peers (
last_message_type TEXT last_message_type TEXT
); );
-- Indexes for prefixes_v4 table
CREATE INDEX IF NOT EXISTS idx_prefixes_v4_prefix ON prefixes_v4(prefix);
-- Indexes for prefixes_v6 table
CREATE INDEX IF NOT EXISTS idx_prefixes_v6_prefix ON prefixes_v6(prefix);
CREATE INDEX IF NOT EXISTS idx_announcements_timestamp ON announcements(timestamp); CREATE INDEX IF NOT EXISTS idx_announcements_timestamp ON announcements(timestamp);
CREATE INDEX IF NOT EXISTS idx_announcements_prefix_id ON announcements(prefix_id); CREATE INDEX IF NOT EXISTS idx_announcements_prefix_id ON announcements(prefix_id);
CREATE INDEX IF NOT EXISTS idx_announcements_peer_asn ON announcements(peer_asn); CREATE INDEX IF NOT EXISTS idx_announcements_peer_asn ON announcements(peer_asn);
@@ -97,9 +89,14 @@ CREATE INDEX IF NOT EXISTS idx_bgp_peers_asn ON bgp_peers(peer_asn);
CREATE INDEX IF NOT EXISTS idx_bgp_peers_last_seen ON bgp_peers(last_seen); CREATE INDEX IF NOT EXISTS idx_bgp_peers_last_seen ON bgp_peers(last_seen);
CREATE INDEX IF NOT EXISTS idx_bgp_peers_ip ON bgp_peers(peer_ip); CREATE INDEX IF NOT EXISTS idx_bgp_peers_ip ON bgp_peers(peer_ip);
-- IPv4 routing table maintained by PrefixHandler -- IPv4 routing table maintained by PrefixHandler. A route write changes the
-- table and its entry in every index on a column the write sets, so the table
-- has only the indexes that the queries named with them use.
CREATE TABLE IF NOT EXISTS live_routes_v4 ( CREATE TABLE IF NOT EXISTS live_routes_v4 (
id TEXT PRIMARY KEY, -- The row id, which the prefix-length pages' sample draws through the
-- mask_length index. Declared as a column, it stays the same for as long
-- as the route is live, also through a VACUUM.
id INTEGER PRIMARY KEY,
prefix TEXT NOT NULL, prefix TEXT NOT NULL,
mask_length INTEGER NOT NULL, -- CIDR mask length (0-32) mask_length INTEGER NOT NULL, -- CIDR mask length (0-32)
origin_asn INTEGER NOT NULL, origin_asn INTEGER NOT NULL,
@@ -107,15 +104,15 @@ CREATE TABLE IF NOT EXISTS live_routes_v4 (
as_path TEXT NOT NULL, -- JSON array as_path TEXT NOT NULL, -- JSON array
next_hop TEXT NOT NULL, next_hop TEXT NOT NULL,
last_updated DATETIME NOT NULL, last_updated DATETIME NOT NULL,
-- A peer has at most one route for a prefix. peer_ip comes first so that -- A peer has at most one route for a prefix. This key finds the route a
-- this key also finds all of a peer's routes, when its BGP session goes -- write updates or deletes and, peer_ip coming first, all of a peer's
-- down; lookups by prefix use the prefix index below. -- routes when its BGP session goes down.
UNIQUE(peer_ip, prefix) UNIQUE(peer_ip, prefix)
); );
-- IPv6 routing table maintained by PrefixHandler -- IPv6 routing table maintained by PrefixHandler, as live_routes_v4.
CREATE TABLE IF NOT EXISTS live_routes_v6 ( CREATE TABLE IF NOT EXISTS live_routes_v6 (
id TEXT PRIMARY KEY, id INTEGER PRIMARY KEY,
prefix TEXT NOT NULL, prefix TEXT NOT NULL,
mask_length INTEGER NOT NULL, -- CIDR mask length (0-128) mask_length INTEGER NOT NULL, -- CIDR mask length (0-128)
origin_asn INTEGER NOT NULL, origin_asn INTEGER NOT NULL,
@@ -123,24 +120,29 @@ CREATE TABLE IF NOT EXISTS live_routes_v6 (
as_path TEXT NOT NULL, -- JSON array as_path TEXT NOT NULL, -- JSON array
next_hop TEXT NOT NULL, next_hop TEXT NOT NULL,
last_updated DATETIME NOT NULL, last_updated DATETIME NOT NULL,
-- As in live_routes_v4: peer_ip first, to find all of a peer's routes.
UNIQUE(peer_ip, prefix) UNIQUE(peer_ip, prefix)
); );
-- Indexes for live_routes_v4 table -- Indexes for both live route tables, and the queries that use each.
-- Finds a prefix's newest route in one search, whatever the number of peers. --
-- prefix, last_updated: a prefix's routes, for the prefix page and its API
-- answer, the IP lookup and its peer count, and the check whether a prefix
-- still has a live route; and a prefix's newest route for the prefix-length
-- pages, in one search whatever the number of peers.
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_prefix_last_updated ON live_routes_v4(prefix, last_updated); CREATE INDEX IF NOT EXISTS idx_live_routes_v4_prefix_last_updated ON live_routes_v4(prefix, last_updated);
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_length ON live_routes_v4(mask_length);
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_asn ON live_routes_v4(origin_asn);
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated);
-- Index to optimize prefix distribution queries
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_prefix ON live_routes_v4(mask_length, prefix);
-- Indexes for live_routes_v6 table
-- Finds a prefix's newest route in one search, whatever the number of peers.
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_prefix_last_updated ON live_routes_v6(prefix, last_updated); CREATE INDEX IF NOT EXISTS idx_live_routes_v6_prefix_last_updated ON live_routes_v6(prefix, last_updated);
-- mask_length: the prefix-length pages' random sample, the row id range of a
-- mask length and the first route of it at or after a row id.
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_length ON live_routes_v4(mask_length);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_length ON live_routes_v6(mask_length); CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_length ON live_routes_v6(mask_length);
-- 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_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); -- last_updated: the oldest and newest route times in /api/v1/stats and
-- Index to optimize prefix distribution queries -- /status.json.
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_prefix ON live_routes_v6(mask_length, prefix); CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_last_updated ON live_routes_v6(last_updated);
+2 -2
View File
@@ -18,12 +18,12 @@ func logSlowQuery(logger *logger.Logger, query string, start time.Time) {
} }
} }
// exec wraps Exec with slow query logging // exec runs a statement on the writing connection, with slow query logging
func (d *Database) exec(query string, args ...any) error { func (d *Database) exec(query string, args ...any) error {
start := time.Now() start := time.Now()
defer logSlowQuery(d.logger, query, start) defer logSlowQuery(d.logger, query, start)
_, err := d.db.ExecContext(context.Background(), query, args...) _, err := d.writer.ExecContext(context.Background(), query, args...)
return err return err
} }
@@ -18,7 +18,6 @@ import (
"git.eeqj.de/sneak/routewatch/internal/ristypes" "git.eeqj.de/sneak/routewatch/internal/ristypes"
"git.eeqj.de/sneak/routewatch/internal/server" "git.eeqj.de/sneak/routewatch/internal/server"
"git.eeqj.de/sneak/routewatch/internal/streamer" "git.eeqj.de/sneak/routewatch/internal/streamer"
"github.com/google/uuid"
) )
// mockStore is a mock implementation of database.Store for testing // mockStore is a mock implementation of database.Store for testing
@@ -83,9 +82,7 @@ func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
defer m.mu.Unlock() defer m.mu.Unlock()
for prefix, timestamp := range prefixes { for prefix, timestamp := range prefixes {
if p, exists := m.Prefixes[prefix]; exists { if _, exists := m.Prefixes[prefix]; !exists {
p.LastSeen = timestamp
} else {
const ( const (
ipVersionV4 = 4 ipVersionV4 = 4
ipVersionV6 = 6 ipVersionV6 = 6
@@ -97,11 +94,9 @@ func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
} }
m.Prefixes[prefix] = &database.Prefix{ m.Prefixes[prefix] = &database.Prefix{
ID: uuid.New(),
Prefix: prefix, Prefix: prefix,
IPVersion: ipVersion, IPVersion: ipVersion,
FirstSeen: timestamp, FirstSeen: timestamp,
LastSeen: timestamp,
} }
} }
} }
@@ -117,8 +112,6 @@ func (m *mockStore) GetOrCreatePrefix(
defer m.mu.Unlock() defer m.mu.Unlock()
if p, exists := m.Prefixes[prefix]; exists { if p, exists := m.Prefixes[prefix]; exists {
p.LastSeen = timestamp
return p, nil return p, nil
} }
@@ -133,11 +126,9 @@ func (m *mockStore) GetOrCreatePrefix(
} }
p := &database.Prefix{ p := &database.Prefix{
ID: uuid.New(),
Prefix: prefix, Prefix: prefix,
IPVersion: ipVersion, IPVersion: ipVersion,
FirstSeen: timestamp, FirstSeen: timestamp,
LastSeen: timestamp,
} }
m.Prefixes[prefix] = p m.Prefixes[prefix] = p
m.PrefixCount++ m.PrefixCount++
@@ -370,11 +361,9 @@ func (m *mockStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
// Track prefix // Track prefix
if _, exists := m.Prefixes[route.Prefix]; !exists { if _, exists := m.Prefixes[route.Prefix]; !exists {
m.Prefixes[route.Prefix] = &database.Prefix{ m.Prefixes[route.Prefix] = &database.Prefix{
ID: uuid.New(),
Prefix: route.Prefix, Prefix: route.Prefix,
IPVersion: route.IPVersion, IPVersion: route.IPVersion,
FirstSeen: route.LastUpdated, FirstSeen: route.LastUpdated,
LastSeen: route.LastUpdated,
} }
m.PrefixCount++ m.PrefixCount++
-2
View File
@@ -12,7 +12,6 @@ import (
"git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/logger"
"git.eeqj.de/sneak/routewatch/internal/metrics" "git.eeqj.de/sneak/routewatch/internal/metrics"
"git.eeqj.de/sneak/routewatch/internal/ristypes" "git.eeqj.de/sneak/routewatch/internal/ristypes"
"github.com/google/uuid"
) )
const ( const (
@@ -461,7 +460,6 @@ func (h *PrefixHandler) createLiveRoute(update prefixUpdate) *database.LiveRoute
// Create live route record // Create live route record
liveRoute := &database.LiveRoute{ liveRoute := &database.LiveRoute{
ID: uuid.New(),
Prefix: update.prefix, Prefix: update.prefix,
MaskLength: maskLength, MaskLength: maskLength,
IPVersion: ipVersion, IPVersion: ipVersion,
+7 -8
View File
@@ -19,7 +19,6 @@ import (
"git.eeqj.de/sneak/routewatch/internal/metrics" "git.eeqj.de/sneak/routewatch/internal/metrics"
"git.eeqj.de/sneak/routewatch/internal/streamer" "git.eeqj.de/sneak/routewatch/internal/streamer"
"git.eeqj.de/sneak/routewatch/pkg/asinfo" "git.eeqj.de/sneak/routewatch/pkg/asinfo"
"github.com/google/uuid"
) )
// The prefix and peer addresses the prefix page tests store routes for. // The prefix and peer addresses the prefix page tests store routes for.
@@ -114,12 +113,12 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) {
err = db.UpsertLiveRouteBatch([]*database.LiveRoute{ err = db.UpsertLiveRouteBatch([]*database.LiveRoute{
{ {
ID: uuid.New(), Prefix: "198.51.100.0/24", MaskLength: 24, IPVersion: 4, Prefix: "198.51.100.0/24", MaskLength: 24, IPVersion: 4,
OriginASN: 64500, PeerIP: "192.0.2.1", ASPath: []int{64500}, OriginASN: 64500, PeerIP: "192.0.2.1", ASPath: []int{64500},
NextHop: "192.0.2.1", LastUpdated: ts, NextHop: "192.0.2.1", LastUpdated: ts,
}, },
{ {
ID: uuid.New(), Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6, Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6,
OriginASN: 64501, PeerIP: testPeerV6, ASPath: []int{64501}, OriginASN: 64501, PeerIP: testPeerV6, ASPath: []int{64501},
NextHop: testPeerV6, LastUpdated: ts, NextHop: testPeerV6, LastUpdated: ts,
}, },
@@ -204,12 +203,12 @@ func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
err = db.UpsertLiveRouteBatch([]*database.LiveRoute{ err = db.UpsertLiveRouteBatch([]*database.LiveRoute{
{ {
ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4,
OriginASN: 64500, PeerIP: testPeerV4, ASPath: []int{64500}, OriginASN: 64500, PeerIP: testPeerV4, ASPath: []int{64500},
NextHop: testPeerV4, LastUpdated: ts, NextHop: testPeerV4, LastUpdated: ts,
}, },
{ {
ID: uuid.New(), Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6, Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6,
OriginASN: 64501, PeerIP: testPeerV6, ASPath: []int{64501}, OriginASN: 64501, PeerIP: testPeerV6, ASPath: []int{64501},
NextHop: testPeerV6, LastUpdated: ts, NextHop: testPeerV6, LastUpdated: ts,
}, },
@@ -287,12 +286,12 @@ func TestPrefixDetailShowsOriginNames(t *testing.T) {
err = db.UpsertLiveRouteBatch([]*database.LiveRoute{ err = db.UpsertLiveRouteBatch([]*database.LiveRoute{
{ {
ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4,
OriginASN: origins[0], PeerIP: testPeerV4, OriginASN: origins[0], PeerIP: testPeerV4,
ASPath: []int{origins[0]}, NextHop: testPeerV4, LastUpdated: ts, ASPath: []int{origins[0]}, NextHop: testPeerV4, LastUpdated: ts,
}, },
{ {
ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4,
OriginASN: origins[1], PeerIP: "198.51.100.2", OriginASN: origins[1], PeerIP: "198.51.100.2",
ASPath: []int{origins[1]}, NextHop: "198.51.100.2", LastUpdated: ts, ASPath: []int{origins[1]}, NextHop: "198.51.100.2", LastUpdated: ts,
}, },
@@ -347,7 +346,7 @@ func TestPagesSayOriginZeroIsNoKnownOrigin(t *testing.T) {
t.Cleanup(func() { _ = db.Close() }) t.Cleanup(func() { _ = db.Close() })
err = db.UpsertLiveRouteBatch([]*database.LiveRoute{{ err = db.UpsertLiveRouteBatch([]*database.LiveRoute{{
ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4, Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4,
OriginASN: 0, PeerIP: testPeerV4, ASPath: []int{64500, 64501, 64502}, OriginASN: 0, PeerIP: testPeerV4, ASPath: []int{64500, 64501, 64502},
NextHop: testPeerV4, LastUpdated: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC), NextHop: testPeerV4, LastUpdated: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
}}) }})