Serve the /api/v1/stats prefix distribution from memory (closes #30)

The prefix distribution was the last query on the stats path. It read one
index entry per live route on every request, so on a large database it took
the whole 4-second deadline and /api/v1/stats answered 500.

The distribution now lives next to the in-memory counts: seeded once at
startup from the same query, then kept exact by the live-route writes. A new
route whose prefix had no live route adds one at its mask length; a delete
that leaves a prefix with no live route takes one away. Each check is one
lookup on the prefix index, made only for new and removed routes, inside the
write.

Model: opus-5-5

.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

Dockerfile

@@ -79,8 +79,8 @@ RUN chown -R routewatch:routewatch /app
 ENV XDG_DATA_HOME=/var/lib
 
 # Cap the Go heap at 1.5 GiB so the runtime collects harder before the
-# container's memory limit is reached. runuser preserves this the way it does
-# XDG_DATA_HOME above.
+# container's memory limit is reached. setpriv in the entrypoint preserves this
+# the way it does XDG_DATA_HOME above.
 ENV GOMEMLIMIT=1536MiB
 
 # Cap glibc's malloc arenas. The SQLite C library allocates and frees millions
@@ -96,8 +96,8 @@ EXPOSE 8080
 
 COPY ./entrypoint.sh /entrypoint.sh
 
-# Health check using the health endpoint
+# Health check using the health endpoint, on the port PORT names
 HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
-    CMD curl -sf http://localhost:8080/.well-known/healthcheck.json || exit 1
+    CMD curl -sf "http://localhost:${PORT:-8080}/.well-known/healthcheck.json" || exit 1
 
 ENTRYPOINT ["/bin/bash", "/entrypoint.sh" ]

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
@@ -166,14 +169,21 @@ Configuration is handled via environment variables and OS-specific paths:
 
 | Variable | Default | Description |
 |----------|----------|-------------|
-| `PORT` | `8080` | HTTP server port |
+| `PORT` | `8080` | HTTP server port, a whole number from 1 to 65535 |
 | `DEBUG` | (empty) | Set to `routewatch` for debug logging |
+| `XDG_DATA_HOME` | `/var/lib` (in the Docker image) | Base of the state directory; must be an absolute path |
 | `GOMEMLIMIT` | `1536MiB` (in the Docker image) | Go soft memory limit; see Memory |
-| `MALLOC_ARENA_MAX` | `2` (in the Docker image) | glibc malloc arena cap; see Memory |
+| `MALLOC_ARENA_MAX` | `2` (in the Docker image) | glibc malloc arena cap, a positive whole number; see Memory |
+
+A variable that is set to an invalid value stops the start with an error and a
+non-zero exit. An empty variable counts as unset.
 
 State directory (database location):
 - macOS: `~/Library/Application Support/routewatch/`
-- Linux: `/var/lib/routewatch/` or `~/.local/share/routewatch/`
+- Linux: `/var/lib/berlin.sneak.app.routewatch/` when running as root,
+  otherwise `$XDG_DATA_HOME/berlin.sneak.app.routewatch/` (with `XDG_DATA_HOME`
+  unset, `~/.local/share/berlin.sneak.app.routewatch/`). In the Docker image
+  this is `/var/lib/berlin.sneak.app.routewatch/`.
 
 ## Memory
 
@@ -220,6 +230,22 @@ What happens at each limit:
 With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
 with the goroutine count and Go memory figures.
 
+## Running under upaas
+
+What the [upaas](https://git.eeqj.de/sneak/upaas) app needs:
+
+- Container port: `8080`.
+- Volume: one, at container path `/var/lib/berlin.sneak.app.routewatch`.
+- Environment: nothing is required. Leave `XDG_DATA_HOME`, `GOMEMLIMIT` and
+  `MALLOC_ARENA_MAX` at the image's values. `DEBUG=routewatch` is optional and
+  adds the `System stats` memory line to the log.
+- Memory Limit: `5g`, the 5 GiB limit from Memory above. upaas sets no swap
+  limit, so on a host with swap Docker allows the same amount of swap again.
+- Health check: the image's `HEALTHCHECK` requests
+  `/.well-known/healthcheck.json` on the container port. upaas reads the
+  container's health 60 seconds after a deploy and fails the deploy unless it
+  is `healthy`.
+
 ## Development
 
 ```bash
@@ -274,4 +300,8 @@ invoke `go test` directly without `-short`:
 
 ## License
 
-See LICENSE file.
+MIT. See [`LICENSE`](LICENSE).
+
+## Author
+
+[@sneak](https://sneak.berlin)

TODO.md

@@ -10,19 +10,80 @@
 
 # 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).
 
 # Completed Steps
 
+- 2026-09-29: `/api/v1/stats` serves the prefix distribution from memory,
+  seeded at startup and adjusted on every live-route write, so a request no
+  longer reads every live route (closes #30)
+- 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 +102,6 @@ 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)

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,203 @@
+package database
+
+import (
+	"context"
+	"fmt"
+	"sync"
+)
+
+// liveCounts holds the running row counts and the prefix distribution 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 query over the tables. The COUNT(*) scans (issue 27) and then
+// the prefix distribution query (issue 30) each grew with the database until
+// they took the whole request timeout and made /api/v1/stats return 500.
+//
+// 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
+	// The prefix distribution: for each mask length, the number of distinct
+	// prefixes that have at least one live route.
+	distributionV4 [ipv4Bits + 1]int
+	distributionV6 [ipv6Bits + 1]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,
+	distributionV4, distributionV6 []PrefixDistribution) {
+	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
+	for _, entry := range distributionV4 {
+		addAtMaskLength(c.distributionV4[:], entry.MaskLength, entry.Count)
+	}
+	for _, entry := range distributionV6 {
+		addAtMaskLength(c.distributionV6[:], entry.MaskLength, entry.Count)
+	}
+}
+
+// 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()
+}
+
+// addToDistribution adds n to the IPv4 and IPv6 prefix distributions once for
+// each listed mask length. A write lists the mask lengths of the prefixes it
+// gave their first live route with n = 1, and of the prefixes it left with no
+// live route with n = -1.
+func (c *liveCounts) addToDistribution(maskLengthsV4, maskLengthsV6 []int, n int) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	for _, maskLength := range maskLengthsV4 {
+		addAtMaskLength(c.distributionV4[:], maskLength, n)
+	}
+	for _, maskLength := range maskLengthsV6 {
+		addAtMaskLength(c.distributionV6[:], maskLength, n)
+	}
+}
+
+// addAtMaskLength adds n to counts[maskLength]. A mask length the array has no
+// entry for is ignored, so a malformed route cannot crash the daemon.
+func addAtMaskLength(counts []int, maskLength, n int) {
+	if maskLength >= 0 && maskLength < len(counts) {
+		counts[maskLength] += n
+	}
+}
+
+// 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
+	s.IPv4PrefixDistribution = distributionList(c.distributionV4[:])
+	s.IPv6PrefixDistribution = distributionList(c.distributionV6[:])
+}
+
+// distributionList lists the mask lengths that have at least one prefix, in
+// ascending order, the way the distribution query returns them.
+func distributionList(counts []int) []PrefixDistribution {
+	var list []PrefixDistribution
+	for maskLength, count := range counts {
+		if count > 0 {
+			list = append(list, PrefixDistribution{MaskLength: maskLength, Count: count})
+		}
+	}
+
+	return list
+}
+
+// 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 and prefix distribution 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
+	}
+	distributionV4, distributionV6, err := d.GetPrefixDistributionContext(ctx)
+	if err != nil {
+		return err
+	}
+
+	d.counts.seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6,
+		distributionV4, distributionV6)
+
+	return nil
+}

internal/database/counts_test.go

@@ -0,0 +1,491 @@
+package database
+
+import (
+	"context"
+	"slices"
+	"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 mask length and range columns
+// taken from the prefix.
+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)
+	}
+	maskLength, err := prefixMaskLength(prefix)
+	if err != nil {
+		t.Fatalf("prefixMaskLength(%s): %v", prefix, err)
+	}
+
+	return &LiveRoute{
+		ID:          uuid.New(),
+		Prefix:      prefix,
+		MaskLength:  maskLength,
+		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 and the prefix distribution 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)
+	}
+	assertDistribution(t, "seeded IPv4 distribution", stats.IPv4PrefixDistribution,
+		[]PrefixDistribution{{MaskLength: 24, Count: 1}})
+	assertDistribution(t, "seeded IPv6 distribution", stats.IPv6PrefixDistribution,
+		[]PrefixDistribution{{MaskLength: 32, Count: 1}})
+}
+
+// TestPrefixDistributionTracksWrites checks that the prefix distribution the
+// stats read reports stays exact across each kind of write, through both the
+// batch and the single-route methods, and that after every step it equals what
+// the distribution query reads from the route tables.
+func TestPrefixDistributionTracksWrites(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)
+	shared := mkV4Route(t, "198.51.100.0/24", 64500, ts)
+	sharedSecondPeer := mkV4Route(t, "198.51.100.0/24", 64500, ts)
+	sharedSecondPeer.PeerIP = "192.0.2.2"
+	other := mkV4Route(t, "203.0.113.0/24", 64501, ts)
+	wide := mkV4Route(t, "172.16.0.0/16", 64502, ts)
+	v6 := mkV6Route("2001:db8::/32", 64503, ts)
+
+	all := []PrefixDistribution{{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 2}}
+	v6Only := []PrefixDistribution{{MaskLength: 32, Count: 1}}
+
+	steps := []struct {
+		name   string
+		write  func() error
+		wantV4 []PrefixDistribution
+		wantV6 []PrefixDistribution
+	}{
+		{
+			name:  "empty database",
+			write: func() error { return nil },
+		},
+		{
+			name:   "new prefixes",
+			write:  func() error { return db.UpsertLiveRouteBatch([]*LiveRoute{shared, other, wide, v6}) },
+			wantV4: all, wantV6: v6Only,
+		},
+		{
+			name:   "re-announcement",
+			write:  func() error { return db.UpsertLiveRouteBatch([]*LiveRoute{shared, other, wide, v6}) },
+			wantV4: all, wantV6: v6Only,
+		},
+		{
+			name:   "second peer announces a prefix that already has a route",
+			write:  func() error { return db.UpsertLiveRoute(sharedSecondPeer) },
+			wantV4: all, wantV6: v6Only,
+		},
+		{
+			name: "withdrawal of a route that is not the last for its prefix",
+			write: func() error {
+				return db.DeleteLiveRouteBatch([]LiveRouteDeletion{
+					{Prefix: shared.Prefix, OriginASN: shared.OriginASN, PeerIP: shared.PeerIP, IPVersion: ipVersionV4},
+				})
+			},
+			wantV4: all, wantV6: v6Only,
+		},
+		{
+			name: "withdrawal of the last route for a prefix",
+			write: func() error {
+				return db.DeleteLiveRoute(sharedSecondPeer.Prefix, sharedSecondPeer.OriginASN, sharedSecondPeer.PeerIP)
+			},
+			wantV4: []PrefixDistribution{{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 1}},
+			wantV6: v6Only,
+		},
+		{
+			name: "withdrawal of every remaining route, one without an origin ASN",
+			write: func() error {
+				return db.DeleteLiveRouteBatch([]LiveRouteDeletion{
+					{Prefix: other.Prefix, PeerIP: other.PeerIP, IPVersion: ipVersionV4},
+					{Prefix: wide.Prefix, OriginASN: wide.OriginASN, PeerIP: wide.PeerIP, IPVersion: ipVersionV4},
+					{Prefix: v6.Prefix, OriginASN: v6.OriginASN, PeerIP: v6.PeerIP, IPVersion: ipVersionV6},
+				})
+			},
+		},
+	}
+
+	ctx := context.Background()
+	for _, step := range steps {
+		if err := step.write(); err != nil {
+			t.Fatalf("%s: %v", step.name, err)
+		}
+
+		stats, err := db.GetStatsContext(ctx)
+		if err != nil {
+			t.Fatalf("%s: GetStatsContext: %v", step.name, err)
+		}
+		queryV4, queryV6, err := db.GetPrefixDistributionContext(ctx)
+		if err != nil {
+			t.Fatalf("%s: GetPrefixDistributionContext: %v", step.name, err)
+		}
+
+		assertDistribution(t, step.name+": IPv4 distribution", stats.IPv4PrefixDistribution, step.wantV4)
+		assertDistribution(t, step.name+": IPv6 distribution", stats.IPv6PrefixDistribution, step.wantV6)
+		assertDistribution(t, step.name+": IPv4 distribution query", queryV4, step.wantV4)
+		assertDistribution(t, step.name+": IPv6 distribution query", queryV6, step.wantV6)
+	}
+}
+
+// TestStatsDistributionNeedsNoQuery checks that the stats read serves the prefix
+// distribution from memory. The read's deadline has already passed, so any
+// query it ran would fail at once. The distribution query it used to run read
+// every live route and passed the /api/v1/stats deadline on a large database
+// (issue 30).
+func TestStatsDistributionNeedsNoQuery(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)
+	if err := db.UpsertLiveRouteBatch([]*LiveRoute{
+		mkV4Route(t, "198.51.100.0/24", 64500, ts),
+		mkV6Route("2001:db8::/32", 64501, ts),
+	}); err != nil {
+		t.Fatalf("UpsertLiveRouteBatch: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	stats, err := db.GetStatsContext(ctx)
+	if err != nil {
+		t.Fatalf("GetStatsContext: %v", err)
+	}
+	assertDistribution(t, "IPv4 distribution", stats.IPv4PrefixDistribution,
+		[]PrefixDistribution{{MaskLength: 24, Count: 1}})
+	assertDistribution(t, "IPv6 distribution", stats.IPv6PrefixDistribution,
+		[]PrefixDistribution{{MaskLength: 32, Count: 1}})
+}
+
+// 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)
+	}
+	// Every writer announced the same prefix, so it counts once.
+	assertDistribution(t, "IPv6 distribution", stats.IPv6PrefixDistribution,
+		[]PrefixDistribution{{MaskLength: 32, Count: 1}})
+}
+
+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)
+	}
+}
+
+func assertDistribution(t *testing.T, what string, got, want []PrefixDistribution) {
+	t.Helper()
+
+	if !slices.Equal(got, want) {
+		t.Errorf("%s = %v, want %v", what, got, want)
+	}
+}

internal/database/database.go

@@ -35,6 +35,7 @@ const (
 	ipv6Length     = 16
 	ipv4Offset     = 12
 	ipv4Bits       = 32
+	ipv6Bits       = 128
 	maxIPv4        = 0xFFFFFFFF
 )
 
@@ -72,6 +73,7 @@ type Database struct {
 	mu       sync.Mutex
 	lockedAt time.Time
 	lockedBy string
+	counts   *liveCounts
 }
 
 // New creates a new database connection and initializes the schema.
@@ -115,12 +117,19 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 	db.SetMaxIdleConns(maxConns)
 	db.SetConnMaxLifetime(0)
 
-	database := &Database{db: db, logger: logger, path: dbPath}
+	database := &Database{db: db, logger: logger, path: dbPath, counts: &liveCounts{}}
 
 	if err := database.Initialize(); err != nil {
 		return nil, fmt.Errorf("failed to initialize database: %w", err)
 	}
 
+	// Seed the in-memory statistics counters from the tables once, before the
+	// streamer starts writing. From here on every write keeps them current, so
+	// the stats endpoints never scan the tables to report counts.
+	if err := database.seedCounts(context.Background()); err != nil {
+		return nil, fmt.Errorf("failed to seed statistics counters: %w", err)
+	}
+
 	return database, nil
 }
 
@@ -208,6 +217,103 @@ 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 (?, ?, ?, ?, ?, ?, ?, ?)`
+)
+
+// Before a new route is inserted and after a route is deleted, the write looks
+// up whether any live route has that prefix, to keep the in-memory prefix
+// distribution exact. The lookup reads one entry of the prefix index.
+const (
+	prefixHasLiveRouteV4SQL = `SELECT EXISTS (SELECT 1 FROM live_routes_v4 WHERE prefix = ?)`
+	prefixHasLiveRouteV6SQL = `SELECT EXISTS (SELECT 1 FROM live_routes_v6 WHERE prefix = ?)`
+)
+
+// upsertRouteRowV4 updates an IPv4 live route, inserting it when no row matched.
+// It reports whether a new row was inserted and whether that row is the first
+// live route for its prefix.
+func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
+	inserted, newPrefix bool, err error) {
+	if route.V4IPStart == nil || route.V4IPEnd == nil {
+		return false, 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, false, err
+	}
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return false, false, err
+	}
+	if affected > 0 {
+		return false, false, nil
+	}
+
+	var prefixHadRoute bool
+	if err := has.QueryRow(route.Prefix).Scan(&prefixHadRoute); err != nil {
+		return false, false, err
+	}
+
+	_, 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, false, err
+	}
+
+	return true, !prefixHadRoute, nil
+}
+
+// upsertRouteRowV6 updates an IPv6 live route, inserting it when no row matched.
+// It reports whether a new row was inserted and whether that row is the first
+// live route for its prefix.
+func upsertRouteRowV6(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
+	inserted, newPrefix 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, false, err
+	}
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return false, false, err
+	}
+	if affected > 0 {
+		return false, false, nil
+	}
+
+	var prefixHadRoute bool
+	if err := has.QueryRow(route.Prefix).Scan(&prefixHadRoute); err != nil {
+		return false, false, err
+	}
+
+	_, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN,
+		route.PeerIP, pathJSON, route.NextHop, route.LastUpdated)
+	if err != nil {
+		return false, false, err
+	}
+
+	return true, !prefixHadRoute, nil
+}
+
 // UpsertLiveRouteBatch inserts or updates multiple live routes in a single transaction
 func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
 	if len(routes) == 0 {
@@ -227,91 +333,86 @@ 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() }()
+
+	hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare IPv4 prefix lookup statement: %w", err)
+	}
+	defer func() { _ = hasV4.Close() }()
+
+	hasV6, err := tx.Prepare(prefixHasLiveRouteV6SQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare IPv6 prefix lookup statement: %w", err)
+	}
+	defer func() { _ = hasV6.Close() }()
+
+	var newV4, newV6 int
+	// Mask lengths of the prefixes that get their first live route in this batch.
+	var newPrefixMaskLengthsV4, newPrefixMaskLengthsV6 []int
 	for _, route := range routes {
-		// Encode AS path as JSON
 		pathJSON, err := json.Marshal(route.ASPath)
 		if err != nil {
 			return fmt.Errorf("failed to encode AS path: %w", err)
 		}
 
-		// Use appropriate statement based on IP version
 		if route.IPVersion == ipVersionV4 {
-			// IPv4 routes must have range values
-			if route.V4IPStart == nil || route.V4IPEnd == nil {
-				return fmt.Errorf("IPv4 route %s missing range values", route.Prefix)
+			inserted, newPrefix, err := upsertRouteRowV4(updV4, insV4, hasV4, route, string(pathJSON))
+			if err != nil {
+				return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
+			}
+			if inserted {
+				newV4++
+			}
+			if newPrefix {
+				newPrefixMaskLengthsV4 = append(newPrefixMaskLengthsV4, route.MaskLength)
 			}
 
-			_, err = stmtV4.Exec(
-				route.ID.String(),
-				route.Prefix,
-				route.MaskLength,
-				route.OriginASN,
-				route.PeerIP,
-				string(pathJSON),
-				route.NextHop,
-				route.LastUpdated,
-				*route.V4IPStart,
-				*route.V4IPEnd,
-			)
-		} else {
-			// IPv6 routes
-			_, err = stmtV6.Exec(
-				route.ID.String(),
-				route.Prefix,
-				route.MaskLength,
-				route.OriginASN,
-				route.PeerIP,
-				string(pathJSON),
-				route.NextHop,
-				route.LastUpdated,
-			)
+			continue
 		}
+
+		inserted, newPrefix, err := upsertRouteRowV6(updV6, insV6, hasV6, route, string(pathJSON))
 		if err != nil {
 			return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
 		}
+		if inserted {
+			newV6++
+		}
+		if newPrefix {
+			newPrefixMaskLengthsV6 = append(newPrefixMaskLengthsV6, route.MaskLength)
+		}
 	}
 
 	if err = tx.Commit(); err != nil {
 		return fmt.Errorf("failed to commit transaction: %w", err)
 	}
 
+	d.counts.addRoutes(newV4, newV6)
+	d.counts.addToDistribution(newPrefixMaskLengthsV4, newPrefixMaskLengthsV6, 1)
+
 	return nil
 }
 
@@ -361,7 +462,22 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 	}
 	defer func() { _ = stmtV6WithoutOrigin.Close() }()
 
+	hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare IPv4 prefix lookup statement: %w", err)
+	}
+	defer func() { _ = hasV4.Close() }()
+
+	hasV6, err := tx.Prepare(prefixHasLiveRouteV6SQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare IPv6 prefix lookup statement: %w", err)
+	}
+	defer func() { _ = hasV6.Close() }()
+
 	// Process deletions
+	var deletedV4, deletedV6 int64
+	// Mask lengths of the prefixes this batch leaves with no live route.
+	var gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6 []int
 	for _, del := range deletions {
 		var stmt *sql.Stmt
 
@@ -382,21 +498,62 @@ 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 affected == 0 {
+			continue
+		}
+
+		// The prefix leaves the distribution when no live route has it any more.
+		has := hasV4
+		if del.IPVersion != ipVersionV4 {
+			has = hasV6
+		}
+		var prefixHasRoute bool
+		if err := has.QueryRow(del.Prefix).Scan(&prefixHasRoute); err != nil {
+			return fmt.Errorf("failed to look up prefix %s: %w", del.Prefix, err)
+		}
+		if prefixHasRoute {
+			continue
+		}
+		maskLength, err := prefixMaskLength(del.Prefix)
+		if err != nil {
+			return err
+		}
+		if del.IPVersion == ipVersionV4 {
+			gonePrefixMaskLengthsV4 = append(gonePrefixMaskLengthsV4, maskLength)
+		} else {
+			gonePrefixMaskLengthsV6 = append(gonePrefixMaskLengthsV6, maskLength)
+		}
 	}
 
 	if err = tx.Commit(); err != nil {
 		return fmt.Errorf("failed to commit transaction: %w", err)
 	}
 
+	d.counts.addRoutes(-int(deletedV4), -int(deletedV6))
+	d.counts.addToDistribution(gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6, -1)
+
 	return nil
 }
 
@@ -456,6 +613,7 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 	}
 	defer func() { _ = insertV6Stmt.Close() }()
 
+	var newV4, newV6 int
 	for prefix, timestamp := range prefixes {
 		ipVersion := detectIPVersion(prefix)
 
@@ -482,6 +640,11 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 			if err != nil {
 				return fmt.Errorf("failed to insert prefix %s: %w", prefix, err)
 			}
+			if ipVersion == ipVersionV4 {
+				newV4++
+			} else {
+				newV6++
+			}
 		default:
 			return fmt.Errorf("failed to query prefix %s: %w", prefix, err)
 		}
@@ -491,6 +654,8 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 		return fmt.Errorf("failed to commit transaction: %w", err)
 	}
 
+	d.counts.addPrefixes(newV4, newV6)
+
 	return nil
 }
 
@@ -534,6 +699,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
 	}
 	defer func() { _ = insertStmt.Close() }()
 
+	var newASNs int
 	for number, timestamp := range asns {
 		var asn ASN
 		var handle, description sql.NullString
@@ -568,6 +734,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
 			if err != nil {
 				return fmt.Errorf("failed to insert ASN %d: %w", number, err)
 			}
+			newASNs++
 
 			continue
 		}
@@ -581,6 +748,8 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
 		return fmt.Errorf("failed to commit transaction: %w", err)
 	}
 
+	d.counts.addASNs(newASNs)
+
 	return nil
 }
 
@@ -652,6 +821,8 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error)
 		return nil, err
 	}
 
+	d.counts.addASNs(1)
+
 	return &asn, nil
 }
 
@@ -727,6 +898,12 @@ func (d *Database) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefi
 		return nil, err
 	}
 
+	if ipVersion == ipVersionV4 {
+		d.counts.addPrefixes(1, 0)
+	} else {
+		d.counts.addPrefixes(0, 1)
+	}
+
 	return &p, nil
 }
 
@@ -803,6 +980,10 @@ func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error {
 		return err
 	}
 
+	if !exists {
+		d.counts.addPeerings(1)
+	}
+
 	return nil
 }
 
@@ -846,6 +1027,7 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error {
 	}
 	defer func() { _ = insertStmt.Close() }()
 
+	var newPeers int
 	for _, update := range peers {
 		var exists bool
 		err = checkStmt.QueryRow(update.PeerIP).Scan(&exists)
@@ -863,12 +1045,17 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error {
 		if err != nil {
 			return fmt.Errorf("failed to update peer %s: %w", update.PeerIP, err)
 		}
+		if !exists {
+			newPeers++
+		}
 	}
 
 	if err = tx.Commit(); err != nil {
 		return fmt.Errorf("failed to commit transaction: %w", err)
 	}
 
+	d.counts.addPeers(newPeers)
+
 	return nil
 }
 
@@ -918,6 +1105,10 @@ func (d *Database) UpdatePeer(peerIP string, peerASN int, messageType string, ti
 		return err
 	}
 
+	if !exists {
+		d.counts.addPeers(1)
+	}
+
 	return nil
 }
 
@@ -926,81 +1117,39 @@ 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) and the prefix
+// distribution come from the in-memory counters, seeded at startup and kept
+// current on every write. The oldest/newest route timestamps are read from the
+// ends of the last_updated index, and the file size from a stat(). No part of
+// the read scans a table or a whole index.
 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 and prefix distribution 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.IPv4Routes = v4Count
-	stats.IPv6Routes = v6Count
-	stats.LiveRoutes = v4Count + v6Count
-
-	// Get oldest and newest route timestamps. Each query reads a single row from
-	// one end of the last_updated index, so the cost is a log-time index lookup
-	// rather than a full scan of both route tables. Selecting the last_updated
-	// column directly (rather than MIN/MAX, whose result has no column type) lets
-	// the driver parse the DATETIME value into time.Time; the union scan aggregate
-	// used before returned an untyped string and logged a warning on every call.
-	stats.OldestRoute, stats.NewestRoute, err = d.routeTimestampRange(ctx)
+	// 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)
-	}
-
-	// Get prefix distribution
-	stats.IPv4PrefixDistribution, stats.IPv6PrefixDistribution, err = d.GetPrefixDistributionContext(ctx)
-	if err != nil {
-		// Log but don't fail
-		d.logger.Warn("Failed to get prefix distribution", "error", err)
+	} else {
+		stats.OldestRoute = oldest
+		stats.NewestRoute = newest
 	}
 
 	return stats, nil
@@ -1069,81 +1218,62 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
 	d.lock("UpsertLiveRoute")
 	defer d.unlock()
 
-	// Choose table based on IP version
-	tableName := "live_routes_v4"
-	if route.IPVersion == ipVersionV6 {
-		tableName = "live_routes_v6"
-	}
-
-	var query string
-	if route.IPVersion == ipVersionV4 {
-		query = fmt.Sprintf(`
-			INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, 
-				last_updated, ip_start, ip_end)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-			ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
-				mask_length = excluded.mask_length,
-				as_path = excluded.as_path,
-				next_hop = excluded.next_hop,
-				last_updated = excluded.last_updated,
-				ip_start = excluded.ip_start,
-				ip_end = excluded.ip_end
-		`, tableName)
-	} else {
-		query = fmt.Sprintf(`
-			INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop, 
-				last_updated)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-			ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
-				mask_length = excluded.mask_length,
-				as_path = excluded.as_path,
-				next_hop = excluded.next_hop,
-				last_updated = excluded.last_updated
-		`, tableName)
-	}
-
-	// Encode AS path as JSON
 	pathJSON, err := json.Marshal(route.ASPath)
 	if err != nil {
 		return fmt.Errorf("failed to encode AS path: %w", err)
 	}
 
-	if route.IPVersion == ipVersionV4 {
-		// Convert v4_ip_start and v4_ip_end to interface{} for SQL NULL handling
-		var v4Start, v4End interface{}
-		if route.V4IPStart != nil {
-			v4Start = *route.V4IPStart
-		}
-		if route.V4IPEnd != nil {
-			v4End = *route.V4IPEnd
-		}
-
-		_, err = d.db.Exec(query,
-			route.ID.String(),
-			route.Prefix,
-			route.MaskLength,
-			route.OriginASN,
-			route.PeerIP,
-			string(pathJSON),
-			route.NextHop,
-			route.LastUpdated,
-			v4Start,
-			v4End,
-		)
-	} else {
-		_, err = d.db.Exec(query,
-			route.ID.String(),
-			route.Prefix,
-			route.MaskLength,
-			route.OriginASN,
-			route.PeerIP,
-			string(pathJSON),
-			route.NextHop,
-			route.LastUpdated,
-		)
+	updateSQL, insertSQL, lookupSQL := updateLiveRouteV4SQL, insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
+	if route.IPVersion == ipVersionV6 {
+		updateSQL, insertSQL, lookupSQL = updateLiveRouteV6SQL, insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
 	}
 
-	return err
+	// The write lock is held, so no other writer can insert this key between the
+	// update and the insert even though they are separate autocommit statements.
+	upd, err := d.db.Prepare(updateSQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare update statement: %w", err)
+	}
+	defer func() { _ = upd.Close() }()
+
+	ins, err := d.db.Prepare(insertSQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare insert statement: %w", err)
+	}
+	defer func() { _ = ins.Close() }()
+
+	has, err := d.db.Prepare(lookupSQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare prefix lookup statement: %w", err)
+	}
+	defer func() { _ = has.Close() }()
+
+	var inserted, newPrefix bool
+	if route.IPVersion == ipVersionV4 {
+		inserted, newPrefix, err = upsertRouteRowV4(upd, ins, has, route, string(pathJSON))
+	} else {
+		inserted, newPrefix, err = upsertRouteRowV6(upd, ins, has, route, string(pathJSON))
+	}
+	if err != nil {
+		return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
+	}
+
+	if inserted {
+		if route.IPVersion == ipVersionV4 {
+			d.counts.addRoutes(1, 0)
+		} else {
+			d.counts.addRoutes(0, 1)
+		}
+	}
+	if newPrefix {
+		if route.IPVersion == ipVersionV4 {
+			d.counts.addToDistribution([]int{route.MaskLength}, nil, 1)
+		} else {
+			d.counts.addToDistribution(nil, []int{route.MaskLength}, 1)
+		}
+	}
+
+	return nil
 }
 
 // DeleteLiveRoute deletes a live route
@@ -1158,23 +1288,62 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string)
 		return fmt.Errorf("invalid prefix format: %w", err)
 	}
 
-	tableName := "live_routes_v4"
-	if ipnet.IP.To4() == nil {
-		tableName = "live_routes_v6"
+	isV4 := ipnet.IP.To4() != nil
+
+	// Literal per-table queries (rather than one formatted with the table name)
+	// so the delete carries no dynamically built SQL. A delete with no origin
+	// ASN can remove several rows.
+	var res sql.Result
+	switch {
+	case isV4 && originASN == 0:
+		res, err = d.db.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
+	case isV4:
+		res, err = d.db.Exec(
+			`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
+			prefix, originASN, peerIP)
+	case originASN == 0:
+		res, err = d.db.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
+	default:
+		res, err = d.db.Exec(
+			`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
+			prefix, originASN, peerIP)
+	}
+	if err != nil {
+		return err
 	}
 
-	var query string
-	if originASN == 0 {
-		// Delete all routes for this prefix from this peer
-		query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND peer_ip = ?`, tableName)
-		_, err = d.db.Exec(query, prefix, peerIP)
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if isV4 {
+		d.counts.addRoutes(-int(affected), 0)
 	} else {
-		// Delete specific route
-		query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, tableName)
-		_, err = d.db.Exec(query, prefix, originASN, peerIP)
+		d.counts.addRoutes(0, -int(affected))
+	}
+	if affected == 0 {
+		return nil
 	}
 
-	return err
+	// The prefix leaves the distribution when no live route has it any more.
+	lookupSQL := prefixHasLiveRouteV6SQL
+	if isV4 {
+		lookupSQL = prefixHasLiveRouteV4SQL
+	}
+	var prefixHasRoute bool
+	if err := d.db.QueryRow(lookupSQL, prefix).Scan(&prefixHasRoute); err != nil {
+		return err
+	}
+	if !prefixHasRoute {
+		maskLength, _ := ipnet.Mask.Size()
+		if isV4 {
+			d.counts.addToDistribution([]int{maskLength}, nil, -1)
+		} else {
+			d.counts.addToDistribution(nil, []int{maskLength}, -1)
+		}
+	}
+
+	return nil
 }
 
 // GetPrefixDistribution returns the distribution of unique prefixes by mask length
@@ -1182,7 +1351,9 @@ func (d *Database) GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []Pr
 	return d.GetPrefixDistributionContext(context.Background())
 }
 
-// GetPrefixDistributionContext returns the distribution of unique prefixes by mask length with context support
+// GetPrefixDistributionContext returns the distribution of unique prefixes by mask length with context support.
+// It reads every live route, so the stats read does not call it; it seeds the
+// in-memory distribution once at startup.
 func (d *Database) GetPrefixDistributionContext(ctx context.Context) (
 	ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) {
 	// IPv4 distribution - count unique prefixes from v4 table

internal/database/database_test.go

@@ -10,7 +10,6 @@ import (
 
 	"git.eeqj.de/sneak/routewatch/internal/config"
 	"git.eeqj.de/sneak/routewatch/internal/logger"
-	"github.com/google/uuid"
 )
 
 // tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in
@@ -426,105 +425,6 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) {
 	wg.Wait()
 }
 
-// TestStatsRouteTimestampsAndCounts checks GetStatsContext reports the correct
-// route counts and the oldest/newest last_updated across both route tables. The
-// old union-scan query read the aggregate result into *time.Time, which the
-// driver could not parse, so it logged a warning every call and left both
-// timestamps nil; this asserts they are populated from the right rows.
-func TestStatsRouteTimestampsAndCounts(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 database: no routes, so both timestamps are nil and no error.
-	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)
-	}
-	if empty.LiveRoutes != 0 {
-		t.Fatalf("empty database LiveRoutes = %d, want 0", empty.LiveRoutes)
-	}
-
-	base := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
-	oldest := base
-	middle := base.Add(time.Minute)
-	newest := base.Add(2 * time.Minute)
-
-	mkV4 := func(prefix string, asn int, ts time.Time) *LiveRoute {
-		start, end, rerr := CalculateIPv4Range(prefix)
-		if rerr != nil {
-			t.Fatalf("CalculateIPv4Range(%s): %v", prefix, rerr)
-		}
-
-		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,
-		}
-	}
-
-	// Two IPv4 routes (one oldest, one middle) and one IPv6 route (newest).
-	routes := []*LiveRoute{
-		mkV4("198.51.100.0/24", 64500, middle),
-		mkV4("203.0.113.0/24", 64501, oldest),
-		{
-			ID:          uuid.New(),
-			Prefix:      "2001:db8::/32",
-			MaskLength:  32,
-			IPVersion:   ipVersionV6,
-			OriginASN:   64502,
-			PeerIP:      "2001:db8::1",
-			ASPath:      []int{64502},
-			NextHop:     "2001:db8::ffff",
-			LastUpdated: newest,
-		},
-	}
-	for _, route := range routes {
-		if err := db.UpsertLiveRoute(route); err != nil {
-			t.Fatalf("UpsertLiveRoute(%s): %v", route.Prefix, err)
-		}
-	}
-
-	stats, err := db.GetStatsContext(ctx)
-	if err != nil {
-		t.Fatalf("GetStatsContext: %v", err)
-	}
-
-	if stats.IPv4Routes != 2 {
-		t.Errorf("IPv4Routes = %d, want 2", stats.IPv4Routes)
-	}
-	if stats.IPv6Routes != 1 {
-		t.Errorf("IPv6Routes = %d, want 1", stats.IPv6Routes)
-	}
-	if stats.LiveRoutes != 3 {
-		t.Errorf("LiveRoutes = %d, want 3", stats.LiveRoutes)
-	}
-	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)
-	}
-}
-
 func BenchmarkIPToUint32(b *testing.B) {
 	ip := net.ParseIP("192.168.1.1")
 	b.ResetTimer()

internal/database/utils.go

@@ -1,6 +1,8 @@
 package database
 
 import (
+	"fmt"
+	"net"
 	"strings"
 
 	"github.com/google/uuid"
@@ -18,3 +20,14 @@ func detectIPVersion(prefix string) int {
 
 	return ipVersionV4
 }
+
+// prefixMaskLength returns the mask length of a prefix such as 192.0.2.0/24.
+func prefixMaskLength(prefix string) (int, error) {
+	_, network, err := net.ParseCIDR(prefix)
+	if err != nil {
+		return 0, fmt.Errorf("invalid prefix %s: %w", prefix, err)
+	}
+	maskLength, _ := network.Mask.Size()
+
+	return maskLength, nil
+}

internal/routewatch/app_integration_test.go

@@ -450,7 +450,7 @@ func TestRouteWatchLiveFeed(t *testing.T) {
 	}
 
 	// Create server
-	srv := server.New(mockDB, s, logger)
+	srv := server.New(mockDB, s, logger, cfg)
 
 	// Create RouteWatch with 5 second limit
 	deps := Dependencies{

internal/server/handlers.go

@@ -179,14 +179,37 @@ func (s *Server) handleStatusJSON() http.HandlerFunc {
 
 		metrics := s.streamer.GetMetrics()
 
-		// Serve database statistics from the cache, which runs the table scans at
-		// most once per interval so this request does not.
-		dbStats, err := s.stats.get()
-		if err != nil {
+		// 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)
+			if err != nil {
+				s.logger.Debug("Database stats query failed", "error", err)
+				errChan <- err
+
+				return
+			}
+			statsChan <- dbStats
+		}()
+
+		var dbStats database.Stats
+		select {
+		case <-ctx.Done():
+			s.logger.Error("Database stats timeout in status.json")
+			writeJSONError(w, http.StatusRequestTimeout, "Database timeout")
+
+			return
+		case err := <-errChan:
 			s.logger.Error("Failed to get database stats", "error", err)
 			writeJSONError(w, http.StatusInternalServerError, err.Error())
 
 			return
+		case dbStats = <-statsChan:
+			// Success
 		}
 
 		uptime := time.Since(metrics.ConnectedSince).Truncate(time.Second).String()
@@ -370,14 +393,36 @@ func (s *Server) handleStats() http.HandlerFunc {
 
 		metrics := s.streamer.GetMetrics()
 
-		// Serve database statistics from the cache, which runs the table scans at
-		// most once per interval so this request does not.
-		dbStats, err := s.stats.get()
-		if err != nil {
+		// 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)
+			if err != nil {
+				s.logger.Debug("Database stats query failed", "error", err)
+				errChan <- err
+
+				return
+			}
+			statsChan <- dbStats
+		}()
+
+		var dbStats database.Stats
+		select {
+		case <-ctx.Done():
+			s.logger.Error("Database stats timeout")
+			// Don't write response here - timeout middleware already handles it
+			return
+		case err := <-errChan:
 			s.logger.Error("Failed to get database stats", "error", err)
 			writeJSONError(w, http.StatusInternalServerError, err.Error())
 
 			return
+		case dbStats = <-statsChan:
+			// Success
 		}
 
 		uptime := time.Since(metrics.ConnectedSince).Truncate(time.Second).String()

internal/server/handlers_test.go

@@ -4,55 +4,97 @@ import (
 	"context"
 	"net/http"
 	"net/http/httptest"
-	"sync/atomic"
+	"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"
 )
 
-// countingStatsDB embeds database.Store (left nil) and overrides only
-// GetStatsContext, counting how many times it runs. The stats handlers read
-// their database statistics through the cache, which calls this; every other
-// Store method is unused on the stats path and would panic if called.
-type countingStatsDB struct {
+// 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
-	calls *atomic.Int64
+	release chan struct{}
 }
 
-func (d countingStatsDB) GetStatsContext(_ context.Context) (database.Stats, error) {
-	d.calls.Add(1)
+func (d blockingStatsDB) GetStatsContext(_ context.Context) (database.Stats, error) {
+	<-d.release
 
 	return database.Stats{}, nil
 }
 
-// TestStatsHandlersServeFromCache drives both stats handlers many times and
-// checks that they answer 200 while the database statistics are computed at most
-// once within the refresh interval. Before the fix each request ran the counts
-// and MIN/MAX scans itself, which took the full timeout and returned 500 once
-// the database grew large.
-func TestStatsHandlersServeFromCache(t *testing.T) {
-	var calls atomic.Int64
-	db := countingStatsDB{calls: &calls}
-	s := New(db, streamer.New(logger.New(), metrics.New()), logger.New())
+// 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 := []http.HandlerFunc{s.handleStatusJSON(), s.handleStats()}
+	handlers := map[string]http.HandlerFunc{
+		"status.json": s.handleStatusJSON(),
+		"stats":       s.handleStats(),
+	}
 
-	const iterations = 20
+	baseline := settledGoroutineCount()
+
+	const (
+		iterations     = 20
+		requestTimeout = 50 * time.Millisecond
+	)
 	for _, handler := range handlers {
 		for range iterations {
-			req := httptest.NewRequest(http.MethodGet, "/", nil)
-			rec := httptest.NewRecorder()
-			handler(rec, req)
-			if rec.Code != http.StatusOK {
-				t.Fatalf("handler returned %d, want %d", rec.Code, http.StatusOK)
-			}
+			ctx, cancel := context.WithTimeout(context.Background(), requestTimeout)
+			req := httptest.NewRequest(http.MethodGet, "/", nil).WithContext(ctx)
+			handler(httptest.NewRecorder(), req)
+			cancel()
 		}
 	}
 
-	if got := calls.Load(); got != 1 {
-		t.Fatalf("GetStatsContext ran %d times, want 1 within the interval", got)
+	// 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,21 +34,19 @@ type Server struct {
 	db         database.Store
 	streamer   *streamer.Streamer
 	logger     *logger.Logger
+	port       int
 	srv        *http.Server
 	asnFetcher ASNFetcher
-	stats      *statsCache
 }
 
 // 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.stats = newStatsCache(func(ctx context.Context) (database.Stats, error) {
-		return s.db.GetStatsContext(ctx)
-	})
 
 	s.setupRoutes()
 
@@ -56,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
@@ -69,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,
@@ -77,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/server/statscache.go

@@ -1,112 +0,0 @@
-package server
-
-import (
-	"context"
-	"sync"
-	"time"
-
-	"git.eeqj.de/sneak/routewatch/internal/database"
-)
-
-const (
-	// statsRefreshInterval is how often the cached database statistics are
-	// recomputed. The scans behind GetStatsContext grow with the tables, so a
-	// request serves the cached copy instead of running them.
-	statsRefreshInterval = 30 * time.Second
-
-	// statsComputeTimeout bounds a single statistics computation so a stuck scan
-	// cannot block the refresh forever.
-	statsComputeTimeout = 20 * time.Second
-)
-
-// statsFetch computes fresh statistics. It is the expensive database scan that
-// the cache runs at most once per interval.
-type statsFetch func(ctx context.Context) (database.Stats, error)
-
-// statsCache serves the most recent database statistics and recomputes them at
-// most once per interval. The first request computes synchronously so it has
-// real data to return; afterwards requests serve the cached copy immediately
-// and a stale copy triggers a single background refresh, so no request waits on
-// the scans.
-type statsCache struct {
-	fetch    statsFetch
-	interval time.Duration
-	now      func() time.Time
-
-	mu         sync.Mutex
-	stats      database.Stats
-	haveStats  bool
-	fetchedAt  time.Time
-	refreshing bool
-}
-
-// newStatsCache returns a cache that recomputes statistics with fetch no more
-// than once per statsRefreshInterval.
-func newStatsCache(fetch statsFetch) *statsCache {
-	return &statsCache{
-		fetch:    fetch,
-		interval: statsRefreshInterval,
-		now:      time.Now,
-	}
-}
-
-// get returns the cached statistics. On the first call it computes them
-// synchronously and returns any error. Later calls return the cached copy, and
-// when that copy is older than the interval they start one background refresh.
-func (c *statsCache) get() (database.Stats, error) {
-	c.mu.Lock()
-
-	if !c.haveStats {
-		// Cold start: compute once under the lock so concurrent first callers
-		// wait for this single computation rather than each starting their own.
-		stats, err := c.compute()
-		if err != nil {
-			c.mu.Unlock()
-
-			return database.Stats{}, err
-		}
-		c.store(stats)
-		c.mu.Unlock()
-
-		return stats, nil
-	}
-
-	if c.now().Sub(c.fetchedAt) >= c.interval && !c.refreshing {
-		c.refreshing = true
-		go c.refresh()
-	}
-
-	stats := c.stats
-	c.mu.Unlock()
-
-	return stats, nil
-}
-
-// refresh recomputes the statistics in the background and replaces the cached
-// copy. A failed computation leaves the previous copy in place.
-func (c *statsCache) refresh() {
-	stats, err := c.compute()
-
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	c.refreshing = false
-	if err == nil {
-		c.store(stats)
-	}
-}
-
-// compute runs the fetch with its own bounded context, independent of any
-// request, so one request's cancellation cannot abort a shared refresh.
-func (c *statsCache) compute() (database.Stats, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), statsComputeTimeout)
-	defer cancel()
-
-	return c.fetch(ctx)
-}
-
-// store records a fresh result. The caller must hold the mutex.
-func (c *statsCache) store(stats database.Stats) {
-	c.stats = stats
-	c.haveStats = true
-	c.fetchedAt = c.now()
-}

internal/server/statscache_test.go

@@ -1,173 +0,0 @@
-package server
-
-import (
-	"context"
-	"errors"
-	"runtime"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"git.eeqj.de/sneak/routewatch/internal/database"
-)
-
-// testClock is a concurrency-safe clock the cache tests advance by hand, so the
-// interval boundary is exercised without waiting real time.
-type testClock struct {
-	ns atomic.Int64
-}
-
-func (c *testClock) now() time.Time          { return time.Unix(0, c.ns.Load()) }
-func (c *testClock) advance(d time.Duration) { c.ns.Add(int64(d)) }
-
-// waitForCalls waits until calls reaches want, giving a background refresh time
-// to finish.
-func waitForCalls(calls *atomic.Int64, want int64) bool {
-	const attempts = 200
-	for range attempts {
-		if calls.Load() >= want {
-			return true
-		}
-		time.Sleep(5 * time.Millisecond)
-	}
-
-	return false
-}
-
-// TestStatsCacheComputesOncePerInterval is the core guarantee: many reads in a
-// row run the expensive fetch at most once per interval, and crossing the
-// interval boundary allows exactly one more computation.
-func TestStatsCacheComputesOncePerInterval(t *testing.T) {
-	clk := &testClock{}
-	clk.ns.Store(int64(time.Hour)) // start at a non-zero instant
-
-	var calls atomic.Int64
-	c := newStatsCache(func(_ context.Context) (database.Stats, error) {
-		calls.Add(1)
-
-		return database.Stats{}, nil
-	})
-	c.now = clk.now
-
-	const reads = 50
-	for range reads {
-		if _, err := c.get(); err != nil {
-			t.Fatalf("get returned error: %v", err)
-		}
-	}
-	if got := calls.Load(); got != 1 {
-		t.Fatalf("fetch ran %d times within the interval, want 1", got)
-	}
-
-	// Cross the interval: the next read serves the stale copy and starts one
-	// background refresh.
-	clk.advance(c.interval)
-	if _, err := c.get(); err != nil {
-		t.Fatalf("get after interval returned error: %v", err)
-	}
-	if !waitForCalls(&calls, 2) {
-		t.Fatalf("background refresh did not run, fetch ran %d times", calls.Load())
-	}
-
-	for range reads {
-		if _, err := c.get(); err != nil {
-			t.Fatalf("get returned error: %v", err)
-		}
-	}
-	if got := calls.Load(); got != 2 {
-		t.Fatalf("fetch ran %d times across one interval boundary, want 2", got)
-	}
-}
-
-// TestStatsCacheColdStartReturnsError checks the first computation's error
-// reaches the caller, since there is no cached copy to serve instead.
-func TestStatsCacheColdStartReturnsError(t *testing.T) {
-	wantErr := errors.New("boom")
-	c := newStatsCache(func(_ context.Context) (database.Stats, error) {
-		return database.Stats{}, wantErr
-	})
-
-	if _, err := c.get(); !errors.Is(err, wantErr) {
-		t.Fatalf("get returned %v, want %v", err, wantErr)
-	}
-}
-
-// TestStatsCacheServesLastGoodCopyOnRefreshError checks that once a copy exists,
-// a later failing refresh does not surface an error or drop the good data.
-func TestStatsCacheServesLastGoodCopyOnRefreshError(t *testing.T) {
-	clk := &testClock{}
-	clk.ns.Store(int64(time.Hour))
-
-	const wantASNs = 7
-	var calls atomic.Int64
-	var failing atomic.Bool
-	c := newStatsCache(func(_ context.Context) (database.Stats, error) {
-		calls.Add(1)
-		if failing.Load() {
-			return database.Stats{}, errors.New("boom")
-		}
-
-		return database.Stats{ASNs: wantASNs}, nil
-	})
-	c.now = clk.now
-
-	got, err := c.get()
-	if err != nil || got.ASNs != wantASNs {
-		t.Fatalf("cold start returned (%+v, %v), want ASNs=%d, nil", got, err, wantASNs)
-	}
-
-	failing.Store(true)
-	clk.advance(c.interval)
-
-	got, err = c.get()
-	if err != nil {
-		t.Fatalf("get during failing refresh returned error: %v", err)
-	}
-	if got.ASNs != wantASNs {
-		t.Fatalf("get returned ASNs=%d, want the last good copy %d", got.ASNs, wantASNs)
-	}
-	if !waitForCalls(&calls, 2) {
-		t.Fatalf("refresh was not attempted, fetch ran %d times", calls.Load())
-	}
-}
-
-// TestStatsCacheBackgroundRefreshDoesNotLeak forces many stale refreshes and
-// checks the goroutine count returns to its starting value.
-func TestStatsCacheBackgroundRefreshDoesNotLeak(t *testing.T) {
-	clk := &testClock{}
-	clk.ns.Store(int64(time.Hour))
-
-	var calls atomic.Int64
-	c := newStatsCache(func(_ context.Context) (database.Stats, error) {
-		calls.Add(1)
-
-		return database.Stats{}, nil
-	})
-	c.now = clk.now
-
-	if _, err := c.get(); err != nil {
-		t.Fatalf("cold start returned error: %v", err)
-	}
-
-	baseline := runtime.NumGoroutine()
-
-	const rounds = 20
-	for i := range rounds {
-		clk.advance(c.interval)
-		if _, err := c.get(); err != nil {
-			t.Fatalf("get returned error: %v", err)
-		}
-		if !waitForCalls(&calls, int64(i+2)) {
-			t.Fatalf("refresh %d did not run", i)
-		}
-	}
-
-	const settleAttempts = 100
-	for range settleAttempts {
-		if runtime.NumGoroutine() <= baseline {
-			return
-		}
-		time.Sleep(10 * time.Millisecond)
-	}
-	t.Fatalf("goroutines did not settle to baseline %d, got %d", baseline, runtime.NumGoroutine())
-}

internal/streamer/streamer.go

@@ -210,9 +210,14 @@ func (s *Streamer) Start() error {
 // the connection status in metrics. This method is safe to call multiple times.
 func (s *Streamer) Stop() {
 	s.mu.Lock()
-	if s.cancel != nil {
-		s.cancel()
+	if s.cancel == nil {
+		// Not started, or already stopped: closing the queues again would panic.
+		s.mu.Unlock()
+
+		return
 	}
+	s.cancel()
+	s.cancel = nil
 	// Close all handler queues to signal workers to stop
 	for _, info := range s.handlers {
 		close(info.queue)
@@ -660,8 +665,15 @@ func (s *Streamer) stream(ctx context.Context) error {
 			continue
 		}
 
-		// Dispatch to interested handlers
+		// Dispatch to interested handlers. Stop cancels ctx and closes the
+		// queues under the write lock, so if ctx is cancelled here, under the
+		// read lock, the queues are closed and must not be sent to.
 		s.mu.RLock()
+		if ctx.Err() != nil {
+			s.mu.RUnlock()
+
+			return ctx.Err()
+		}
 		for _, info := range s.handlers {
 			if !info.handler.WantsMessage(msg.Type) {
 				continue

internal/streamer/streamer_test.go

@@ -2,6 +2,8 @@ package streamer
 
 import (
 	"context"
+	"errors"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"runtime"
@@ -10,6 +12,7 @@ import (
 
 	"git.eeqj.de/sneak/routewatch/internal/logger"
 	"git.eeqj.de/sneak/routewatch/internal/metrics"
+	"git.eeqj.de/sneak/routewatch/internal/ristypes"
 )
 
 func TestNewStreamer(t *testing.T) {
@@ -75,6 +78,45 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
 	}
 }
 
+// updateHandler wants UPDATE messages and does nothing with them.
+type updateHandler struct{}
+
+func (updateHandler) WantsMessage(messageType string) bool { return messageType == "UPDATE" }
+func (updateHandler) HandleMessage(*ristypes.RISMessage)   {}
+func (updateHandler) QueueCapacity() int                   { return 10 }
+
+// TestStopBeforeMessageReachesQueues stops the streamer after the read loop
+// has checked for cancellation but before it hands the message to the handler
+// queues. That is the gap a stop from another goroutine can land in, and it
+// used to end in "send on closed channel". The raw handler runs in that gap on
+// the read loop itself, so calling Stop from it hits the gap every time.
+func TestStopBeforeMessageReachesQueues(t *testing.T) {
+	const line = `{"type":"ris_message","data":{"type":"UPDATE","peer":"192.0.2.1",` +
+		`"peer_asn":"64496","timestamp":1700000000}}` + "\n"
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, line)
+	}))
+	defer srv.Close()
+
+	s := New(logger.New(), metrics.New())
+	s.url = srv.URL
+	s.RegisterHandler(updateHandler{})
+	s.RegisterRawHandler(func(string) { s.Stop() })
+
+	// Start would run the stream in the background, where the test cannot
+	// wait for it. Setting cancel as Start does lets Stop cancel the stream
+	// run here instead.
+	ctx, cancel := context.WithCancel(context.Background())
+	s.cancel = cancel
+
+	if err := s.stream(ctx); !errors.Is(err, context.Canceled) {
+		t.Fatalf("stream returned %v, want %v", err, context.Canceled)
+	}
+
+	// A second Stop must not close the queues again.
+	s.Stop()
+}
+
 // settledGoroutineCount lets transient goroutines finish, then reports the
 // current count.
 func settledGoroutineCount() int {