.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

.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

.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

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
@@ -53,13 +78,26 @@ 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. 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
+# 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
 
 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" ]

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Jeffrey Paul <sneak@sneak.berlin>
+
+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.

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
@@ -165,13 +168,83 @@ 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 |
+|----------|----------|-------------|
+| `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, 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
+
+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.
+- 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:
+
+```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. `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
+  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.
+
+## 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`.
+- 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
 
@@ -204,7 +277,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 +293,15 @@ 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.
+MIT. See [`LICENSE`](LICENSE).
+
+## Author
+
+[@sneak](https://sneak.berlin)

TODO.md

@@ -10,19 +10,78 @@
 
 # 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).
+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)
+- 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
+  `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
+  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
+  #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
+  #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
@@ -41,8 +100,9 @@ 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)

entrypoint.sh

@@ -1,7 +1,20 @@
 #!/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
+    echo "MALLOC_ARENA_MAX must be a positive whole number, got '$MALLOC_ARENA_MAX'" >&2
+    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 .
 
-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

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

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)
+	}
+}

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
+}

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)
+	}
+}

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
@@ -56,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.
@@ -71,11 +88,17 @@ 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. _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",
+		"file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d&_journal_mode=WAL&_txlock=immediate",
 		dbPath,
+		sqliteCacheSizeKiB,
+		sqliteBusyTimeoutMs,
 	)
 	db, err := sql.Open("sqlite3", dsn)
 	if err != nil {
@@ -93,26 +116,35 @@ 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
 }
 
 // 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 {
@@ -184,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 {
@@ -203,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)
-			}
-
-			_, 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,
-			)
-		}
+			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++
+			}
+
+			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
 }
 
@@ -338,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
 
@@ -358,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
 }
 
@@ -432,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)
 
@@ -458,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)
 		}
@@ -467,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
 }
 
@@ -510,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
@@ -544,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
 		}
@@ -557,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
 }
 
@@ -628,6 +737,8 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error)
 		return nil, err
 	}
 
+	d.counts.addASNs(1)
+
 	return &asn, nil
 }
 
@@ -703,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
 }
 
@@ -779,6 +896,10 @@ func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error {
 		return err
 	}
 
+	if !exists {
+		d.counts.addPeerings(1)
+	}
+
 	return nil
 }
 
@@ -822,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)
@@ -839,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
 }
 
@@ -894,6 +1021,10 @@ func (d *Database) UpdatePeer(peerIP string, peerASN int, messageType string, ti
 		return err
 	}
 
+	if !exists {
+		d.counts.addPeers(1)
+	}
+
 	return nil
 }
 
@@ -902,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)
 	}
 
+	updateSQL, insertSQL := updateLiveRouteV4SQL, insertLiveRouteV4SQL
+	if route.IPVersion == ipVersionV6 {
+		updateSQL, insertSQL = updateLiveRouteV6SQL, insertLiveRouteV6SQL
+	}
+
+	// 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 {
-		// 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,
-		)
+		inserted, err = upsertRouteRowV4(upd, ins, route, string(pathJSON))
 	} else {
-		_, err = d.db.Exec(query,
-			route.ID.String(),
-			route.Prefix,
-			route.MaskLength,
-			route.OriginASN,
-			route.PeerIP,
-			string(pathJSON),
-			route.NextHop,
-			route.LastUpdated,
-		)
+		inserted, err = upsertRouteRowV6(upd, ins, route, string(pathJSON))
+	}
+	if err != nil {
+		return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
 	}
 
-	return 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
@@ -1079,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
 
-	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)
-	} 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)
+	// 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
+	}
+
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if isV4 {
+		d.counts.addRoutes(-int(affected), 0)
+	} else {
+		d.counts.addRoutes(0, -int(affected))
+	}
+
+	return nil
 }
 
 // GetPrefixDistribution returns the distribution of unique prefixes by mask length

internal/database/database_test.go

@@ -1,8 +1,35 @@
 package database
 
 import (
+	"context"
+	"database/sql"
 	"net"
+	"sync"
 	"testing"
+	"time"
+
+	"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
+
+// 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) {
@@ -282,6 +309,122 @@ 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)
+		}
+	}
+}
+
+// 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()

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

internal/ristypes/ris.go

@@ -74,13 +74,17 @@ type RISMessage struct {
 	MrtTime         float64   `json:"mrt_time,omitempty"`
 	SocketTime      float64   `json:"socket_time,omitempty"`
 	Path            ASPath    `json:"path,omitempty"`
-	Community       [][]int           `json:"community,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:"raw,omitempty"`
+	Raw           string            `json:"-"`
 }
 
 // RISAnnouncement represents a BGP route announcement within a RIS message.

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)
+	}
+}

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()
@@ -447,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{

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

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

internal/routewatch/peeringhandler.go

@@ -11,29 +11,25 @@ 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
 	// 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
+	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() {

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)
+	}
+}

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

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)
@@ -217,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()
 
@@ -257,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,
@@ -398,9 +393,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)
@@ -435,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()
 
@@ -533,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,

internal/server/handlers_test.go

@@ -0,0 +1,100 @@
+package server
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"runtime"
+	"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"
+	"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(), &config.Config{})
+
+	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
+}

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() {

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{
@@ -208,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)
@@ -463,7 +470,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 +530,7 @@ func (s *Streamer) stream(ctx context.Context) error {
 			select {
 			case <-metricsTicker.C:
 				s.logMetrics()
-			case <-ctx.Done():
+			case <-connCtx.Done():
 				return
 			}
 		}
@@ -536,7 +550,7 @@ func (s *Streamer) stream(ctx context.Context) error {
 					s.metrics.RecordWireBytes(delta)
 					lastWireBytes = currentBytes
 				}
-			case <-ctx.Done():
+			case <-connCtx.Done():
 				return
 			}
 		}
@@ -651,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

internal/streamer/streamer_test.go

@@ -1,10 +1,18 @@
 package streamer
 
 import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"runtime"
 	"testing"
+	"time"
 
 	"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) {
@@ -32,3 +40,107 @@ 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())
+	}
+}
+
+// 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 {
+	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
+}

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
     }
 }