From b0cd884019eab3c1544abe40c6df185d7ea01079 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 09:05:33 +0200 Subject: [PATCH 01/19] Skip live network feed test under -short for deterministic make check (closes #2) TestRouteWatchLiveFeed streams the live RIPE RIS feed for a few seconds, so the default test run depended on the network and tripped the race detector. It now skips under testing.Short(), and script/test passes -short in both go test lines, so make check runs offline and gives the same result every time. The README says how to run the live test on demand. No non-test code changed. The race itself is not fixed: the test reads the peeringHandler field of RouteWatch while Run is still setting it. Nothing outside the test reads that field concurrently. Model: opus-4-8 (implementation, review); fable-5-1 (summary) --- README.md | 9 ++++++++- internal/routewatch/app_integration_test.go | 3 +++ script/test | 4 ++-- 3 files changed, 13 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index ccf2218..5bfe36c 100644 --- a/README.md +++ b/README.md @@ -204,7 +204,9 @@ them. We provide: - `script/projectname` — print the project name (used for the Docker image tag) - `script/test` — run the test suite - (`go test -timeout 30s -race -cover ./...`, verbose rerun on failure) + (`go test -short -timeout 30s -race -cover ./...`, verbose rerun on + failure). The `-short` flag skips the live-network integration test so the + default run is deterministic and offline. - `script/lint` — run `go vet ./...` and `golangci-lint run` - `script/fmt` — format all code (writes) - `script/fmt-check` — check formatting (read-only) @@ -218,6 +220,11 @@ them. We provide: - `script/install-precommit` — install the git pre-commit hook that runs `script/precommit` +The live-network integration test `TestRouteWatchLiveFeed` streams the RIPE +RIS feed for a few seconds and is skipped in short mode. To run it on demand, +invoke `go test` directly without `-short`: +`go test -run TestRouteWatchLiveFeed ./internal/routewatch/`. + ## License See LICENSE file. diff --git a/internal/routewatch/app_integration_test.go b/internal/routewatch/app_integration_test.go index 619210d..807a0bd 100644 --- a/internal/routewatch/app_integration_test.go +++ b/internal/routewatch/app_integration_test.go @@ -426,6 +426,9 @@ func (m *mockStore) Ping(ctx context.Context) error { } func TestRouteWatchLiveFeed(t *testing.T) { + if testing.Short() { + t.Skip("skipping live RIPE RIS network feed test in short mode; run without -short to include it") + } // Create mock database mockDB := newMockStore() diff --git a/script/test b/script/test index 87c1a57..bffb59e 100755 --- a/script/test +++ b/script/test @@ -7,9 +7,9 @@ ROOT="$(cd "$(dirname "$0")/.." && pwd -P)" main() { cd "$ROOT" - go test -timeout 30s -race -cover ./... || { + go test -short -timeout 30s -race -cover ./... || { echo "--- Rerunning with -v for details ---" - go test -timeout 30s -race -v ./... + go test -short -timeout 30s -race -v ./... exit 1 } } -- 2.54.0 From 54014c88c83f2427504b4dafdead27e2823b1ece Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 09:35:16 +0200 Subject: [PATCH 02/19] Run make check inside the Docker build with a pinned lint stage (closes #5) REPO_POLICIES.md requires the container build to run the checks; the Dockerfile only compiled the binary, so script/cibuild checked nothing. A new lint stage on the golangci-lint v2.7.2 image runs make fmt-check and make lint, and the build stage waits for it and then runs make test, so make docker now fails when formatting, lint or a test fails. All three base images are pinned by digest with their versions unchanged. The final image and how it starts are unchanged. make docker is now the gate to use on hosts where the installed linter is older than the Go toolchain and make lint cannot run directly. Model: opus-4-8 (implementation, review); fable-5-1 (summary) --- Dockerfile | 29 +++++++++++++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/Dockerfile b/Dockerfile index fa1460b..04e60ea 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,22 @@ +# Lint stage — fast feedback on formatting and lint issues. +# The golangci-lint image bundles Go, gcc and make, so it can run go vet on +# the CGO sqlite package and golangci-lint without extra installs. +# golangci/golangci-lint:v2.7.2 (Go 1.25.5), 2026-09-21 +FROM golangci/golangci-lint@sha256:5d6d5c70a61f1356adfd9dd6316ce286799fefc9d743421356ff1b00842368ba AS lint + +WORKDIR /src + +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . + +RUN make fmt-check +RUN make lint + # Build stage -FROM golang:1.24-bookworm AS builder +# golang:1.24-bookworm, 2026-09-21 +FROM golang@sha256:1a6d4452c65dea36aac2e2d606b01b4a029ec90cc1ae53890540ce6173ea77ac AS builder # Install build dependencies (zstd for archive, gcc for CGO/sqlite3) RUN apt-get update && apt-get install -y --no-install-recommends \ @@ -10,12 +27,19 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ WORKDIR /src +# Force BuildKit to run the lint stage before compiling or testing. +COPY --from=lint /src/go.sum /dev/null + # Copy everything COPY . . # Vendor dependencies (must be after copying source) RUN go mod download && go mod vendor +# Run the test suite in the build stage: -race needs cgo and the C compiler +# installed above. The suite is offline (the live-feed test is opt-in). +RUN make test + # Build the binary with CGO enabled (required for sqlite3) RUN CGO_ENABLED=1 GOOS=linux go build -o /routewatch ./cmd/routewatch @@ -26,7 +50,8 @@ RUN tar --zstd -cf /routewatch-source.tar.zst \ . # Runtime stage -FROM debian:bookworm-slim +# debian:bookworm-slim, 2026-09-21 +FROM debian@sha256:3783cc01769c7b2b1b83a5c5ad96c815348e28ed7da68e2e3687004faa906251 # Install runtime dependencies # - ca-certificates: for HTTPS connections -- 2.54.0 From 594e7a504b03dc4573e5b6444008a0390146ac7d Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 15:46:34 +0200 Subject: [PATCH 03/19] Bound SQLite memory across the whole connection pool (closes #8) The 3 GiB page cache and temp_store=MEMORY were set once in Initialize, so only one pooled connection carried them and the other nine got no busy_timeout, which drove many "database is locked" errors. Move the per-connection settings into the DSN so every pooled connection gets a 64 MiB cache (640 MiB worst case over ten connections), synchronous OFF, a 5 s busy_timeout and WAL. Drop those pragmas from Initialize; DISTINCT temp B-trees now spill to disk. Add process-wide soft (1 GiB) and hard (1.5 GiB) heap limits; at the hard limit a statement returns SQLITE_NOMEM and the existing batch paths log and drop, so nothing panics or exits. Model: opus-4-8 --- internal/database/database.go | 36 +++++++++++--- internal/database/database_test.go | 79 ++++++++++++++++++++++++++++++ 2 files changed, 107 insertions(+), 8 deletions(-) diff --git a/internal/database/database.go b/internal/database/database.go index aac7744..629c1ed 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -38,6 +38,22 @@ const ( maxIPv4 = 0xFFFFFFFF ) +// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every +// pooled connection gets them; the heap limits are process-wide and set once. +const ( + // sqliteCacheSizeKiB is the per-connection page cache; negative means KiB. + // -65536 = 64 MiB, so at most 640 MiB across the 10-connection pool. + sqliteCacheSizeKiB = -65536 + // sqliteBusyTimeoutMs is how long a connection waits on a locked database. + sqliteBusyTimeoutMs = 5000 + // sqliteSoftHeapLimitBytes (1 GiB) makes SQLite recycle its cache rather + // than allocate once its C heap passes this size. + sqliteSoftHeapLimitBytes = 1073741824 + // sqliteHardHeapLimitBytes (1.5 GiB) fails a statement with SQLITE_NOMEM + // instead of growing the C heap without bound. + sqliteHardHeapLimitBytes = 1610612736 +) + // Common errors var ( // ErrInvalidIP is returned when an IP address is malformed @@ -71,11 +87,13 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { return nil, fmt.Errorf("failed to create database directory: %w", err) } - // Add connection parameters for go-sqlite3 - // Configure SQLite connection parameters + // Per-connection SQLite settings go in the DSN so every pooled connection + // gets them, not just the one that runs the Initialize pragmas. dsn := fmt.Sprintf( - "file:%s", + "file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d&_journal_mode=WAL", dbPath, + sqliteCacheSizeKiB, + sqliteBusyTimeoutMs, ) db, err := sql.Open("sqlite3", dsn) if err != nil { @@ -104,15 +122,17 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { // Initialize creates the database schema if it doesn't exist. func (d *Database) Initialize() error { - // Set SQLite pragmas for performance + // Set SQLite pragmas for performance. Per-connection settings (cache_size, + // synchronous, busy_timeout, journal_mode) live in the DSN; temp_store is + // left at its default so DISTINCT temp B-trees spill to disk instead of C + // heap. The heap limits below are process-wide, so setting them once here is + // enough for the whole pool. pragmas := []string{ "PRAGMA journal_mode=WAL", // Write-Ahead Logging - "PRAGMA synchronous=OFF", // Don't wait for disk writes - "PRAGMA cache_size=-3145728", // 3GB cache (upper limit for 2.4GB DB) - "PRAGMA temp_store=MEMORY", // Use memory for temp tables - "PRAGMA busy_timeout=5000", // 5 second busy timeout "PRAGMA analysis_limit=0", // Disable automatic ANALYZE "PRAGMA auto_vacuum=INCREMENTAL", // Enable incremental vacuum + fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes), + fmt.Sprintf("PRAGMA hard_heap_limit=%d", sqliteHardHeapLimitBytes), } for _, pragma := range pragmas { diff --git a/internal/database/database_test.go b/internal/database/database_test.go index 6636871..2bbe938 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -1,10 +1,23 @@ package database import ( + "context" + "database/sql" "net" "testing" + + "git.eeqj.de/sneak/routewatch/internal/config" + "git.eeqj.de/sneak/routewatch/internal/logger" ) +// 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 + +// 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 + func TestIPToUint32(t *testing.T) { tests := []struct { name string @@ -282,6 +295,72 @@ func TestIPv4RangeIntegration(t *testing.T) { } } +// TestConnectionPoolPragmas holds several pooled connections open at once and +// checks each one carries the per-connection settings from the DSN, plus the +// process-wide hard heap limit. +func TestConnectionPoolPragmas(t *testing.T) { + 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 := 0; i < heldConnections; i++ { + 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 { + var cacheSize int + if err := c.QueryRowContext(ctx, "PRAGMA cache_size").Scan(&cacheSize); err != nil { + t.Fatalf("conn %d: failed to read cache_size: %v", i, err) + } + if cacheSize != sqliteCacheSizeKiB { + t.Errorf("conn %d: cache_size = %d, want %d", i, cacheSize, sqliteCacheSizeKiB) + } + + var busyTimeout int + if err := c.QueryRowContext(ctx, "PRAGMA busy_timeout").Scan(&busyTimeout); err != nil { + t.Fatalf("conn %d: failed to read busy_timeout: %v", i, err) + } + if busyTimeout != sqliteBusyTimeoutMs { + t.Errorf("conn %d: busy_timeout = %d, want %d", i, busyTimeout, sqliteBusyTimeoutMs) + } + + var tempStore int + if err := c.QueryRowContext(ctx, "PRAGMA temp_store").Scan(&tempStore); err != nil { + t.Fatalf("conn %d: failed to read temp_store: %v", i, err) + } + if tempStore == tempStoreMemory { + t.Errorf("conn %d: temp_store = %d, want anything but %d (MEMORY)", i, tempStore, tempStoreMemory) + } + + var hardHeapLimit int64 + if err := c.QueryRowContext(ctx, "PRAGMA hard_heap_limit").Scan(&hardHeapLimit); err != nil { + t.Fatalf("conn %d: failed to read hard_heap_limit: %v", i, err) + } + if hardHeapLimit != sqliteHardHeapLimitBytes { + t.Errorf("conn %d: hard_heap_limit = %d, want %d", i, hardHeapLimit, sqliteHardHeapLimitBytes) + } + } +} + func BenchmarkIPToUint32(b *testing.B) { ip := net.ParseIP("192.168.1.1") b.ResetTimer() -- 2.54.0 From 211fdad9c0ef1bf5ee0c4a3946de50a9802f54f0 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 16:01:08 +0200 Subject: [PATCH 04/19] Bound peering AS-path map and swap it instead of copying (closes #10) processPeerings now takes the accumulated AS-path map under the lock and replaces it with a fresh empty one, so each path is processed once and the map never grows past a single 30-second interval's traffic. HandleMessage stops adding new paths once maxTrackedPaths (500000) is reached and counts the drops, which are logged with each run. The 30-minute prune and its ticker are removed as dead code under the swap, and the map mutex drops from RWMutex to Mutex since the read path is gone. RecordPeering already upserts last_seen, so stored peerings are unchanged. Model: opus-4-8 --- internal/routewatch/peeringhandler.go | 97 +++++------- internal/routewatch/peeringhandler_test.go | 163 +++++++++++++++++++++ 2 files changed, 197 insertions(+), 63 deletions(-) create mode 100644 internal/routewatch/peeringhandler_test.go diff --git a/internal/routewatch/peeringhandler.go b/internal/routewatch/peeringhandler.go index 050cb23..820ccde 100644 --- a/internal/routewatch/peeringhandler.go +++ b/internal/routewatch/peeringhandler.go @@ -21,19 +21,15 @@ const ( // than 2 ASNs cannot contain any peering information. minPathLengthForPeering = 2 - // pathExpirationTime determines how long AS paths are kept in memory - // before being eligible for pruning. Paths older than this are removed - // to prevent unbounded memory growth. - pathExpirationTime = 30 * time.Minute - // peeringProcessInterval controls how frequently the handler processes // accumulated AS paths and extracts peering relationships to store // in the database. peeringProcessInterval = 30 * time.Second - // pathPruneInterval determines how often the handler checks for and - // removes expired AS paths from memory. - pathPruneInterval = 5 * time.Minute + // maxTrackedPaths bounds how many distinct AS paths are held in memory + // between processing runs. Once the map is full, further new paths are + // dropped and counted until the next run empties it. + maxTrackedPaths = 500000 ) // PeeringHandler processes BGP UPDATE messages to extract and track @@ -46,17 +42,17 @@ type PeeringHandler struct { logger *logger.Logger // In-memory AS path tracking - mu sync.RWMutex - asPaths map[string]time.Time // key is JSON-encoded AS path + mu sync.Mutex + asPaths map[string]time.Time // key is JSON-encoded AS path + droppedPaths int // paths dropped because the map was full stopCh chan struct{} } // NewPeeringHandler creates and initializes a new PeeringHandler with the -// provided database store and logger. It starts two background goroutines: -// one for periodic processing of accumulated AS paths into peering records, -// and one for pruning expired paths from memory. The handler begins -// processing immediately upon creation. +// provided database store and logger. It starts one background goroutine that +// periodically processes accumulated AS paths into peering records. The +// handler begins processing immediately upon creation. func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler { h := &PeeringHandler{ db: db, @@ -65,9 +61,8 @@ func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler stopCh: make(chan struct{}), } - // Start the periodic processing goroutines + // Start the periodic processing goroutine go h.processLoop() - go h.pruneLoop() return h } @@ -106,9 +101,18 @@ func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) { return } + key := string(pathJSON) h.mu.Lock() - h.asPaths[string(pathJSON)] = timestamp + if _, exists := h.asPaths[key]; exists { + // Already tracked: refresh its timestamp. + h.asPaths[key] = timestamp + } else if len(h.asPaths) >= maxTrackedPaths { + // Map is full; drop this new path and count it. + h.droppedPaths++ + } else { + h.asPaths[key] = timestamp + } h.mu.Unlock() } @@ -130,41 +134,6 @@ func (h *PeeringHandler) processLoop() { } } -// pruneLoop runs periodically to remove old AS paths -func (h *PeeringHandler) pruneLoop() { - ticker := time.NewTicker(pathPruneInterval) - defer ticker.Stop() - - for { - select { - case <-ticker.C: - h.prunePaths() - case <-h.stopCh: - return - } - } -} - -// prunePaths removes AS paths older than pathExpirationTime -func (h *PeeringHandler) prunePaths() { - cutoff := time.Now().Add(-pathExpirationTime) - var removed int - - h.mu.Lock() - for pathKey, timestamp := range h.asPaths { - if timestamp.Before(cutoff) { - delete(h.asPaths, pathKey) - removed++ - } - } - pathCount := len(h.asPaths) - h.mu.Unlock() - - if removed > 0 { - h.logger.Debug("Pruned old AS paths", "removed", removed, "remaining", pathCount) - } -} - // ProcessPeeringsNow triggers immediate processing of all accumulated AS // paths into peering records. This bypasses the normal periodic processing // schedule and is primarily intended for testing purposes. @@ -174,15 +143,16 @@ func (h *PeeringHandler) ProcessPeeringsNow() { // processPeerings extracts peerings from AS paths and writes to database func (h *PeeringHandler) processPeerings() { - // Take a snapshot of current AS paths - h.mu.RLock() - pathsCopy := make(map[string]time.Time, len(h.asPaths)) - for k, v := range h.asPaths { - pathsCopy[k] = v - } - h.mu.RUnlock() + // Take the accumulated paths and replace the map with a fresh empty one + // under the lock. Each path is processed exactly once and the memory is + // released, so the map never grows past a single interval's traffic. + h.mu.Lock() + paths := h.asPaths + h.asPaths = make(map[string]time.Time) + dropped := h.droppedPaths + h.mu.Unlock() - if len(pathsCopy) == 0 { + if len(paths) == 0 { return } @@ -192,7 +162,7 @@ func (h *PeeringHandler) processPeerings() { } peerings := make(map[peeringKey]time.Time) - for pathJSON, timestamp := range pathsCopy { + for pathJSON, timestamp := range paths { var path []int if err := json.Unmarshal([]byte(pathJSON), &path); err != nil { h.logger.Error("Failed to decode AS path", "error", err) @@ -241,15 +211,16 @@ func (h *PeeringHandler) processPeerings() { } h.logger.Info("Processed AS peerings", - "paths", len(pathsCopy), + "paths", len(paths), "unique_peerings", len(peerings), "success", successCount, + "dropped_paths", dropped, "duration", time.Since(start), ) } // Stop gracefully shuts down the handler by signaling the background -// goroutines to stop and performing a final synchronous processing of +// goroutine to stop and performing a final synchronous processing of // any remaining AS paths. This ensures no peering data is lost during // shutdown. func (h *PeeringHandler) Stop() { diff --git a/internal/routewatch/peeringhandler_test.go b/internal/routewatch/peeringhandler_test.go new file mode 100644 index 0000000..364fa46 --- /dev/null +++ b/internal/routewatch/peeringhandler_test.go @@ -0,0 +1,163 @@ +package routewatch + +import ( + "encoding/json" + "strconv" + "sync" + "testing" + "time" + + "git.eeqj.de/sneak/routewatch/internal/database" + "git.eeqj.de/sneak/routewatch/internal/logger" + "git.eeqj.de/sneak/routewatch/internal/ristypes" +) + +const ( + testASNA = 64500 + testASNB = 64501 + testASNC = 64502 +) + +// recordingStore wraps mockStore to count every RecordPeering call, so a +// test can tell how many times a peering was written across separate runs. +type recordingStore struct { + *mockStore + + mu sync.Mutex + calls int +} + +func (r *recordingStore) RecordPeering(asA, asB int, ts time.Time) error { + r.mu.Lock() + r.calls++ + r.mu.Unlock() + + return r.mockStore.RecordPeering(asA, asB, ts) +} + +func (r *recordingStore) callCount() int { + r.mu.Lock() + defer r.mu.Unlock() + + return r.calls +} + +// newTestHandler builds a PeeringHandler without starting the periodic +// processing goroutine, so tests drive processing explicitly. +func newTestHandler(db database.Store) *PeeringHandler { + return &PeeringHandler{ + db: db, + logger: logger.New(), + asPaths: make(map[string]time.Time), + stopCh: make(chan struct{}), + } +} + +func snapshot(h *PeeringHandler) (tracked, dropped int) { + h.mu.Lock() + defer h.mu.Unlock() + + return len(h.asPaths), h.droppedPaths +} + +func pathKey(t *testing.T, asns ...int) string { + t.Helper() + + b, err := json.Marshal(ristypes.ASPath(asns)) + if err != nil { + t.Fatalf("failed to marshal path: %v", err) + } + + return string(b) +} + +func handle(h *PeeringHandler, ts time.Time, asns ...int) { + h.HandleMessage(&ristypes.RISMessage{ + Path: ristypes.ASPath(asns), + ParsedTimestamp: ts, + }) +} + +// TestPeeringHandlerProcessesEachRunAndEmpties verifies that a processing run +// empties the path map (the swap) and that a path seen again after a run is +// recorded in the next run too. +func TestPeeringHandlerProcessesEachRunAndEmpties(t *testing.T) { + store := &recordingStore{mockStore: newMockStore()} + h := newTestHandler(store) + + now := time.Now().UTC() + + // Run 1: one path, one peering recorded, map emptied afterwards. + handle(h, now, testASNA, testASNB) + h.ProcessPeeringsNow() + + if tracked, _ := snapshot(h); tracked != 0 { + t.Fatalf("map not empty after first run: %d paths remain", tracked) + } + if got := store.callCount(); got != 1 { + t.Fatalf("want 1 RecordPeering call after first run, got %d", got) + } + + // Run 2: the same path again is recorded again (RecordPeering upserts). + handle(h, now.Add(time.Second), testASNA, testASNB) + h.ProcessPeeringsNow() + + if tracked, _ := snapshot(h); tracked != 0 { + t.Fatalf("map not empty after second run: %d paths remain", tracked) + } + if got := store.callCount(); got != 2 { + t.Fatalf("want 2 RecordPeering calls after second run, got %d", got) + } +} + +// TestPeeringHandlerCapDropsAndCounts verifies that a full map drops new paths +// and counts them, while a path already tracked is refreshed rather than +// dropped. +func TestPeeringHandlerCapDropsAndCounts(t *testing.T) { + store := &recordingStore{mockStore: newMockStore()} + h := newTestHandler(store) + + now := time.Now().UTC() + + // Fill the map to exactly maxTrackedPaths, including one real path key so + // the "already tracked" branch can be exercised. The filler keys are never + // processed in this test, so their contents do not matter. + existing := pathKey(t, testASNA, testASNB) + + h.mu.Lock() + h.asPaths[existing] = now + for i := 0; len(h.asPaths) < maxTrackedPaths; i++ { + h.asPaths[strconv.Itoa(i)] = now + } + h.mu.Unlock() + + // A new path is dropped and counted because the map is full. + handle(h, now.Add(time.Second), testASNA, testASNC) + + tracked, dropped := snapshot(h) + if tracked != maxTrackedPaths { + t.Fatalf("want map size %d after drop, got %d", maxTrackedPaths, tracked) + } + if dropped != 1 { + t.Fatalf("want dropped count 1, got %d", dropped) + } + + // A path already tracked is refreshed, not dropped. + refreshed := now.Add(2 * time.Second) + handle(h, refreshed, testASNA, testASNB) + + tracked, dropped = snapshot(h) + if tracked != maxTrackedPaths { + t.Fatalf("want map size %d after refresh, got %d", maxTrackedPaths, tracked) + } + if dropped != 1 { + t.Fatalf("want dropped count still 1 after refresh, got %d", dropped) + } + + h.mu.Lock() + gotTS := h.asPaths[existing] + h.mu.Unlock() + if !gotTS.Equal(refreshed) { + t.Fatalf("existing path timestamp not refreshed: want %v, got %v", refreshed, gotTS) + } +} -- 2.54.0 From 6d8ae7f592e932911c24d651dc06ead72c724702 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 16:01:14 +0200 Subject: [PATCH 05/19] Add .editorconfig and Gitea CI workflow (closes #14) Copy .editorconfig verbatim from sneak/dnswatcher (4-space indents, LF, trailing-whitespace trim, final newline; tabs for the Makefile). Add .gitea/workflows/check.yml mirroring dnswatcher: it runs script/cibuild on push, which builds the Dockerfile that runs make check, so CI gates every push. The checkout action is pinned by commit SHA as that repo pins it. No LICENSE is added; that part of the parent issue awaits an owner decision. Model: opus-4-8 --- .editorconfig | 12 ++++++++++++ .gitea/workflows/check.yml | 9 +++++++++ 2 files changed, 21 insertions(+) create mode 100644 .editorconfig create mode 100644 .gitea/workflows/check.yml diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 0000000..2fe0ce0 --- /dev/null +++ b/.editorconfig @@ -0,0 +1,12 @@ +root = true + +[*] +indent_style = space +indent_size = 4 +end_of_line = lf +charset = utf-8 +trim_trailing_whitespace = true +insert_final_newline = true + +[Makefile] +indent_style = tab diff --git a/.gitea/workflows/check.yml b/.gitea/workflows/check.yml new file mode 100644 index 0000000..a55bc55 --- /dev/null +++ b/.gitea/workflows/check.yml @@ -0,0 +1,9 @@ +name: check +on: [push] +jobs: + check: + runs-on: ubuntu-latest + steps: + # actions/checkout v4.2.2, 2026-02-28 + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 + - run: script/cibuild -- 2.54.0 From cb2374f03373d3a7298d5209d9aeedccd8d4ce4c Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 16:12:41 +0200 Subject: [PATCH 06/19] Stop decoding unused Community and Raw fields of RIS messages (closes #9) Community and Raw are decoded from every live message and read by no handler; Raw is the hex of the whole BGP message. Each parsed message sits in up to four handler queues, so retaining them is a large share of queue memory. Tagging both json:"-" keeps them out of the decoded message while the fields handlers use still decode. A decode test over docs/message-examples.json confirms both stay empty and the used fields (Path, Announcements) still populate. Model: opus-4-8 --- internal/ristypes/ris.go | 40 +++++++++++---------- internal/ristypes/ris_test.go | 67 +++++++++++++++++++++++++++++++++++ 2 files changed, 89 insertions(+), 18 deletions(-) create mode 100644 internal/ristypes/ris_test.go diff --git a/internal/ristypes/ris.go b/internal/ristypes/ris.go index ba36b91..a6831a4 100644 --- a/internal/ristypes/ris.go +++ b/internal/ristypes/ris.go @@ -63,24 +63,28 @@ type RISLiveMessage struct { // the actual BGP update data including AS path, communities, announcements, // and withdrawals. type RISMessage struct { - Type string `json:"type"` - Timestamp float64 `json:"timestamp"` - ParsedTimestamp time.Time `json:"-"` // Parsed from Timestamp field - Peer string `json:"peer"` - PeerASN string `json:"peer_asn"` - ID string `json:"id"` - Host string `json:"host"` - RRC string `json:"rrc,omitempty"` - MrtTime float64 `json:"mrt_time,omitempty"` - SocketTime float64 `json:"socket_time,omitempty"` - Path ASPath `json:"path,omitempty"` - Community [][]int `json:"community,omitempty"` - Origin string `json:"origin,omitempty"` - MED *int `json:"med,omitempty"` - LocalPref *int `json:"local_pref,omitempty"` - Announcements []RISAnnouncement `json:"announcements,omitempty"` - Withdrawals []string `json:"withdrawals,omitempty"` - Raw string `json:"raw,omitempty"` + Type string `json:"type"` + Timestamp float64 `json:"timestamp"` + ParsedTimestamp time.Time `json:"-"` // Parsed from Timestamp field + Peer string `json:"peer"` + PeerASN string `json:"peer_asn"` + ID string `json:"id"` + Host string `json:"host"` + RRC string `json:"rrc,omitempty"` + MrtTime float64 `json:"mrt_time,omitempty"` + SocketTime float64 `json:"socket_time,omitempty"` + Path ASPath `json:"path,omitempty"` + // Community and Raw are present in the feed but read by no handler. + // They are the largest fields on a message that lives in up to four + // handler queues, so json:"-" keeps them out of the decoded message + // to save queue memory. Do not decode them without a consumer. + Community [][]int `json:"-"` + Origin string `json:"origin,omitempty"` + MED *int `json:"med,omitempty"` + LocalPref *int `json:"local_pref,omitempty"` + Announcements []RISAnnouncement `json:"announcements,omitempty"` + Withdrawals []string `json:"withdrawals,omitempty"` + Raw string `json:"-"` } // RISAnnouncement represents a BGP route announcement within a RIS message. diff --git a/internal/ristypes/ris_test.go b/internal/ristypes/ris_test.go new file mode 100644 index 0000000..6a3b7b0 --- /dev/null +++ b/internal/ristypes/ris_test.go @@ -0,0 +1,67 @@ +package ristypes + +import ( + "bufio" + "encoding/json" + "os" + "testing" +) + +// messageExamplesPath is the captured RIS Live feed used as decode fixtures, +// one JSON message per line. +const messageExamplesPath = "../../docs/message-examples.json" + +// TestDecodeDropsCommunityAndRaw decodes every captured message the way the +// streamer does and checks that the fields no handler reads (Community, Raw) +// stay empty while the fields handlers use (Path, Announcements) still decode. +func TestDecodeDropsCommunityAndRaw(t *testing.T) { + f, err := os.Open(messageExamplesPath) + if err != nil { + t.Fatalf("open fixtures: %v", err) + } + defer f.Close() + + var messages, withPath, withAnnouncements int + + scanner := bufio.NewScanner(f) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for scanner.Scan() { + line := scanner.Bytes() + if len(line) == 0 { + continue + } + + var wrapper RISLiveMessage + if err := json.Unmarshal(line, &wrapper); err != nil { + t.Fatalf("unmarshal message %d: %v", messages+1, err) + } + messages++ + + msg := wrapper.Data + if msg.Community != nil { + t.Errorf("message %d: Community decoded, want empty: %v", messages, msg.Community) + } + if msg.Raw != "" { + t.Errorf("message %d: Raw decoded, want empty", messages) + } + + if len(msg.Path) > 0 { + withPath++ + } + if len(msg.Announcements) > 0 { + withAnnouncements++ + } + } + if err := scanner.Err(); err != nil { + t.Fatalf("scan fixtures: %v", err) + } + + if messages == 0 { + t.Fatal("no messages decoded from fixtures") + } + // The fixtures include announcement messages; a used field must still decode, + // otherwise an empty Community/Raw would prove nothing. + if withPath == 0 || withAnnouncements == 0 { + t.Fatalf("used fields did not decode: withPath=%d withAnnouncements=%d", withPath, withAnnouncements) + } +} -- 2.54.0 From 63d62b7bc1cabb6ecd1f5acce60d201853bcbc87 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 16:29:32 +0200 Subject: [PATCH 07/19] Fix two goroutine leaks: stats handlers on timeout, streamer tickers on reconnect (closes #12) The stats handlers ran the database query in a goroutine that sent on an unbuffered channel. When the 4s request timeout won, nothing received and the goroutine blocked forever; the status page polls every 2s, so once the query exceeds the timeout every poll leaked one goroutine. Give both channels capacity 1 so the send always completes. The streamer started two ticker goroutines per connection that exited only with the streamer's lifetime context, leaking two on every reconnect. Scope them to a per-connection context cancelled when the stream call returns. Tests force the stats timeout repeatedly and drive many reconnects, then assert the goroutine count settles back to its starting value. The streamer gains an internal endpoint field so a test can point it at a local server. Model: opus-4-8 --- internal/server/handlers.go | 16 +++-- internal/server/handlers_test.go | 99 ++++++++++++++++++++++++++++++ internal/streamer/streamer.go | 15 ++++- internal/streamer/streamer_test.go | 70 +++++++++++++++++++++ 4 files changed, 191 insertions(+), 9 deletions(-) create mode 100644 internal/server/handlers_test.go diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 98ff6b2..d04d0fd 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -179,9 +179,11 @@ func (s *Server) handleStatusJSON() http.HandlerFunc { metrics := s.streamer.GetMetrics() - // Get database stats with timeout - statsChan := make(chan database.Stats) - errChan := make(chan error) + // Get database stats with timeout. The channels are buffered so the + // goroutine's send never blocks if the timeout wins and nothing here + // receives; otherwise it would block forever and leak. + statsChan := make(chan database.Stats, 1) + errChan := make(chan error, 1) go func() { dbStats, err := s.db.GetStatsContext(ctx) @@ -398,9 +400,11 @@ func (s *Server) handleStats() http.HandlerFunc { metrics := s.streamer.GetMetrics() - // Get database stats with timeout - statsChan := make(chan database.Stats) - errChan := make(chan error) + // Get database stats with timeout. The channels are buffered so the + // goroutine's send never blocks if the timeout wins and nothing here + // receives; otherwise it would block forever and leak. + statsChan := make(chan database.Stats, 1) + errChan := make(chan error, 1) go func() { dbStats, err := s.db.GetStatsContext(ctx) diff --git a/internal/server/handlers_test.go b/internal/server/handlers_test.go new file mode 100644 index 0000000..2ac0f43 --- /dev/null +++ b/internal/server/handlers_test.go @@ -0,0 +1,99 @@ +package server + +import ( + "context" + "net/http" + "net/http/httptest" + "runtime" + "testing" + "time" + + "git.eeqj.de/sneak/routewatch/internal/database" + "git.eeqj.de/sneak/routewatch/internal/logger" + "git.eeqj.de/sneak/routewatch/internal/metrics" + "git.eeqj.de/sneak/routewatch/internal/streamer" +) + +// blockingStatsDB embeds database.Store (left nil) and overrides only +// GetStatsContext, which blocks until release is closed. The stats handlers +// call it in a goroutine; every other Store method is unused on the timeout +// path and would panic if called. +type blockingStatsDB struct { + database.Store + release chan struct{} +} + +func (d blockingStatsDB) GetStatsContext(_ context.Context) (database.Stats, error) { + <-d.release + + return database.Stats{}, nil +} + +// TestStatsHandlersDoNotLeakOnTimeout drives each stats handler repeatedly with +// a request whose context times out before the database responds, then releases +// the blocked queries and asserts the goroutine count returns to its starting +// value. Before the fix the per-request goroutine sent on an unbuffered channel +// that nothing received once the timeout won, so it blocked forever and every +// poll leaked one goroutine. +func TestStatsHandlersDoNotLeakOnTimeout(t *testing.T) { + release := make(chan struct{}) + db := blockingStatsDB{release: release} + s := New(db, streamer.New(logger.New(), metrics.New()), logger.New()) + + handlers := map[string]http.HandlerFunc{ + "status.json": s.handleStatusJSON(), + "stats": s.handleStats(), + } + + baseline := settledGoroutineCount() + + const ( + iterations = 20 + requestTimeout = 50 * time.Millisecond + ) + for _, handler := range handlers { + for range iterations { + ctx, cancel := context.WithTimeout(context.Background(), requestTimeout) + req := httptest.NewRequest(http.MethodGet, "/", nil).WithContext(ctx) + handler(httptest.NewRecorder(), req) + cancel() + } + } + + // Let the blocked queries finish; with buffered channels each goroutine's + // send now succeeds and the goroutine exits. + close(release) + + if !waitForGoroutines(baseline) { + t.Fatalf("goroutines did not return to baseline %d, got %d", + baseline, runtime.NumGoroutine()) + } +} + +// settledGoroutineCount lets transient goroutines finish, then reports the +// current count. +func settledGoroutineCount() int { + prev := runtime.NumGoroutine() + for range 20 { + time.Sleep(10 * time.Millisecond) + cur := runtime.NumGoroutine() + if cur == prev { + return cur + } + prev = cur + } + + return prev +} + +// waitForGoroutines waits until the goroutine count drops to target or below. +func waitForGoroutines(target int) bool { + for range 100 { + if runtime.NumGoroutine() <= target { + return true + } + time.Sleep(10 * time.Millisecond) + } + + return false +} diff --git a/internal/streamer/streamer.go b/internal/streamer/streamer.go index d345a9a..42d6547 100644 --- a/internal/streamer/streamer.go +++ b/internal/streamer/streamer.go @@ -106,6 +106,7 @@ type handlerInfo struct { type Streamer struct { logger *logger.Logger client *http.Client + url string handlers []*handlerInfo rawHandler RawMessageHandler mu sync.RWMutex @@ -124,6 +125,7 @@ type Streamer struct { func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer { return &Streamer{ logger: logger, + url: risLiveURL, client: &http.Client{ Timeout: 0, // No timeout for streaming Transport: &http.Transport{ @@ -463,7 +465,14 @@ func (s *Streamer) streamWithReconnect(ctx context.Context) { } func (s *Streamer) stream(ctx context.Context) error { - req, err := http.NewRequestWithContext(ctx, "GET", risLiveURL, nil) + // connCtx is scoped to this single connection: cancelling it when stream + // returns stops the ticker goroutines below, so a reconnect does not leak + // them. Without this they would live until the streamer's lifetime context + // is cancelled, leaking two per reconnect. + connCtx, connCancel := context.WithCancel(ctx) + defer connCancel() + + req, err := http.NewRequestWithContext(ctx, "GET", s.url, nil) if err != nil { return fmt.Errorf("failed to create request: %w", err) } @@ -516,7 +525,7 @@ func (s *Streamer) stream(ctx context.Context) error { select { case <-metricsTicker.C: s.logMetrics() - case <-ctx.Done(): + case <-connCtx.Done(): return } } @@ -536,7 +545,7 @@ func (s *Streamer) stream(ctx context.Context) error { s.metrics.RecordWireBytes(delta) lastWireBytes = currentBytes } - case <-ctx.Done(): + case <-connCtx.Done(): return } } diff --git a/internal/streamer/streamer_test.go b/internal/streamer/streamer_test.go index 1c788a4..3f0b4aa 100644 --- a/internal/streamer/streamer_test.go +++ b/internal/streamer/streamer_test.go @@ -1,7 +1,12 @@ package streamer import ( + "context" + "net/http" + "net/http/httptest" + "runtime" "testing" + "time" "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/metrics" @@ -32,3 +37,68 @@ func TestNewStreamer(t *testing.T) { t.Error("metrics tracker not set correctly") } } + +// TestStreamDoesNotLeakTickersAcrossReconnects drives many short-lived +// connections (each stream call is one reconnect cycle) and asserts the +// goroutine count returns to its starting value. Each connection starts two +// ticker goroutines; before the fix they lived until the streamer's lifetime +// context was cancelled, so every reconnect leaked two. +func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) { + // The handler returns immediately, so the response body is empty and each + // stream call ends at once, standing in for a dropped connection. + srv := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {})) + defer srv.Close() + + s := New(logger.New(), metrics.New()) + s.url = srv.URL + + // One warm-up connection so any persistent HTTP transport goroutine exists + // before we take the baseline. + if err := s.stream(context.Background()); err != nil { + t.Fatalf("warm-up stream returned error: %v", err) + } + s.client.CloseIdleConnections() + + baseline := settledGoroutineCount() + + const reconnects = 20 + for range reconnects { + if err := s.stream(context.Background()); err != nil { + t.Fatalf("stream returned error: %v", err) + } + } + s.client.CloseIdleConnections() + + if !waitForGoroutines(baseline) { + t.Fatalf("goroutines did not return to baseline %d after %d reconnects, got %d", + baseline, reconnects, runtime.NumGoroutine()) + } +} + +// settledGoroutineCount lets transient goroutines finish, then reports the +// current count. +func settledGoroutineCount() int { + prev := runtime.NumGoroutine() + for range 20 { + time.Sleep(10 * time.Millisecond) + cur := runtime.NumGoroutine() + if cur == prev { + return cur + } + prev = cur + } + + return prev +} + +// waitForGoroutines waits until the goroutine count drops to target or below. +func waitForGoroutines(target int) bool { + for range 100 { + if runtime.NumGoroutine() <= target { + return true + } + time.Sleep(10 * time.Millisecond) + } + + return false +} -- 2.54.0 From efaf79c4e3364fc4e8a589fb0d1a1b20c65f4389 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 16:29:39 +0200 Subject: [PATCH 08/19] Shrink the four handler queues from 100,000 to 20,000 (closes #11) Each handler queue held 100,000 message pointers; a message is retained until the slowest handler drains it, so all four full was a derived worst case near 800 MiB. Twenty thousand is about four seconds of feed at peak and caps that at roughly 160 MiB. The streamer already drops rather than blocks on a full queue, so the smaller bound is safe. Batch sizes are unchanged. The largest, asnBatchSize, is 30,000 and now exceeds its queue, but each queued message contributes every ASN in its path, and every handler also flushes on its own timer regardless of fill, so batches still flush and no size change is warranted. Model: opus-4-8 --- internal/routewatch/ashandler.go | 8 +++++--- internal/routewatch/peerhandler.go | 7 +++++-- internal/routewatch/peeringhandler.go | 8 ++++---- internal/routewatch/prefixhandler.go | 8 +++++--- 4 files changed, 19 insertions(+), 12 deletions(-) diff --git a/internal/routewatch/ashandler.go b/internal/routewatch/ashandler.go index c1ff34f..3eae6f3 100644 --- a/internal/routewatch/ashandler.go +++ b/internal/routewatch/ashandler.go @@ -10,9 +10,11 @@ import ( ) const ( - // asHandlerQueueSize is the queue capacity for ASN operations - // DO NOT set this higher than 100000 without explicit instructions - asHandlerQueueSize = 100000 + // asHandlerQueueSize is the queue capacity for ASN operations, about 4 + // seconds of feed at peak. The streamer drops rather than blocks when a + // queue is full, so this bounds memory. Batches still flush on a timer + // (asnBatchTimeout), so a queue smaller than asnBatchSize is fine. + asHandlerQueueSize = 20000 // asnBatchSize is the number of ASN operations to batch together asnBatchSize = 30000 diff --git a/internal/routewatch/peerhandler.go b/internal/routewatch/peerhandler.go index 7ffabab..2f7bf70 100644 --- a/internal/routewatch/peerhandler.go +++ b/internal/routewatch/peerhandler.go @@ -14,8 +14,11 @@ import ( ) const ( - // peerHandlerQueueSize is the queue capacity for peer tracking operations - peerHandlerQueueSize = 100000 + // peerHandlerQueueSize is the queue capacity for peer tracking operations, + // about 4 seconds of feed at peak. The streamer drops rather than blocks + // when a queue is full, so this bounds memory. Batches still flush on a + // timer (peerBatchTimeout). + peerHandlerQueueSize = 20000 // peerBatchSize is the number of peer updates to batch together peerBatchSize = 10000 diff --git a/internal/routewatch/peeringhandler.go b/internal/routewatch/peeringhandler.go index 820ccde..b4f3277 100644 --- a/internal/routewatch/peeringhandler.go +++ b/internal/routewatch/peeringhandler.go @@ -11,10 +11,10 @@ import ( ) const ( - // peeringHandlerQueueSize defines the buffer capacity for the peering - // handler's message queue. This should be large enough to handle bursts - // of BGP UPDATE messages without blocking. - peeringHandlerQueueSize = 100000 + // peeringHandlerQueueSize is the buffer capacity for the peering handler's + // message queue, about 4 seconds of feed at peak. The streamer drops + // rather than blocks when a queue is full, so this bounds memory. + peeringHandlerQueueSize = 20000 // minPathLengthForPeering specifies the minimum number of ASNs required // in a BGP AS path to extract peering relationships. A path with fewer diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 4223b20..9ce5675 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -14,9 +14,11 @@ import ( ) const ( - // prefixHandlerQueueSize is the queue capacity for prefix tracking operations - // DO NOT set this higher than 100000 without explicit instructions - prefixHandlerQueueSize = 100000 + // prefixHandlerQueueSize is the queue capacity for prefix tracking + // operations, about 4 seconds of feed at peak. The streamer drops rather + // than blocks when a queue is full, so this bounds memory. Batches still + // flush on a timer (prefixBatchTimeout). + prefixHandlerQueueSize = 20000 // prefixBatchSize is the number of prefix updates to batch together prefixBatchSize = 25000 -- 2.54.0 From 658aadbb81a4ecd5c6960e9d6f82b8b72134ac5c Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 17:29:29 +0200 Subject: [PATCH 09/19] Set GOMEMLIMIT in the image and document the memory budget (closes #13) Add ENV GOMEMLIMIT=1536MiB to the runtime stage so the Go runtime keeps a 1.5 GiB soft heap limit. runuser preserves it the way it already does XDG_DATA_HOME, so the routewatch process inherits it. Add a Memory section to the README describing the budget (Go 1.5 GiB soft, SQLite 640 MiB pool cache and 1.5 GiB hard heap, ~0.2 GiB other), the required 5 GiB container limit, how to override GOMEMLIMIT, what happens at each limit, and the DEBUG=routewatch System stats line. Every sentence matches the behaviour already merged to next. Model: opus-4-8 --- Dockerfile | 5 +++++ README.md | 39 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index 04e60ea..46e3169 100644 --- a/Dockerfile +++ b/Dockerfile @@ -78,6 +78,11 @@ RUN chown -R routewatch:routewatch /app ENV XDG_DATA_HOME=/var/lib +# Cap the Go heap at 1.5 GiB so the runtime collects harder before the +# container's memory limit is reached. runuser preserves this the way it does +# XDG_DATA_HOME above. +ENV GOMEMLIMIT=1536MiB + # Expose HTTP port EXPOSE 8080 diff --git a/README.md b/README.md index 5bfe36c..e856eb4 100644 --- a/README.md +++ b/README.md @@ -165,14 +165,51 @@ bgp_peers(id, peer_ip, peer_asn, last_message_type, last_seen) Configuration is handled via environment variables and OS-specific paths: | Variable | Default | Description | -|----------|---------|-------------| +|----------|----------|-------------| | `PORT` | `8080` | HTTP server port | | `DEBUG` | (empty) | Set to `routewatch` for debug logging | +| `GOMEMLIMIT` | `1536MiB` (in the Docker image) | Go soft memory limit; see Memory | State directory (database location): - macOS: `~/Library/Application Support/routewatch/` - Linux: `/var/lib/routewatch/` or `~/.local/share/routewatch/` +## Memory + +The daemon holds a live routing table, so its memory grows with the size of the +data it tracks. The image sets ceilings that keep it inside a 5 GiB container. + +Budget: +- Go heap: a 1.5 GiB soft limit (`GOMEMLIMIT=1536MiB`, set in the image). +- SQLite: at most 640 MiB of page cache across the connection pool (64 MiB per + connection, 10 connections) and a 1.5 GiB hard heap limit for the C library. +- About 0.2 GiB for everything else in the runtime. + +Run the container with a memory limit of 5 GiB and swap disabled: + +```bash +docker run --memory=5g --memory-swap=5g ... +``` + +or the equivalent in your deployment tool. This leaves headroom above the +ceilings for spikes and the kernel page cache. + +Override the Go soft limit by setting `GOMEMLIMIT` in the environment (for +example `-e GOMEMLIMIT=1GiB`); this replaces the image default. + +What happens at each limit: +- Go soft limit: as the heap approaches `GOMEMLIMIT`, the runtime runs garbage + collection more aggressively rather than growing further. +- SQLite: at a 1 GiB soft heap limit it recycles its page cache instead of + allocating more; at the 1.5 GiB hard heap limit a statement fails with an + out-of-memory error, and the handler logs it and drops that batch. The process + keeps running. +- Handler queues: each of the four handler queues holds at most 20,000 messages. + When a queue fills, the streamer drops messages instead of blocking. + +With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds +with the goroutine count and Go memory figures. + ## Development ```bash -- 2.54.0 From 08f060045b25eac516db4cb5d7d8fadba3604c03 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 19:29:31 +0200 Subject: [PATCH 10/19] Cap glibc malloc arenas to keep non-Go RSS bounded (closes #23) After the earlier memory work, SQLite's live heap is bounded but process RSS still climbed about 70 MiB/min, nearly all anonymous and outside the Go runtime and outside SQLite's own accounting, without any SQLITE_NOMEM. The cause is glibc: the SQLite C library allocates and frees millions of small page-cache chunks from many threads, and glibc keeps each arena's freed chunks resident. With arenas uncapped it creates up to eight per core, so on a large host RSS grows with the core count. Set MALLOC_ARENA_MAX=2 in the image to bound the retained memory. Writes are already serialized, so the two-arena cap costs no throughput. Update the README Memory section. Model: opus-4-8 --- Dockerfile | 8 ++++++++ README.md | 12 +++++++++++- 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index 46e3169..4112847 100644 --- a/Dockerfile +++ b/Dockerfile @@ -83,6 +83,14 @@ ENV XDG_DATA_HOME=/var/lib # XDG_DATA_HOME above. ENV GOMEMLIMIT=1536MiB +# Cap glibc's malloc arenas. The SQLite C library allocates and frees millions +# of small page-cache chunks from many threads; glibc otherwise creates up to +# eight arenas per core (hundreds on a large host) and keeps each arena's freed +# chunks resident, so process RSS climbs far above SQLite's live heap and never +# comes back down. Two arenas keep that retained memory bounded; database writes +# are already serialized, so the lost allocator concurrency costs nothing here. +ENV MALLOC_ARENA_MAX=2 + # Expose HTTP port EXPOSE 8080 diff --git a/README.md b/README.md index e856eb4..aaf5f9a 100644 --- a/README.md +++ b/README.md @@ -169,6 +169,7 @@ Configuration is handled via environment variables and OS-specific paths: | `PORT` | `8080` | HTTP server port | | `DEBUG` | (empty) | Set to `routewatch` for debug logging | | `GOMEMLIMIT` | `1536MiB` (in the Docker image) | Go soft memory limit; see Memory | +| `MALLOC_ARENA_MAX` | `2` (in the Docker image) | glibc malloc arena cap; see Memory | State directory (database location): - macOS: `~/Library/Application Support/routewatch/` @@ -183,6 +184,13 @@ Budget: - Go heap: a 1.5 GiB soft limit (`GOMEMLIMIT=1536MiB`, set in the image). - SQLite: at most 640 MiB of page cache across the connection pool (64 MiB per connection, 10 connections) and a 1.5 GiB hard heap limit for the C library. +- glibc allocator: the SQLite C library runs on glibc `malloc`, which frees + page-cache chunks back to per-arena free lists rather than to the kernel, so + process RSS tracks the high-water mark of those arenas, not SQLite's live + heap. glibc creates up to eight arenas per core, so on a many-core host the + retained memory — and thus RSS — grows with the core count. The image sets + `MALLOC_ARENA_MAX=2` to bound it; the two-arena cap costs nothing here because + database writes are already serialized. - About 0.2 GiB for everything else in the runtime. Run the container with a memory limit of 5 GiB and swap disabled: @@ -195,7 +203,9 @@ or the equivalent in your deployment tool. This leaves headroom above the ceilings for spikes and the kernel page cache. Override the Go soft limit by setting `GOMEMLIMIT` in the environment (for -example `-e GOMEMLIMIT=1GiB`); this replaces the image default. +example `-e GOMEMLIMIT=1GiB`); this replaces the image default. `MALLOC_ARENA_MAX` +can be overridden the same way, but raising it lets RSS climb again on a +many-core host. What happens at each limit: - Go soft limit: as the heap approaches `GOMEMLIMIT`, the runtime runs garbage -- 2.54.0 From 3898daad4ee19ee59e51dbc6f92bed285d20997d Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 20:12:33 +0200 Subject: [PATCH 11/19] Add _txlock=immediate so batch writes wait instead of dropping (closes #25) Batch flush paths read before they write, so a deferred transaction starts as a reader and must upgrade to the write lock on its first INSERT/UPDATE. When the background maintainer holds the write lock for a WAL checkpoint, that upgrade fails immediately with "database is locked" and the busy timeout does not apply, so the batch is dropped. Adding _txlock=immediate to the DSN makes every transaction take the write lock at BEGIN, so it waits up to busy_timeout instead of failing. A regression test drives batch writes against a running checkpoint loop and fails with "database is locked" without the change. Model: opus-4-8 --- internal/database/database.go | 8 +++- internal/database/database_test.go | 64 ++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 2 deletions(-) diff --git a/internal/database/database.go b/internal/database/database.go index 629c1ed..d7a7be4 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -88,9 +88,13 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { } // Per-connection SQLite settings go in the DSN so every pooled connection - // gets them, not just the one that runs the Initialize pragmas. + // gets them, not just the one that runs the Initialize pragmas. _txlock= + // immediate makes every transaction take the write lock at BEGIN. Without it + // a transaction that reads before writing starts as a reader and, when it + // then writes while another connection holds the write lock, fails at once + // with "database is locked" without waiting for _busy_timeout. dsn := fmt.Sprintf( - "file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d&_journal_mode=WAL", + "file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d&_journal_mode=WAL&_txlock=immediate", dbPath, sqliteCacheSizeKiB, sqliteBusyTimeoutMs, diff --git a/internal/database/database_test.go b/internal/database/database_test.go index 2bbe938..fa4bb48 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -4,7 +4,9 @@ import ( "context" "database/sql" "net" + "sync" "testing" + "time" "git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/logger" @@ -18,6 +20,18 @@ const tempStoreMemory = 2 // once so each is a distinct SQLite connection that parsed the DSN. const heldConnections = 5 +// 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 +) + func TestIPToUint32(t *testing.T) { tests := []struct { name string @@ -361,6 +375,56 @@ func TestConnectionPoolPragmas(t *testing.T) { } } +// TestBatchWriteDuringCheckpoint reproduces issue #25. A batch write reads +// (SELECT) before it writes (INSERT/UPDATE). Under the default deferred locking +// the transaction begins as a reader and, when the maintainer's WAL checkpoint +// holds the write lock, its upgrade to writer fails immediately with "database +// is locked" without honouring busy_timeout, dropping the batch. With +// _txlock=immediate the transaction takes the write lock at BEGIN and waits, so +// no batch is dropped. The checkpoint runs without the Database mutex, exactly +// as the background maintainer does in production. +func TestBatchWriteDuringCheckpoint(t *testing.T) { + 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.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-ctx.Done(): + return + default: + _ = db.Checkpoint(ctx) // errors are the checkpoint's own to absorb + } + } + }() + + ts := time.Now().UTC() + for i := 0; i < contentionIterations; i++ { + asns := map[int]time.Time{ + i % contendedASNCount: ts, + (i % contendedASNCount) + asnSecondBand: ts, + } + if err := db.GetOrCreateASNBatch(asns); err != nil { + cancel() + wg.Wait() + t.Fatalf("batch write failed under checkpoint contention: %v", err) + } + } + + cancel() + wg.Wait() +} + func BenchmarkIPToUint32(b *testing.B) { ip := net.ParseIP("192.168.1.1") b.ResetTimer() -- 2.54.0 From df9e23d5039237c47e2462102969b7d5b2c799a9 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Tue, 22 Sep 2026 09:41:16 +0200 Subject: [PATCH 12/19] Serve /api/v1/stats counts from realtime in-memory counters (closes #27) Realtime in-memory counters seeded at startup and adjusted on every insert, update and delete; no periodic recompute. Independent review passed: https://git.eeqj.de/sneak/routewatch/pulls/29#issuecomment-98622 model: claude-opus-4-8 (implementation and review); merged by claude-fable-5 --- TODO.md | 4 + internal/database/counts.go | 146 +++++++++ internal/database/counts_test.go | 335 ++++++++++++++++++++ internal/database/database.go | 525 +++++++++++++++++++------------ internal/database/interface.go | 2 + internal/server/handlers.go | 22 +- 6 files changed, 812 insertions(+), 222 deletions(-) create mode 100644 internal/database/counts.go create mode 100644 internal/database/counts_test.go diff --git a/TODO.md b/TODO.md index e4df8b9..9e25bd4 100644 --- a/TODO.md +++ b/TODO.md @@ -23,6 +23,10 @@ runs make check on main. # Completed Steps +- 2026-09-22: realtime in-memory database statistics: counts seeded at + startup and adjusted on every write, oldest/newest route timestamps via + index-end lookups; `/api/v1/stats` no longer scans the tables (closes + #27) - 2026-07-07 Adopted scripts-to-rule-them-all: `script/` entrypoints, Makefile shims, README Entrypoints section - 2026-02-22: repo policy compliance: required policy files, .gitignore diff --git a/internal/database/counts.go b/internal/database/counts.go new file mode 100644 index 0000000..a305ba2 --- /dev/null +++ b/internal/database/counts.go @@ -0,0 +1,146 @@ +package database + +import ( + "context" + "fmt" + "sync" +) + +// liveCounts holds the running row counts that the stats endpoints report. They +// are seeded once at startup from the tables and then adjusted on every write, +// so a stats read serves them from memory instead of running a COUNT(*) over +// each table. Those scans, once the database passed a few GiB, took the whole +// request timeout and made /api/v1/stats return 500 (issue 27). +// +// A single mutex guards all fields so the stats reader takes a consistent +// snapshot at one instant and writers, which already run under the database +// write lock, adjust the counts after their transaction commits. +type liveCounts struct { + mu sync.RWMutex + asns int + prefixesV4 int + prefixesV6 int + peerings int + peers int + routesV4 int + routesV6 int +} + +// seed sets every count to the value read from the tables at startup. It runs +// before any writer, so it needs no coordination with the adjust methods. +func (c *liveCounts) seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6 int) { + c.mu.Lock() + defer c.mu.Unlock() + + c.asns = asns + c.prefixesV4 = prefixesV4 + c.prefixesV6 = prefixesV6 + c.peerings = peerings + c.peers = peers + c.routesV4 = routesV4 + c.routesV6 = routesV6 +} + +// addASNs adds n to the ASN count. +func (c *liveCounts) addASNs(n int) { + c.mu.Lock() + c.asns += n + c.mu.Unlock() +} + +// addPrefixes adds to the IPv4 and IPv6 prefix counts. +func (c *liveCounts) addPrefixes(v4, v6 int) { + c.mu.Lock() + c.prefixesV4 += v4 + c.prefixesV6 += v6 + c.mu.Unlock() +} + +// addPeerings adds n to the peering count. +func (c *liveCounts) addPeerings(n int) { + c.mu.Lock() + c.peerings += n + c.mu.Unlock() +} + +// addPeers adds n to the BGP peer count. +func (c *liveCounts) addPeers(n int) { + c.mu.Lock() + c.peers += n + c.mu.Unlock() +} + +// addRoutes adds to the IPv4 and IPv6 live-route counts. Deletions pass +// negative values. +func (c *liveCounts) addRoutes(v4, v6 int) { + c.mu.Lock() + c.routesV4 += v4 + c.routesV6 += v6 + c.mu.Unlock() +} + +// fill copies the counts into a Stats, including the derived totals, under a +// single read lock so the reader sees one consistent snapshot. +func (c *liveCounts) fill(s *Stats) { + c.mu.RLock() + defer c.mu.RUnlock() + + s.ASNs = c.asns + s.IPv4Prefixes = c.prefixesV4 + s.IPv6Prefixes = c.prefixesV6 + s.Prefixes = c.prefixesV4 + c.prefixesV6 + s.Peerings = c.peerings + s.Peers = c.peers + s.IPv4Routes = c.routesV4 + s.IPv6Routes = c.routesV6 + s.LiveRoutes = c.routesV4 + c.routesV6 +} + +// countRows returns the number of rows in the named table. It is used only at +// startup to seed the in-memory counters, so a full COUNT(*) is acceptable. +func (d *Database) countRows(ctx context.Context, table string) (int, error) { + var n int + // table is one of a fixed set of literals below, never external input. + if err := d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+table).Scan(&n); err != nil { + return 0, fmt.Errorf("failed to count %s: %w", table, err) + } + + return n, nil +} + +// seedCounts reads the current row counts from the tables into the in-memory +// counters. It runs once at startup, before the streamer begins writing. +func (d *Database) seedCounts(ctx context.Context) error { + asns, err := d.countRows(ctx, "asns") + if err != nil { + return err + } + prefixesV4, err := d.countRows(ctx, "prefixes_v4") + if err != nil { + return err + } + prefixesV6, err := d.countRows(ctx, "prefixes_v6") + if err != nil { + return err + } + peerings, err := d.countRows(ctx, "peerings") + if err != nil { + return err + } + peers, err := d.countRows(ctx, "bgp_peers") + if err != nil { + return err + } + routesV4, err := d.countRows(ctx, "live_routes_v4") + if err != nil { + return err + } + routesV6, err := d.countRows(ctx, "live_routes_v6") + if err != nil { + return err + } + + d.counts.seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6) + + return nil +} diff --git a/internal/database/counts_test.go b/internal/database/counts_test.go new file mode 100644 index 0000000..6db2859 --- /dev/null +++ b/internal/database/counts_test.go @@ -0,0 +1,335 @@ +package database + +import ( + "context" + "sync" + "testing" + "time" + + "git.eeqj.de/sneak/routewatch/internal/config" + "git.eeqj.de/sneak/routewatch/internal/logger" + "github.com/google/uuid" +) + +// mkV4Route builds an IPv4 live route with its range columns populated. +func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute { + t.Helper() + + start, end, err := CalculateIPv4Range(prefix) + if err != nil { + t.Fatalf("CalculateIPv4Range(%s): %v", prefix, err) + } + + return &LiveRoute{ + ID: uuid.New(), + Prefix: prefix, + MaskLength: 24, + IPVersion: ipVersionV4, + OriginASN: asn, + PeerIP: "192.0.2.1", + ASPath: []int{asn}, + NextHop: "192.0.2.254", + LastUpdated: ts, + V4IPStart: &start, + V4IPEnd: &end, + } +} + +// mkV6Route builds an IPv6 live route. +func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute { + return &LiveRoute{ + ID: uuid.New(), + Prefix: prefix, + MaskLength: 32, + IPVersion: ipVersionV6, + OriginASN: asn, + PeerIP: "2001:db8::1", + ASPath: []int{asn}, + NextHop: "2001:db8::ffff", + LastUpdated: ts, + } +} + +// TestLiveCountsTrackWritesInRealtime checks that the stats counts start at +// zero, reflect each write the moment it commits (no recompute, no timer), do +// not move when a route is merely re-announced, and drop when a route is +// deleted. These counts are what /api/v1/stats reports; before this change the +// endpoint recomputed them with a COUNT(*) over each table on every request. +func TestLiveCountsTrackWritesInRealtime(t *testing.T) { + 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() + + empty, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext on empty database: %v", err) + } + if empty.ASNs != 0 || empty.Prefixes != 0 || empty.Peerings != 0 || + empty.Peers != 0 || empty.LiveRoutes != 0 { + t.Fatalf("empty database counts nonzero: %+v", empty) + } + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + if err := db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts}); err != nil { + t.Fatalf("GetOrCreateASNBatch: %v", err) + } + if err := db.UpdatePrefixesBatch(map[string]time.Time{ + "198.51.100.0/24": ts, + "2001:db8::/32": ts, + }); err != nil { + t.Fatalf("UpdatePrefixesBatch: %v", err) + } + if err := db.UpdatePeerBatch(map[string]PeerUpdate{ + "192.0.2.1": {PeerIP: "192.0.2.1", PeerASN: 64500, MessageType: "UPDATE", Timestamp: ts}, + }); err != nil { + t.Fatalf("UpdatePeerBatch: %v", err) + } + if err := db.RecordPeering(64500, 64501, ts); err != nil { + t.Fatalf("RecordPeering: %v", err) + } + + routes := []*LiveRoute{ + mkV4Route(t, "198.51.100.0/24", 64500, ts), + mkV4Route(t, "203.0.113.0/24", 64501, ts.Add(time.Minute)), + mkV6Route("2001:db8::/32", 64502, ts.Add(2*time.Minute)), + } + if err := db.UpsertLiveRouteBatch(routes); err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + stats, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + assertCounts(t, "after inserts", stats, wantCounts{ + asns: 2, prefixes: 2, peerings: 1, peers: 1, + ipv4Routes: 2, ipv6Routes: 1, liveRoutes: 3, + }) + + // Re-announcing the same routes is an update, not an insert: counts hold. + if err := db.UpsertLiveRouteBatch(routes); err != nil { + t.Fatalf("UpsertLiveRouteBatch (re-announce): %v", err) + } + stats, err = db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + assertCounts(t, "after re-announce", stats, wantCounts{ + asns: 2, prefixes: 2, peerings: 1, peers: 1, + ipv4Routes: 2, ipv6Routes: 1, liveRoutes: 3, + }) + + // A withdrawal removes one route. + if err := db.DeleteLiveRouteBatch([]LiveRouteDeletion{ + {Prefix: "203.0.113.0/24", OriginASN: 64501, PeerIP: "192.0.2.1", IPVersion: ipVersionV4}, + }); err != nil { + t.Fatalf("DeleteLiveRouteBatch: %v", err) + } + stats, err = db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + assertCounts(t, "after delete", stats, wantCounts{ + asns: 2, prefixes: 2, peerings: 1, peers: 1, + ipv4Routes: 1, ipv6Routes: 1, liveRoutes: 2, + }) +} + +// TestLiveCountsSeededFromDatabaseAtStartup writes rows, reopens the same +// database file, and checks the counts come back from the seed scan rather than +// starting at zero. +func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) { + cfg := &config.Config{StateDir: t.TempDir()} + + db, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + if err := db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts, 64502: ts}); err != nil { + t.Fatalf("GetOrCreateASNBatch: %v", err) + } + if err := db.UpsertLiveRouteBatch([]*LiveRoute{ + mkV4Route(t, "198.51.100.0/24", 64500, ts), + mkV6Route("2001:db8::/32", 64502, ts), + }); err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + if err := db.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + reopened, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to reopen database: %v", err) + } + defer func() { _ = reopened.Close() }() + + stats, err := reopened.GetStatsContext(context.Background()) + if err != nil { + t.Fatalf("GetStatsContext after reopen: %v", err) + } + if stats.ASNs != 3 { + t.Errorf("seeded ASNs = %d, want 3", stats.ASNs) + } + if stats.IPv4Routes != 1 || stats.IPv6Routes != 1 || stats.LiveRoutes != 2 { + t.Errorf("seeded routes = (v4 %d, v6 %d, total %d), want (1, 1, 2)", + stats.IPv4Routes, stats.IPv6Routes, stats.LiveRoutes) + } +} + +// TestStatsRouteTimestamps checks the oldest/newest route timestamps are read +// from the right rows across both tables and parse into time.Time. The old +// MIN/MAX union query read its result into *time.Time, which the driver could +// not parse, so it logged a warning every call and left both timestamps nil. +func TestStatsRouteTimestamps(t *testing.T) { + 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() + + empty, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext on empty database: %v", err) + } + if empty.OldestRoute != nil || empty.NewestRoute != nil { + t.Fatalf("empty database timestamps = (%v, %v), want (nil, nil)", + empty.OldestRoute, empty.NewestRoute) + } + + base := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + oldest := base + newest := base.Add(2 * time.Minute) + if err := db.UpsertLiveRouteBatch([]*LiveRoute{ + mkV4Route(t, "198.51.100.0/24", 64500, base.Add(time.Minute)), + mkV4Route(t, "203.0.113.0/24", 64501, oldest), + mkV6Route("2001:db8::/32", 64502, newest), + }); err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + stats, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + if stats.OldestRoute == nil || !stats.OldestRoute.Equal(oldest) { + t.Errorf("OldestRoute = %v, want %v", stats.OldestRoute, oldest) + } + if stats.NewestRoute == nil || !stats.NewestRoute.Equal(newest) { + t.Errorf("NewestRoute = %v, want %v", stats.NewestRoute, newest) + } +} + +// TestLiveCountsConcurrentReadWrite runs writers and stats readers at once so +// the race detector proves the counters are safe under concurrent use. +func TestLiveCountsConcurrentReadWrite(t *testing.T) { + 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) + + const writers = 4 + var wg sync.WaitGroup + + wg.Add(writers) + for w := range writers { + go func(base int) { + defer wg.Done() + for i := range 25 { + asn := 65000 + base*100 + i + route := mkV6Route("2001:db8::/32", asn, ts) + route.PeerIP = "2001:db8::" + uuid.NewString() + if err := db.UpsertLiveRoute(route); err != nil { + t.Errorf("UpsertLiveRoute: %v", err) + + return + } + } + }(w) + } + + var readerWG sync.WaitGroup + readerWG.Add(1) + stop := make(chan struct{}) + go func() { + defer readerWG.Done() + for { + select { + case <-stop: + return + default: + if _, err := db.GetStatsContext(context.Background()); err != nil { + t.Errorf("GetStatsContext: %v", err) + + return + } + } + } + }() + + wg.Wait() + close(stop) + readerWG.Wait() + + stats, err := db.GetStatsContext(context.Background()) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + if want := writers * 25; stats.IPv6Routes != want { + t.Errorf("IPv6Routes = %d, want %d", stats.IPv6Routes, want) + } +} + +type wantCounts struct { + asns int + prefixes int + peerings int + peers int + ipv4Routes int + ipv6Routes int + liveRoutes int +} + +func assertCounts(t *testing.T, when string, got Stats, want wantCounts) { + t.Helper() + + if got.ASNs != want.asns { + t.Errorf("%s: ASNs = %d, want %d", when, got.ASNs, want.asns) + } + if got.Prefixes != want.prefixes { + t.Errorf("%s: Prefixes = %d, want %d", when, got.Prefixes, want.prefixes) + } + if got.Peerings != want.peerings { + t.Errorf("%s: Peerings = %d, want %d", when, got.Peerings, want.peerings) + } + if got.Peers != want.peers { + t.Errorf("%s: Peers = %d, want %d", when, got.Peers, want.peers) + } + if got.IPv4Routes != want.ipv4Routes { + t.Errorf("%s: IPv4Routes = %d, want %d", when, got.IPv4Routes, want.ipv4Routes) + } + if got.IPv6Routes != want.ipv6Routes { + t.Errorf("%s: IPv6Routes = %d, want %d", when, got.IPv6Routes, want.ipv6Routes) + } + if got.LiveRoutes != want.liveRoutes { + t.Errorf("%s: LiveRoutes = %d, want %d", when, got.LiveRoutes, want.liveRoutes) + } +} diff --git a/internal/database/database.go b/internal/database/database.go index d7a7be4..4007b48 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -72,6 +72,7 @@ type Database struct { mu sync.Mutex lockedAt time.Time lockedBy string + counts *liveCounts } // New creates a new database connection and initializes the schema. @@ -115,12 +116,19 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { db.SetMaxIdleConns(maxConns) db.SetConnMaxLifetime(0) - database := &Database{db: db, logger: logger, path: dbPath} + database := &Database{db: db, logger: logger, path: dbPath, counts: &liveCounts{}} if err := database.Initialize(); err != nil { return nil, fmt.Errorf("failed to initialize database: %w", err) } + // Seed the in-memory statistics counters from the tables once, before the + // streamer starts writing. From here on every write keeps them current, so + // the stats endpoints never scan the tables to report counts. + if err := database.seedCounts(context.Background()); err != nil { + return nil, fmt.Errorf("failed to seed statistics counters: %w", err) + } + return database, nil } @@ -208,6 +216,81 @@ func (d *Database) beginTx() (*loggingTx, error) { return &loggingTx{Tx: tx, logger: d.logger}, nil } +// A live-route upsert is an UPDATE followed, only when no row matched, by an +// INSERT. The UPDATE's rows-affected count (1 for an existing key, 0 for a new +// one) is what lets the in-memory route counters stay exact without a COUNT(*). +// Callers hold the database write lock, so no other writer can insert the same +// key between the two statements. The id column is set only on INSERT, so an +// updated route keeps its original id, exactly as the previous ON CONFLICT +// upsert did. +const ( + updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?, + last_updated = ?, ip_start = ?, ip_end = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?` + + insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip, + as_path, next_hop, last_updated, ip_start, ip_end) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + + updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?, + last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?` + + insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip, + as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` +) + +// upsertRouteRowV4 updates an IPv4 live route, inserting it when no row matched, +// and reports whether a new row was inserted. +func upsertRouteRowV4(upd, ins *sql.Stmt, route *LiveRoute, pathJSON string) (inserted bool, err error) { + if route.V4IPStart == nil || route.V4IPEnd == nil { + return false, fmt.Errorf("IPv4 route %s missing range values", route.Prefix) + } + + res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated, + *route.V4IPStart, *route.V4IPEnd, route.Prefix, route.OriginASN, route.PeerIP) + if err != nil { + return false, err + } + affected, err := res.RowsAffected() + if err != nil { + return false, err + } + if affected > 0 { + return false, nil + } + + _, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN, + route.PeerIP, pathJSON, route.NextHop, route.LastUpdated, *route.V4IPStart, *route.V4IPEnd) + if err != nil { + return false, err + } + + return true, nil +} + +// upsertRouteRowV6 updates an IPv6 live route, inserting it when no row matched, +// and reports whether a new row was inserted. +func upsertRouteRowV6(upd, ins *sql.Stmt, route *LiveRoute, pathJSON string) (inserted bool, err error) { + res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated, + route.Prefix, route.OriginASN, route.PeerIP) + if err != nil { + return false, err + } + affected, err := res.RowsAffected() + if err != nil { + return false, err + } + if affected > 0 { + return false, nil + } + + _, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN, + route.PeerIP, pathJSON, route.NextHop, route.LastUpdated) + if err != nil { + return false, err + } + + return true, nil +} + // UpsertLiveRouteBatch inserts or updates multiple live routes in a single transaction func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error { if len(routes) == 0 { @@ -227,91 +310,65 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error { } }() - // Prepare statements for both IPv4 and IPv6 - queryV4 := ` - INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, - last_updated, ip_start, ip_end) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET - mask_length = excluded.mask_length, - as_path = excluded.as_path, - next_hop = excluded.next_hop, - last_updated = excluded.last_updated, - ip_start = excluded.ip_start, - ip_end = excluded.ip_end - ` - - queryV6 := ` - INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, - last_updated) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET - mask_length = excluded.mask_length, - as_path = excluded.as_path, - next_hop = excluded.next_hop, - last_updated = excluded.last_updated - ` - - stmtV4, err := tx.Prepare(queryV4) + // Prepare the update and insert statements for both tables. + updV4, err := tx.Prepare(updateLiveRouteV4SQL) if err != nil { - return fmt.Errorf("failed to prepare IPv4 statement: %w", err) + return fmt.Errorf("failed to prepare IPv4 update statement: %w", err) } - defer func() { _ = stmtV4.Close() }() + defer func() { _ = updV4.Close() }() - stmtV6, err := tx.Prepare(queryV6) + insV4, err := tx.Prepare(insertLiveRouteV4SQL) if err != nil { - return fmt.Errorf("failed to prepare IPv6 statement: %w", err) + return fmt.Errorf("failed to prepare IPv4 insert statement: %w", err) } - defer func() { _ = stmtV6.Close() }() + defer func() { _ = insV4.Close() }() + updV6, err := tx.Prepare(updateLiveRouteV6SQL) + if err != nil { + return fmt.Errorf("failed to prepare IPv6 update statement: %w", err) + } + defer func() { _ = updV6.Close() }() + + insV6, err := tx.Prepare(insertLiveRouteV6SQL) + if err != nil { + return fmt.Errorf("failed to prepare IPv6 insert statement: %w", err) + } + defer func() { _ = insV6.Close() }() + + var newV4, newV6 int for _, route := range routes { - // Encode AS path as JSON pathJSON, err := json.Marshal(route.ASPath) if err != nil { return fmt.Errorf("failed to encode AS path: %w", err) } - // Use appropriate statement based on IP version if route.IPVersion == ipVersionV4 { - // IPv4 routes must have range values - if route.V4IPStart == nil || route.V4IPEnd == nil { - return fmt.Errorf("IPv4 route %s missing range values", route.Prefix) + inserted, err := upsertRouteRowV4(updV4, insV4, route, string(pathJSON)) + if err != nil { + return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) + } + if inserted { + newV4++ } - _, err = stmtV4.Exec( - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - *route.V4IPStart, - *route.V4IPEnd, - ) - } else { - // IPv6 routes - _, err = stmtV6.Exec( - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - ) + continue } + + inserted, err := upsertRouteRowV6(updV6, insV6, route, string(pathJSON)) if err != nil { return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) } + if inserted { + newV6++ + } } if err = tx.Commit(); err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } + d.counts.addRoutes(newV4, newV6) + return nil } @@ -362,6 +419,7 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { defer func() { _ = stmtV6WithoutOrigin.Close() }() // Process deletions + var deletedV4, deletedV6 int64 for _, del := range deletions { var stmt *sql.Stmt @@ -382,21 +440,36 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error { } // Execute deletion + var res sql.Result if del.OriginASN == 0 { - _, err = stmt.Exec(del.Prefix, del.PeerIP) + res, err = stmt.Exec(del.Prefix, del.PeerIP) } else { - _, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP) + res, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP) } if err != nil { return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err) } + + // A deletion with no origin ASN can remove several rows, so use the + // exact rows-affected count to keep the in-memory route counters right. + affected, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err) + } + if del.IPVersion == ipVersionV4 { + deletedV4 += affected + } else { + deletedV6 += affected + } } if err = tx.Commit(); err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } + d.counts.addRoutes(-int(deletedV4), -int(deletedV6)) + return nil } @@ -456,6 +529,7 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { } defer func() { _ = insertV6Stmt.Close() }() + var newV4, newV6 int for prefix, timestamp := range prefixes { ipVersion := detectIPVersion(prefix) @@ -482,6 +556,11 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { if err != nil { return fmt.Errorf("failed to insert prefix %s: %w", prefix, err) } + if ipVersion == ipVersionV4 { + newV4++ + } else { + newV6++ + } default: return fmt.Errorf("failed to query prefix %s: %w", prefix, err) } @@ -491,6 +570,8 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { return fmt.Errorf("failed to commit transaction: %w", err) } + d.counts.addPrefixes(newV4, newV6) + return nil } @@ -534,6 +615,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error { } defer func() { _ = insertStmt.Close() }() + var newASNs int for number, timestamp := range asns { var asn ASN var handle, description sql.NullString @@ -568,6 +650,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error { if err != nil { return fmt.Errorf("failed to insert ASN %d: %w", number, err) } + newASNs++ continue } @@ -581,6 +664,8 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error { return fmt.Errorf("failed to commit transaction: %w", err) } + d.counts.addASNs(newASNs) + return nil } @@ -652,6 +737,8 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error) return nil, err } + d.counts.addASNs(1) + return &asn, nil } @@ -727,6 +814,12 @@ func (d *Database) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefi return nil, err } + if ipVersion == ipVersionV4 { + d.counts.addPrefixes(1, 0) + } else { + d.counts.addPrefixes(0, 1) + } + return &p, nil } @@ -803,6 +896,10 @@ func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error { return err } + if !exists { + d.counts.addPeerings(1) + } + return nil } @@ -846,6 +943,7 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error { } defer func() { _ = insertStmt.Close() }() + var newPeers int for _, update := range peers { var exists bool err = checkStmt.QueryRow(update.PeerIP).Scan(&exists) @@ -863,12 +961,17 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error { if err != nil { return fmt.Errorf("failed to update peer %s: %w", update.PeerIP, err) } + if !exists { + newPeers++ + } } if err = tx.Commit(); err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } + d.counts.addPeers(newPeers) + return nil } @@ -918,6 +1021,10 @@ func (d *Database) UpdatePeer(peerIP string, peerASN int, messageType string, ti return err } + if !exists { + d.counts.addPeers(1) + } + return nil } @@ -926,169 +1033,161 @@ func (d *Database) GetStats() (Stats, error) { return d.GetStatsContext(context.Background()) } -// GetStatsContext returns database statistics with context support +// GetStatsContext returns database statistics with context support. +// +// The row counts (ASNs, prefixes, peerings, peers, live routes) come from the +// in-memory counters, seeded at startup and kept current on every write, so a +// read runs no COUNT(*) over the tables. The oldest/newest route timestamps are +// read from the ends of the last_updated index, and the file size from a +// stat(); neither is a table scan. The only remaining query is the prefix +// distribution. func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) { var stats Stats - // Count ASNs - err := d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM asns").Scan(&stats.ASNs) - if err != nil { - return stats, err - } + // Row counts from memory, as a single consistent snapshot. + d.counts.fill(&stats) - // Count prefixes from both tables - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM prefixes_v4").Scan(&stats.IPv4Prefixes) - if err != nil { - return stats, err - } - - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM prefixes_v6").Scan(&stats.IPv6Prefixes) - if err != nil { - return stats, err - } - - stats.Prefixes = stats.IPv4Prefixes + stats.IPv6Prefixes - - // Count peerings - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM peerings").Scan(&stats.Peerings) - if err != nil { - return stats, err - } - - // Count peers - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM bgp_peers").Scan(&stats.Peers) - if err != nil { - return stats, err - } - - // Get database file size - fileInfo, err := os.Stat(d.path) - if err != nil { + // Database file size is a cheap stat() on the file. + if fileInfo, err := os.Stat(d.path); err != nil { d.logger.Warn("Failed to get database file size", "error", err) - stats.FileSizeBytes = 0 } else { stats.FileSizeBytes = fileInfo.Size() } - // Get live routes count from both tables - var v4Count, v6Count int - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v4").Scan(&v4Count) - if err != nil { - return stats, fmt.Errorf("failed to count IPv4 routes: %w", err) - } - err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v6").Scan(&v6Count) - if err != nil { - return stats, fmt.Errorf("failed to count IPv6 routes: %w", err) - } - stats.LiveRoutes = v4Count + v6Count - - // Get oldest and newest route timestamps - routeTimestampQuery := ` - SELECT MIN(last_updated), MAX(last_updated) FROM ( - SELECT last_updated FROM live_routes_v4 - UNION ALL - SELECT last_updated FROM live_routes_v6 - ) - ` - var oldestRoute, newestRoute *time.Time - err = d.db.QueryRowContext(ctx, routeTimestampQuery).Scan(&oldestRoute, &newestRoute) + // Oldest and newest route timestamps read one row from each end of the + // last_updated index (a log-time lookup, not a scan). Selecting the column + // directly lets the driver parse the DATETIME into time.Time; the old + // MIN/MAX union scan returned an untyped string that failed to scan and + // logged a warning on every call. + oldest, newest, err := d.routeTimestampRange(ctx) if err != nil { + // Display-only fields; log but keep the rest of the stats. d.logger.Warn("Failed to get route timestamps", "error", err) } else { - stats.OldestRoute = oldestRoute - stats.NewestRoute = newestRoute + stats.OldestRoute = oldest + stats.NewestRoute = newest } - // Get prefix distribution + // Prefix distribution counts distinct prefixes per mask length. It stays a + // query over the covering (mask_length, prefix) index rather than an + // in-memory counter: maintaining distinct-prefix-per-mask in memory would + // need a per-prefix table of roughly a million entries, memory this service + // is tuned to avoid. stats.IPv4PrefixDistribution, stats.IPv6PrefixDistribution, err = d.GetPrefixDistributionContext(ctx) if err != nil { - // Log but don't fail + // Log but don't fail. d.logger.Warn("Failed to get prefix distribution", "error", err) } return stats, nil } +// routeTimestampRange returns the earliest and latest last_updated across both +// live route tables, or nil values when both tables are empty. Each query reads +// one row from an end of the last_updated index rather than scanning the tables. +func (d *Database) routeTimestampRange(ctx context.Context) (oldest, newest *time.Time, err error) { + oldestV4, ok, err := d.scanRouteTimestamp(ctx, + "SELECT last_updated FROM live_routes_v4 ORDER BY last_updated ASC LIMIT 1") + if err != nil { + return nil, nil, err + } + if ok { + oldest = &oldestV4 + } + + oldestV6, ok, err := d.scanRouteTimestamp(ctx, + "SELECT last_updated FROM live_routes_v6 ORDER BY last_updated ASC LIMIT 1") + if err != nil { + return nil, nil, err + } + if ok && (oldest == nil || oldestV6.Before(*oldest)) { + oldest = &oldestV6 + } + + newestV4, ok, err := d.scanRouteTimestamp(ctx, + "SELECT last_updated FROM live_routes_v4 ORDER BY last_updated DESC LIMIT 1") + if err != nil { + return nil, nil, err + } + if ok { + newest = &newestV4 + } + + newestV6, ok, err := d.scanRouteTimestamp(ctx, + "SELECT last_updated FROM live_routes_v6 ORDER BY last_updated DESC LIMIT 1") + if err != nil { + return nil, nil, err + } + if ok && (newest == nil || newestV6.After(*newest)) { + newest = &newestV6 + } + + return oldest, newest, nil +} + +// scanRouteTimestamp runs a single-row timestamp query. ok is false when the +// table is empty. The query selects the last_updated column directly so the +// driver parses the DATETIME value into a time.Time. +func (d *Database) scanRouteTimestamp(ctx context.Context, query string) (ts time.Time, ok bool, err error) { + err = d.db.QueryRowContext(ctx, query).Scan(&ts) + switch { + case errors.Is(err, sql.ErrNoRows): + return time.Time{}, false, nil + case err != nil: + return time.Time{}, false, err + default: + return ts, true, nil + } +} + // UpsertLiveRoute inserts or updates a live route func (d *Database) UpsertLiveRoute(route *LiveRoute) error { d.lock("UpsertLiveRoute") defer d.unlock() - // Choose table based on IP version - tableName := "live_routes_v4" - if route.IPVersion == ipVersionV6 { - tableName = "live_routes_v6" - } - - var query string - if route.IPVersion == ipVersionV4 { - query = fmt.Sprintf(` - INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, - last_updated, ip_start, ip_end) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET - mask_length = excluded.mask_length, - as_path = excluded.as_path, - next_hop = excluded.next_hop, - last_updated = excluded.last_updated, - ip_start = excluded.ip_start, - ip_end = excluded.ip_end - `, tableName) - } else { - query = fmt.Sprintf(` - INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, - last_updated) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET - mask_length = excluded.mask_length, - as_path = excluded.as_path, - next_hop = excluded.next_hop, - last_updated = excluded.last_updated - `, tableName) - } - - // Encode AS path as JSON pathJSON, err := json.Marshal(route.ASPath) if err != nil { return fmt.Errorf("failed to encode AS path: %w", err) } - if route.IPVersion == ipVersionV4 { - // Convert v4_ip_start and v4_ip_end to interface{} for SQL NULL handling - var v4Start, v4End interface{} - if route.V4IPStart != nil { - v4Start = *route.V4IPStart - } - if route.V4IPEnd != nil { - v4End = *route.V4IPEnd - } - - _, err = d.db.Exec(query, - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - v4Start, - v4End, - ) - } else { - _, err = d.db.Exec(query, - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - ) + updateSQL, insertSQL := updateLiveRouteV4SQL, insertLiveRouteV4SQL + if route.IPVersion == ipVersionV6 { + updateSQL, insertSQL = updateLiveRouteV6SQL, insertLiveRouteV6SQL } - return err + // The write lock is held, so no other writer can insert this key between the + // update and the insert even though they are separate autocommit statements. + upd, err := d.db.Prepare(updateSQL) + if err != nil { + return fmt.Errorf("failed to prepare update statement: %w", err) + } + defer func() { _ = upd.Close() }() + + ins, err := d.db.Prepare(insertSQL) + if err != nil { + return fmt.Errorf("failed to prepare insert statement: %w", err) + } + defer func() { _ = ins.Close() }() + + var inserted bool + if route.IPVersion == ipVersionV4 { + inserted, err = upsertRouteRowV4(upd, ins, route, string(pathJSON)) + } else { + inserted, err = upsertRouteRowV6(upd, ins, route, string(pathJSON)) + } + if err != nil { + return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) + } + + if inserted { + if route.IPVersion == ipVersionV4 { + d.counts.addRoutes(1, 0) + } else { + d.counts.addRoutes(0, 1) + } + } + + return nil } // DeleteLiveRoute deletes a live route @@ -1103,23 +1202,41 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) return fmt.Errorf("invalid prefix format: %w", err) } - tableName := "live_routes_v4" - if ipnet.IP.To4() == nil { - tableName = "live_routes_v6" + isV4 := ipnet.IP.To4() != nil + + // Literal per-table queries (rather than one formatted with the table name) + // so the delete carries no dynamically built SQL. A delete with no origin + // ASN can remove several rows. + var res sql.Result + switch { + case isV4 && originASN == 0: + res, err = d.db.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP) + case isV4: + res, err = d.db.Exec( + `DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, + prefix, originASN, peerIP) + case originASN == 0: + res, err = d.db.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP) + default: + res, err = d.db.Exec( + `DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, + prefix, originASN, peerIP) + } + if err != nil { + return err } - var query string - if originASN == 0 { - // Delete all routes for this prefix from this peer - query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND peer_ip = ?`, tableName) - _, err = d.db.Exec(query, prefix, peerIP) + affected, err := res.RowsAffected() + if err != nil { + return err + } + if isV4 { + d.counts.addRoutes(-int(affected), 0) } else { - // Delete specific route - query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, tableName) - _, err = d.db.Exec(query, prefix, originASN, peerIP) + d.counts.addRoutes(0, -int(affected)) } - return err + return nil } // GetPrefixDistribution returns the distribution of unique prefixes by mask length diff --git a/internal/database/interface.go b/internal/database/interface.go index 7f3f8e9..038dfd5 100644 --- a/internal/database/interface.go +++ b/internal/database/interface.go @@ -18,6 +18,8 @@ type Stats struct { Peers int FileSizeBytes int64 LiveRoutes int + IPv4Routes int + IPv6Routes int OldestRoute *time.Time NewestRoute *time.Time IPv4PrefixDistribution []PrefixDistribution diff --git a/internal/server/handlers.go b/internal/server/handlers.go index d04d0fd..195cb33 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -219,13 +219,6 @@ func (s *Server) handleStatusJSON() http.HandlerFunc { const bitsPerMegabit = 1000000.0 - // Get route counts from database - ipv4Routes, ipv6Routes, err := s.db.GetLiveRouteCountsContext(ctx) - if err != nil { - s.logger.Warn("Failed to get live route counts", "error", err) - // Continue with zero counts - } - // Get route update metrics routeMetrics := s.streamer.GetMetricsTracker().GetRouteMetrics() @@ -259,8 +252,8 @@ func (s *Server) handleStatusJSON() http.HandlerFunc { Peers: dbStats.Peers, DatabaseSizeBytes: dbStats.FileSizeBytes, LiveRoutes: dbStats.LiveRoutes, - IPv4Routes: ipv4Routes, - IPv6Routes: ipv6Routes, + IPv4Routes: dbStats.IPv4Routes, + IPv6Routes: dbStats.IPv6Routes, OldestRoute: dbStats.OldestRoute, NewestRoute: dbStats.NewestRoute, IPv4UpdatesPerSec: routeMetrics.IPv4UpdatesPerSec, @@ -439,13 +432,6 @@ func (s *Server) handleStats() http.HandlerFunc { const bitsPerMegabit = 1000000.0 - // Get route counts from database - ipv4Routes, ipv6Routes, err := s.db.GetLiveRouteCountsContext(ctx) - if err != nil { - s.logger.Warn("Failed to get live route counts", "error", err) - // Continue with zero counts - } - // Get route update metrics routeMetrics := s.streamer.GetMetricsTracker().GetRouteMetrics() @@ -537,8 +523,8 @@ func (s *Server) handleStats() http.HandlerFunc { Peers: dbStats.Peers, DatabaseSizeBytes: dbStats.FileSizeBytes, LiveRoutes: dbStats.LiveRoutes, - IPv4Routes: ipv4Routes, - IPv6Routes: ipv6Routes, + IPv4Routes: dbStats.IPv4Routes, + IPv6Routes: dbStats.IPv6Routes, OldestRoute: dbStats.OldestRoute, NewestRoute: dbStats.NewestRoute, IPv4UpdatesPerSec: routeMetrics.IPv4UpdatesPerSec, -- 2.54.0 From f2a9e90625f1e648622933cf406c495196279990 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 28 Sep 2026 20:08:42 +0200 Subject: [PATCH 13/19] Refuse invalid settings at start, health check follows PORT (closes #31) A set but invalid PORT (anything but plain digits from 1 to 65535) or a relative XDG_DATA_HOME now stops the start before the database opens; before, a bad PORT left the daemon running without HTTP. entrypoint.sh refuses a MALLOC_ARENA_MAX that is not a positive whole number, since glibc ignores a bad one silently. The HEALTHCHECK probes the port PORT names, 8080 when unset. PORT is now read in internal/config, so server.New takes the config. The README gives the real Linux state directory, lists XDG_DATA_HOME, and adds "Running under upaas": port, volume, environment, the 5g memory limit and the health check. Unverified: the 5g memory limit could not be exercised on the build host. Model: opus-5-5 --- Dockerfile | 4 +- README.md | 31 ++++++++++- TODO.md | 4 ++ entrypoint.sh | 6 ++ internal/config/config.go | 41 +++++++++++++- internal/config/config_test.go | 62 +++++++++++++++++++++ internal/routewatch/app_integration_test.go | 2 +- internal/server/handlers_test.go | 3 +- internal/server/server.go | 16 +++--- 9 files changed, 152 insertions(+), 17 deletions(-) create mode 100644 internal/config/config_test.go diff --git a/Dockerfile b/Dockerfile index 4112847..83d9ed0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -96,8 +96,8 @@ EXPOSE 8080 COPY ./entrypoint.sh /entrypoint.sh -# Health check using the health endpoint +# Health check using the health endpoint, on the port PORT names HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ - CMD curl -sf http://localhost:8080/.well-known/healthcheck.json || exit 1 + CMD curl -sf "http://localhost:${PORT:-8080}/.well-known/healthcheck.json" || exit 1 ENTRYPOINT ["/bin/bash", "/entrypoint.sh" ] diff --git a/README.md b/README.md index aaf5f9a..e1cddf5 100644 --- a/README.md +++ b/README.md @@ -166,14 +166,21 @@ Configuration is handled via environment variables and OS-specific paths: | Variable | Default | Description | |----------|----------|-------------| -| `PORT` | `8080` | HTTP server port | +| `PORT` | `8080` | HTTP server port, a whole number from 1 to 65535 | | `DEBUG` | (empty) | Set to `routewatch` for debug logging | +| `XDG_DATA_HOME` | `/var/lib` (in the Docker image) | Base of the state directory; must be an absolute path | | `GOMEMLIMIT` | `1536MiB` (in the Docker image) | Go soft memory limit; see Memory | -| `MALLOC_ARENA_MAX` | `2` (in the Docker image) | glibc malloc arena cap; see Memory | +| `MALLOC_ARENA_MAX` | `2` (in the Docker image) | glibc malloc arena cap, a positive whole number; see Memory | + +A variable that is set to an invalid value stops the start with an error and a +non-zero exit. An empty variable counts as unset. State directory (database location): - macOS: `~/Library/Application Support/routewatch/` -- Linux: `/var/lib/routewatch/` or `~/.local/share/routewatch/` +- Linux: `/var/lib/berlin.sneak.app.routewatch/` when running as root, + otherwise `$XDG_DATA_HOME/berlin.sneak.app.routewatch/` (with `XDG_DATA_HOME` + unset, `~/.local/share/berlin.sneak.app.routewatch/`). In the Docker image + this is `/var/lib/berlin.sneak.app.routewatch/`. ## Memory @@ -220,6 +227,24 @@ What happens at each limit: With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds with the goroutine count and Go memory figures. +## Running under upaas + +What the [upaas](https://git.eeqj.de/sneak/upaas) app needs: + +- Container port: `8080`. +- Volume: one, at container path `/var/lib/berlin.sneak.app.routewatch`. upaas + does not create the host directory, so create it before the first deploy. The + entrypoint takes ownership of it, so a root-owned directory works. +- Environment: nothing is required. Leave `XDG_DATA_HOME`, `GOMEMLIMIT` and + `MALLOC_ARENA_MAX` at the image's values. `DEBUG=routewatch` is optional and + adds the `System stats` memory line to the log. +- Memory Limit: `5g`, the 5 GiB limit from Memory above. upaas sets no swap + limit, so on a host with swap Docker allows the same amount of swap again. +- Health check: the image's `HEALTHCHECK` requests + `/.well-known/healthcheck.json` on the container port. upaas reads the + container's health 60 seconds after a deploy and fails the deploy unless it + is `healthy`. + ## Development ```bash diff --git a/TODO.md b/TODO.md index 9e25bd4..fea8ac6 100644 --- a/TODO.md +++ b/TODO.md @@ -23,6 +23,10 @@ runs make check on main. # Completed Steps +- 2026-09-28: ready to run under upaas: a set but invalid `PORT`, + `XDG_DATA_HOME` or `MALLOC_ARENA_MAX` stops the start, the health + check follows `PORT`, README "Running under upaas" section (closes + #31) - 2026-09-22: realtime in-memory database statistics: counts seeded at startup and adjusted on every write, oldest/newest route timestamps via index-end lookups; `/api/v1/stats` no longer scans the tables (closes diff --git a/entrypoint.sh b/entrypoint.sh index 82acbbe..3f0d092 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -1,5 +1,11 @@ #!/bin/bash +# glibc silently ignores a malformed MALLOC_ARENA_MAX, so refuse it here. +if [[ -n "${MALLOC_ARENA_MAX:-}" && ! "$MALLOC_ARENA_MAX" =~ ^[1-9][0-9]*$ ]]; then + echo "MALLOC_ARENA_MAX must be a positive whole number, got '$MALLOC_ARENA_MAX'" >&2 + exit 1 +fi + cd /var/lib/berlin.sneak.app.routewatch chown -R routewatch:routewatch . chmod 700 . diff --git a/internal/config/config.go b/internal/config/config.go index 14ae95d..87a6ce3 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "runtime" + "strconv" "time" ) @@ -18,6 +19,12 @@ const ( // defaultRouteExpirationMinutes is the default route expiration timeout in minutes defaultRouteExpirationMinutes = 5 + + // defaultPort is the HTTP port used when PORT is not set + defaultPort = 8080 + + // maxPort is the highest TCP port number + maxPort = 65535 ) // Config holds configuration for the entire application @@ -25,6 +32,9 @@ type Config struct { // StateDir is the directory for all application state (database, snapshots) StateDir string + // Port is the TCP port the HTTP server listens on + Port int + // MaxRuntime is the maximum runtime (0 = run forever) MaxRuntime time.Duration @@ -43,8 +53,14 @@ func New() (*Config, error) { return nil, fmt.Errorf("failed to determine state directory: %w", err) } + port, err := getPort() + if err != nil { + return nil, err + } + return &Config{ StateDir: stateDir, + Port: port, MaxRuntime: 0, // Run forever by default EnableBatchedDatabaseWrites: true, // Enable batching by default RouteExpirationTimeout: defaultRouteExpirationMinutes * time.Minute, // For active route monitoring @@ -69,13 +85,20 @@ func getStateDirectory() (string, error) { return filepath.Join(home, "Library", "Application Support", AppIdentifier), nil case "linux", "freebsd", "openbsd", "netbsd": + // The XDG spec requires an absolute path; a relative one would put + // the database somewhere unexpected. + xdgData := os.Getenv("XDG_DATA_HOME") + if xdgData != "" && !filepath.IsAbs(xdgData) { + return "", fmt.Errorf("XDG_DATA_HOME must be an absolute path, got %q", xdgData) + } + // Unix-like: /var/lib/berlin.sneak.app.routewatch if root, else XDG_DATA_HOME if os.Geteuid() == 0 { return filepath.Join("/var/lib", AppIdentifier), nil } // Check XDG_DATA_HOME first - if xdgData := os.Getenv("XDG_DATA_HOME"); xdgData != "" { + if xdgData != "" { return filepath.Join(xdgData, AppIdentifier), nil } @@ -92,6 +115,22 @@ func getStateDirectory() (string, error) { } } +// getPort returns the HTTP port from PORT, or defaultPort when PORT is not set +func getPort() (int, error) { + value := os.Getenv("PORT") + if value == "" { + return defaultPort, nil + } + + // ParseUint, unlike Atoi, refuses a sign: the health check URL cannot use "+9090" + port, err := strconv.ParseUint(value, 10, 0) + if err != nil || port < 1 || port > maxPort { + return 0, fmt.Errorf("PORT must be a whole number from 1 to %d, got %q", maxPort, value) + } + + return int(port), nil +} + // EnsureDirectories creates all necessary directories if they don't exist func (c *Config) EnsureDirectories() error { // Ensure state directory exists diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..84f9414 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,62 @@ +package config + +import ( + "runtime" + "testing" +) + +func TestNewReadsPort(t *testing.T) { + tests := map[string]int{ + "": defaultPort, + "1": 1, + "9090": 9090, + "65535": 65535, + } + + for value, want := range tests { + t.Run(value, func(t *testing.T) { + t.Setenv("PORT", value) + t.Setenv("XDG_DATA_HOME", "") + + cfg, err := New() + if err != nil { + t.Fatalf("New() with PORT=%q: %v", value, err) + } + + if cfg.Port != want { + t.Errorf("New() with PORT=%q: Port = %d, want %d", value, cfg.Port, want) + } + }) + } +} + +func TestNewRefusesInvalidPort(t *testing.T) { + for _, value := range []string{"0", "65536", "-1", "+9090", "http", "80.5"} { + t.Run(value, func(t *testing.T) { + t.Setenv("PORT", value) + t.Setenv("XDG_DATA_HOME", "") + + if _, err := New(); err == nil { + t.Errorf("New() with PORT=%q returned no error", value) + } + }) + } +} + +func TestNewRefusesRelativeXDGDataHome(t *testing.T) { + if runtime.GOOS == "darwin" { + t.Skip("macOS does not read XDG_DATA_HOME") + } + + t.Setenv("PORT", "") + + t.Setenv("XDG_DATA_HOME", "relative/path") + if _, err := New(); err == nil { + t.Error("New() with a relative XDG_DATA_HOME returned no error") + } + + t.Setenv("XDG_DATA_HOME", "/var/lib") + if _, err := New(); err != nil { + t.Errorf("New() with XDG_DATA_HOME=/var/lib: %v", err) + } +} diff --git a/internal/routewatch/app_integration_test.go b/internal/routewatch/app_integration_test.go index 807a0bd..56efc06 100644 --- a/internal/routewatch/app_integration_test.go +++ b/internal/routewatch/app_integration_test.go @@ -450,7 +450,7 @@ func TestRouteWatchLiveFeed(t *testing.T) { } // Create server - srv := server.New(mockDB, s, logger) + srv := server.New(mockDB, s, logger, cfg) // Create RouteWatch with 5 second limit deps := Dependencies{ diff --git a/internal/server/handlers_test.go b/internal/server/handlers_test.go index 2ac0f43..46756bd 100644 --- a/internal/server/handlers_test.go +++ b/internal/server/handlers_test.go @@ -8,6 +8,7 @@ import ( "testing" "time" + "git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/database" "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/metrics" @@ -38,7 +39,7 @@ func (d blockingStatsDB) GetStatsContext(_ context.Context) (database.Stats, err func TestStatsHandlersDoNotLeakOnTimeout(t *testing.T) { release := make(chan struct{}) db := blockingStatsDB{release: release} - s := New(db, streamer.New(logger.New(), metrics.New()), logger.New()) + s := New(db, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{}) handlers := map[string]http.HandlerFunc{ "status.json": s.handleStatusJSON(), diff --git a/internal/server/server.go b/internal/server/server.go index afdd426..10f8649 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -4,9 +4,10 @@ package server import ( "context" "net/http" - "os" + "strconv" "time" + "git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/database" "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/streamer" @@ -33,16 +34,18 @@ type Server struct { db database.Store streamer *streamer.Streamer logger *logger.Logger + port int srv *http.Server asnFetcher ASNFetcher } // New creates a new HTTP server -func New(db database.Store, streamer *streamer.Streamer, logger *logger.Logger) *Server { +func New(db database.Store, streamer *streamer.Streamer, logger *logger.Logger, cfg *config.Config) *Server { s := &Server{ db: db, streamer: streamer, logger: logger, + port: cfg.Port, } s.setupRoutes() @@ -52,11 +55,6 @@ func New(db database.Store, streamer *streamer.Streamer, logger *logger.Logger) // Start starts the HTTP server func (s *Server) Start() error { - port := os.Getenv("PORT") - if port == "" { - port = "8080" - } - const ( readHeaderTimeout = 40 * time.Second readTimeout = 60 * time.Second @@ -65,7 +63,7 @@ func (s *Server) Start() error { ) s.srv = &http.Server{ - Addr: ":" + port, + Addr: ":" + strconv.Itoa(s.port), Handler: s.router, ReadHeaderTimeout: readHeaderTimeout, ReadTimeout: readTimeout, @@ -73,7 +71,7 @@ func (s *Server) Start() error { IdleTimeout: idleTimeout, } - s.logger.Info("Starting HTTP server", "port", port, "addr", s.srv.Addr) + s.logger.Info("Starting HTTP server", "port", s.port, "addr", s.srv.Addr) // Start in goroutine but log when actually listening go func() { -- 2.54.0 From 6187ac8503ab48284c0fd04ff7b158886aefd8d2 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 28 Sep 2026 21:08:33 +0200 Subject: [PATCH 14/19] Let the daemon receive docker stop's signal itself (closes #33) entrypoint.sh now switches to the routewatch user (UID 1000) with setpriv instead of runuser. setpriv replaces itself with the daemon, so the daemon is the container's main process and receives docker stop's signal itself. runuser stayed in between, passed the signal on and killed the daemon 2 seconds later, so every stop ended with exit 143. The daemon now gets the whole wait the caller allows, up to its own 60-second limit, and a clean stop exits 0. Taking ownership of the state directory and the MALLOC_ARENA_MAX check still run as root first. Not fixed here: a stop while the feed is flowing can still panic (https://git.eeqj.de/sneak/routewatch/issues/34). Model: opus-5-5 --- Dockerfile | 4 ++-- TODO.md | 5 +++++ entrypoint.sh | 5 ++++- 3 files changed, 11 insertions(+), 3 deletions(-) diff --git a/Dockerfile b/Dockerfile index 83d9ed0..cedd1c9 100644 --- a/Dockerfile +++ b/Dockerfile @@ -79,8 +79,8 @@ RUN chown -R routewatch:routewatch /app ENV XDG_DATA_HOME=/var/lib # Cap the Go heap at 1.5 GiB so the runtime collects harder before the -# container's memory limit is reached. runuser preserves this the way it does -# XDG_DATA_HOME above. +# container's memory limit is reached. setpriv in the entrypoint preserves this +# the way it does XDG_DATA_HOME above. ENV GOMEMLIMIT=1536MiB # Cap glibc's malloc arenas. The SQLite C library allocates and frees millions diff --git a/TODO.md b/TODO.md index fea8ac6..f74ae68 100644 --- a/TODO.md +++ b/TODO.md @@ -23,6 +23,11 @@ runs make check on main. # Completed Steps +- 2026-09-28: `docker stop` no longer kills the daemon 2 seconds after the + stop signal: the entrypoint switches to the `routewatch` user with + `setpriv` instead of `runuser`, so the daemon receives the signal itself + and gets the whole wait `docker stop` allows, up to its own 60-second + limit (closes #33) - 2026-09-28: ready to run under upaas: a set but invalid `PORT`, `XDG_DATA_HOME` or `MALLOC_ARENA_MAX` stops the start, the health check follows `PORT`, README "Running under upaas" section (closes diff --git a/entrypoint.sh b/entrypoint.sh index 3f0d092..7679fad 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -10,4 +10,7 @@ cd /var/lib/berlin.sneak.app.routewatch chown -R routewatch:routewatch . chmod 700 . -exec runuser -u routewatch -- /app/routewatch +# setpriv replaces itself with the daemon, so the daemon receives the stop +# signal directly. runuser would stay in between and kill the daemon 2 seconds +# after passing the signal on. +exec setpriv --reuid=routewatch --regid=routewatch --init-groups -- /app/routewatch -- 2.54.0 From 1ac24669d3c211f08442bac5359153d97384a6e3 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 28 Sep 2026 21:42:33 +0200 Subject: [PATCH 15/19] Stop the streamer without sending on closed queues (closes #34) Stopping the daemon while the RIS Live feed was flowing could panic with "send on closed channel" and skip the rest of the shutdown. Stop cancels the stream and closes the handler queues under the streamer's write lock, but the read loop checked for a stop only before parsing each line. It now checks again under the read lock it already takes just before handing a message to the queues, so it never sends to a closed queue. Stop also clears its cancel function and returns early when there is none, so a second call no longer closes the queues again. A test forces both cases. Behaviour change: Stop before Start now does nothing. Model: opus-5-5 --- TODO.md | 4 +++ internal/streamer/streamer.go | 18 ++++++++++--- internal/streamer/streamer_test.go | 42 ++++++++++++++++++++++++++++++ 3 files changed, 61 insertions(+), 3 deletions(-) diff --git a/TODO.md b/TODO.md index f74ae68..70ef510 100644 --- a/TODO.md +++ b/TODO.md @@ -23,6 +23,10 @@ runs make check on main. # Completed Steps +- 2026-09-28: stopping the daemon while the feed is flowing no longer + panics with "send on closed channel": the read loop checks for a stop + just before handing a message to the handler queues, and a second + `Stop` no longer closes the queues again (closes #34) - 2026-09-28: `docker stop` no longer kills the daemon 2 seconds after the stop signal: the entrypoint switches to the `routewatch` user with `setpriv` instead of `runuser`, so the daemon receives the signal itself diff --git a/internal/streamer/streamer.go b/internal/streamer/streamer.go index 42d6547..b8b75a7 100644 --- a/internal/streamer/streamer.go +++ b/internal/streamer/streamer.go @@ -210,9 +210,14 @@ func (s *Streamer) Start() error { // the connection status in metrics. This method is safe to call multiple times. func (s *Streamer) Stop() { s.mu.Lock() - if s.cancel != nil { - s.cancel() + if s.cancel == nil { + // Not started, or already stopped: closing the queues again would panic. + s.mu.Unlock() + + return } + s.cancel() + s.cancel = nil // Close all handler queues to signal workers to stop for _, info := range s.handlers { close(info.queue) @@ -660,8 +665,15 @@ func (s *Streamer) stream(ctx context.Context) error { continue } - // Dispatch to interested handlers + // Dispatch to interested handlers. Stop cancels ctx and closes the + // queues under the write lock, so if ctx is cancelled here, under the + // read lock, the queues are closed and must not be sent to. s.mu.RLock() + if ctx.Err() != nil { + s.mu.RUnlock() + + return ctx.Err() + } for _, info := range s.handlers { if !info.handler.WantsMessage(msg.Type) { continue diff --git a/internal/streamer/streamer_test.go b/internal/streamer/streamer_test.go index 3f0b4aa..5d9582b 100644 --- a/internal/streamer/streamer_test.go +++ b/internal/streamer/streamer_test.go @@ -2,6 +2,8 @@ package streamer import ( "context" + "errors" + "io" "net/http" "net/http/httptest" "runtime" @@ -10,6 +12,7 @@ import ( "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/metrics" + "git.eeqj.de/sneak/routewatch/internal/ristypes" ) func TestNewStreamer(t *testing.T) { @@ -75,6 +78,45 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) { } } +// updateHandler wants UPDATE messages and does nothing with them. +type updateHandler struct{} + +func (updateHandler) WantsMessage(messageType string) bool { return messageType == "UPDATE" } +func (updateHandler) HandleMessage(*ristypes.RISMessage) {} +func (updateHandler) QueueCapacity() int { return 10 } + +// TestStopBeforeMessageReachesQueues stops the streamer after the read loop +// has checked for cancellation but before it hands the message to the handler +// queues. That is the gap a stop from another goroutine can land in, and it +// used to end in "send on closed channel". The raw handler runs in that gap on +// the read loop itself, so calling Stop from it hits the gap every time. +func TestStopBeforeMessageReachesQueues(t *testing.T) { + const line = `{"type":"ris_message","data":{"type":"UPDATE","peer":"192.0.2.1",` + + `"peer_asn":"64496","timestamp":1700000000}}` + "\n" + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, line) + })) + defer srv.Close() + + s := New(logger.New(), metrics.New()) + s.url = srv.URL + s.RegisterHandler(updateHandler{}) + s.RegisterRawHandler(func(string) { s.Stop() }) + + // Start would run the stream in the background, where the test cannot + // wait for it. Setting cancel as Start does lets Stop cancel the stream + // run here instead. + ctx, cancel := context.WithCancel(context.Background()) + s.cancel = cancel + + if err := s.stream(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("stream returned %v, want %v", err, context.Canceled) + } + + // A second Stop must not close the queues again. + s.Stop() +} + // settledGoroutineCount lets transient goroutines finish, then reports the // current count. func settledGoroutineCount() int { -- 2.54.0 From d28a59023a327ad0061dc054d11a1968bc6fa818 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Tue, 29 Sep 2026 05:05:35 +0200 Subject: [PATCH 16/19] Add MIT LICENSE (closes #1) Adds LICENSE with the MIT licence sneak chose for routewatch, copied byte for byte from sneak/webhooker's LICENSE: copyright 2026, Jeffrey Paul. The rest of the issue, .editorconfig and the Gitea check workflow, was already on next, and the README already points at LICENSE. Judgement call: our repos write the copyright holder in several different ways and sneak named no form; webhooker's gives his full name and email address, the most explicit of them. Model: opus-5-5 --- LICENSE | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) create mode 100644 LICENSE diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..3274443 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Jeffrey Paul + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. -- 2.54.0 From 87a3bc115e4170ebee04ce23d57f8377497bd7fb Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Tue, 29 Sep 2026 07:06:30 +0200 Subject: [PATCH 17/19] README: license and author up front; bring TODO.md up to date (closes #38) Brings README.md and TODO.md in line with REPO_POLICIES.md and with what is on next. The README's first line now names routewatch, what it is, its MIT licence and its author, @sneak; the License section says MIT and links LICENSE; a new Author section names @sneak. TODO.md's Status, Next Step and Future Steps now describe next as it is: it waits for sneak to merge the milestone PR, after which the upaas deploy and the run under a real 5 GiB limit are his. Completed Steps gains a line for each issue that landed without one. Docs only. make fmt formats only Go here, so the markdown was wrapped by hand. Model: opus-5-5 --- README.md | 11 ++++++++-- TODO.md | 64 ++++++++++++++++++++++++++++++++++++++++++++----------- 2 files changed, 61 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index e1cddf5..6204266 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,9 @@ # RouteWatch -RouteWatch is a real-time BGP routing table monitor that streams BGP UPDATE messages from the RIPE RIS Live service, maintains a live routing table in SQLite, and provides HTTP APIs for querying routing information. +RouteWatch is an MIT-licensed Go daemon by @sneak that monitors the BGP routing +table in real time: it streams BGP UPDATE messages from the RIPE RIS Live +service, maintains a live routing table in SQLite, and provides HTTP APIs for +querying routing information. ## Features @@ -299,4 +302,8 @@ invoke `go test` directly without `-short`: ## License -See LICENSE file. +MIT. See [`LICENSE`](LICENSE). + +## Author + +[@sneak](https://sneak.berlin) diff --git a/TODO.md b/TODO.md index 70ef510..fbab888 100644 --- a/TODO.md +++ b/TODO.md @@ -10,19 +10,30 @@ # Status -pre-1.0. No git tags. Runs in production-style Docker deployment, but -the policy compliance branch (repo-policies-compliance, make check -passing, clean tree) is unmerged to main and the CI workflow is missing. +pre-1.0. No git tags. The Docker build runs the format check, the linter +and the tests, and the Gitea workflow runs that build on every push. The +image sets memory ceilings for a 5 GiB container (README "Memory") and the +README says how to run it under upaas (README "Running under upaas"). A +35-hour run of `3898daa` on the live feed peaked at about 1 GiB, without a +container memory limit. # Next Step -Merge repo-policies-compliance into main (3 commits: policy files and -.gitignore, Makefile targets fmt-check/check/docker/hooks, gofmt pass), -then add .gitea/workflows/check.yml as a small follow-up commit so CI -runs make check on main. +`next` waits for sneak to merge it to `main` through +https://git.eeqj.de/sneak/routewatch/pulls/6. After that, setting +routewatch up 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). +Nothing on the record orders +https://git.eeqj.de/sneak/routewatch/issues/39 and +https://git.eeqj.de/sneak/routewatch/issues/30. # Completed Steps +- 2026-09-29: README first line names the MIT license and the author; + the License section now says MIT and links `LICENSE`, and an Author + section was added; this file brought up to date (closes #38) +- 2026-09-29: MIT `LICENSE` (closes #1) - 2026-09-28: stopping the daemon while the feed is flowing no longer panics with "send on closed channel": the read loop checks for a stop just before handing a message to the handler queues, and a second @@ -40,6 +51,33 @@ runs make check on main. startup and adjusted on every write, oldest/newest route timestamps via index-end lookups; `/api/v1/stats` no longer scans the tables (closes #27) +- 2026-09-21: batch writes take the write lock when their transaction + begins (`_txlock=immediate`), so they wait out a WAL checkpoint instead + of failing with "database is locked" (closes #25) +- 2026-09-21: `MALLOC_ARENA_MAX=2` in the image caps glibc malloc arenas, + so memory outside the Go runtime no longer grows with the core count + (closes #23) +- 2026-09-21: `GOMEMLIMIT=1536MiB` in the image; README Memory section + with the memory budget and the 5 GiB container limit (closes #13) +- 2026-09-21: the four handler queues hold at most 20,000 messages each, + down from 100,000 (closes #11) +- 2026-09-21: two goroutine leaks fixed: the stats handlers after a + timeout and the streamer's tickers on every reconnect (closes #12) +- 2026-09-21: parsed RIS messages no longer keep the unused `Community` + and `Raw` fields (closes #9) +- 2026-09-21: `.editorconfig`, and a Gitea workflow that runs + `script/cibuild` on every push (closes #14) +- 2026-09-21: the peering handler's AS-path map holds at most 500,000 + paths and is swapped for an empty one every 30 seconds instead of copied + (closes #10) +- 2026-09-21: SQLite memory bounded across the whole connection pool: a + 64 MiB page cache on each connection, 1 GiB soft and 1.5 GiB hard heap + limits (closes #8) +- 2026-09-21: the Docker build runs the format check, the linter and the + tests, linting in a separate stage on a golangci-lint image pinned by + digest (closes #5) +- 2026-09-21: `make test` skips the live-network feed test (`-short`), so + `make check` no longer depends on the network (closes #2) - 2026-07-07 Adopted scripts-to-rule-them-all: `script/` entrypoints, Makefile shims, README Entrypoints section - 2026-02-22: repo policy compliance: required policy files, .gitignore @@ -58,8 +96,10 @@ runs make check on main. # Future Steps -- Verify main is green after the merge: make check locally and the new - CI workflow passing -- Review stale remote branches fix-min-time-calculation and - optimize-sqlite-settings: land or delete -- Clean up the tmp/ directory at the repo root: gitignore or remove +- Production memory under 5 GiB: whether to test under a real 5 GiB + container limit on fsn1app1 is open for sneak + (https://git.eeqj.de/sneak/routewatch/issues/3) +- `/api/v1/stats` answered HTTP 500 after 35 hours on the live feed, seen + on `3898daa`, which predates the 2026-09-22 in-memory statistics + (https://git.eeqj.de/sneak/routewatch/issues/30) +- Add a `.dockerignore` (https://git.eeqj.de/sneak/routewatch/issues/39) -- 2.54.0 From 057e0bd9a92b36fc5804a49288dd743be37ff5ec Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Tue, 29 Sep 2026 07:42:20 +0200 Subject: [PATCH 18/19] Add a .dockerignore (closes #39) Adds a .dockerignore, which REPO_POLICIES.md lists among the files every repo has. Until now every docker build sent the whole working tree as its build context, and the image's source archive is made from that context. It now keeps out .git, local build and test output, archives, local databases, .env and a local Go workspace. Every tracked file stays in, so the format check, the linter, the tests, the build and the source archive read the same files as before. Without .git in the context the binary's build info records no git revision; nothing in routewatch reads it. Model: opus-5-5 --- .dockerignore | 28 ++++++++++++++++++++++++++++ TODO.md | 8 ++++---- 2 files changed, 32 insertions(+), 4 deletions(-) create mode 100644 .dockerignore diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..063d4cc --- /dev/null +++ b/.dockerignore @@ -0,0 +1,28 @@ +# Docker does not read .gitignore, and a pattern here matches from the root of +# the build context only: a pattern meant for every directory needs `**/`. + +.git + +# Local build and debug output: `make build`, `make run`, `make asupdate`, test +# binaries, coverage profiles and source archives. +/bin +/log.txt +/out +/pkg/asinfo/asdata.json +**/*.tar.zst +**/*.test +**/*.out +**/*.tmp + +# Local databases and secrets. The image carries a source archive of the whole +# build context, so these would otherwise ship inside it. +**/*.db +**/*.db-journal +**/*.db-wal +**/.env + +# A local Go workspace points at directories outside the build context. +/go.work +/go.work.sum + +**/.DS_Store diff --git a/TODO.md b/TODO.md index fbab888..03fd25f 100644 --- a/TODO.md +++ b/TODO.md @@ -24,12 +24,13 @@ https://git.eeqj.de/sneak/routewatch/pulls/6. After that, setting routewatch up 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). -Nothing on the record orders -https://git.eeqj.de/sneak/routewatch/issues/39 and -https://git.eeqj.de/sneak/routewatch/issues/30. +The other open issue is https://git.eeqj.de/sneak/routewatch/issues/30. # Completed Steps +- 2026-09-29: `.dockerignore` keeps `.git`, local build output, local + databases and `.env` out of the Docker build context, and so out of the + source archive in the image (closes #39) - 2026-09-29: README first line names the MIT license and the author; the License section now says MIT and links `LICENSE`, and an Author section was added; this file brought up to date (closes #38) @@ -102,4 +103,3 @@ https://git.eeqj.de/sneak/routewatch/issues/30. - `/api/v1/stats` answered HTTP 500 after 35 hours on the live feed, seen on `3898daa`, which predates the 2026-09-22 in-memory statistics (https://git.eeqj.de/sneak/routewatch/issues/30) -- Add a `.dockerignore` (https://git.eeqj.de/sneak/routewatch/issues/39) -- 2.54.0 From 6422d9fa0ce75b7122adfa6f87ac7d76193e1505 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Tue, 29 Sep 2026 12:23:42 +0200 Subject: [PATCH 19/19] Container makes its data directory usable itself (closes #42) sneak's standing rule: the container makes its data directory usable itself, with no step on the host. entrypoint.sh now creates /var/lib/berlin.sneak.app.routewatch if it is missing and stops the start when any step fails (set -euo pipefail); before, a failed cd went on to change the ownership of whatever directory the script was in, and a failed chown still started the daemon. Taking ownership of the directory and switching to the routewatch user through setpriv are unchanged. The README's upaas volume line now says only which path to mount. The empty-directory and other-uid cases were run by hand on the built image with upaas-style bind mounts, not added as an automated test. Model: opus-5-5 --- README.md | 4 +--- TODO.md | 3 +++ entrypoint.sh | 4 ++++ 3 files changed, 8 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 6204266..cf77465 100644 --- a/README.md +++ b/README.md @@ -235,9 +235,7 @@ with the goroutine count and Go memory figures. What the [upaas](https://git.eeqj.de/sneak/upaas) app needs: - Container port: `8080`. -- Volume: one, at container path `/var/lib/berlin.sneak.app.routewatch`. upaas - does not create the host directory, so create it before the first deploy. The - entrypoint takes ownership of it, so a root-owned directory works. +- Volume: one, at container path `/var/lib/berlin.sneak.app.routewatch`. - Environment: nothing is required. Leave `XDG_DATA_HOME`, `GOMEMLIMIT` and `MALLOC_ARENA_MAX` at the image's values. `DEBUG=routewatch` is optional and adds the `System stats` memory line to the log. diff --git a/TODO.md b/TODO.md index 03fd25f..b967d9a 100644 --- a/TODO.md +++ b/TODO.md @@ -28,6 +28,9 @@ The other open issue is https://git.eeqj.de/sneak/routewatch/issues/30. # Completed Steps +- 2026-09-29: the entrypoint creates the data directory if it is missing and + stops the start if a step fails; README "Running under upaas" no longer + asks for the host directory to be created first (closes #42) - 2026-09-29: `.dockerignore` keeps `.git`, local build output, local databases and `.env` out of the Docker build context, and so out of the source archive in the image (closes #39) diff --git a/entrypoint.sh b/entrypoint.sh index 7679fad..50d1aed 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -1,4 +1,5 @@ #!/bin/bash +set -euo pipefail # glibc silently ignores a malformed MALLOC_ARENA_MAX, so refuse it here. if [[ -n "${MALLOC_ARENA_MAX:-}" && ! "$MALLOC_ARENA_MAX" =~ ^[1-9][0-9]*$ ]]; then @@ -6,6 +7,9 @@ if [[ -n "${MALLOC_ARENA_MAX:-}" && ! "$MALLOC_ARENA_MAX" =~ ^[1-9][0-9]*$ ]]; t exit 1 fi +# Give the data directory to the routewatch user before the daemon starts, +# whether it is missing, an empty root-owned mount, or holds another uid's files. +mkdir -p /var/lib/berlin.sneak.app.routewatch cd /var/lib/berlin.sneak.app.routewatch chown -R routewatch:routewatch . chmod 700 . -- 2.54.0