diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..063d4cc --- /dev/null +++ b/.dockerignore @@ -0,0 +1,28 @@ +# Docker does not read .gitignore, and a pattern here matches from the root of +# the build context only: a pattern meant for every directory needs `**/`. + +.git + +# Local build and debug output: `make build`, `make run`, `make asupdate`, test +# binaries, coverage profiles and source archives. +/bin +/log.txt +/out +/pkg/asinfo/asdata.json +**/*.tar.zst +**/*.test +**/*.out +**/*.tmp + +# Local databases and secrets. The image carries a source archive of the whole +# build context, so these would otherwise ship inside it. +**/*.db +**/*.db-journal +**/*.db-wal +**/.env + +# A local Go workspace points at directories outside the build context. +/go.work +/go.work.sum + +**/.DS_Store diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 0000000..2fe0ce0 --- /dev/null +++ b/.editorconfig @@ -0,0 +1,12 @@ +root = true + +[*] +indent_style = space +indent_size = 4 +end_of_line = lf +charset = utf-8 +trim_trailing_whitespace = true +insert_final_newline = true + +[Makefile] +indent_style = tab diff --git a/.gitea/workflows/check.yml b/.gitea/workflows/check.yml new file mode 100644 index 0000000..a55bc55 --- /dev/null +++ b/.gitea/workflows/check.yml @@ -0,0 +1,9 @@ +name: check +on: [push] +jobs: + check: + runs-on: ubuntu-latest + steps: + # actions/checkout v4.2.2, 2026-02-28 + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 + - run: script/cibuild diff --git a/Dockerfile b/Dockerfile index fa1460b..cedd1c9 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,22 @@ +# Lint stage — fast feedback on formatting and lint issues. +# The golangci-lint image bundles Go, gcc and make, so it can run go vet on +# the CGO sqlite package and golangci-lint without extra installs. +# golangci/golangci-lint:v2.7.2 (Go 1.25.5), 2026-09-21 +FROM golangci/golangci-lint@sha256:5d6d5c70a61f1356adfd9dd6316ce286799fefc9d743421356ff1b00842368ba AS lint + +WORKDIR /src + +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . + +RUN make fmt-check +RUN make lint + # Build stage -FROM golang:1.24-bookworm AS builder +# golang:1.24-bookworm, 2026-09-21 +FROM golang@sha256:1a6d4452c65dea36aac2e2d606b01b4a029ec90cc1ae53890540ce6173ea77ac AS builder # Install build dependencies (zstd for archive, gcc for CGO/sqlite3) RUN apt-get update && apt-get install -y --no-install-recommends \ @@ -10,12 +27,19 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ WORKDIR /src +# Force BuildKit to run the lint stage before compiling or testing. +COPY --from=lint /src/go.sum /dev/null + # Copy everything COPY . . # Vendor dependencies (must be after copying source) RUN go mod download && go mod vendor +# Run the test suite in the build stage: -race needs cgo and the C compiler +# installed above. The suite is offline (the live-feed test is opt-in). +RUN make test + # Build the binary with CGO enabled (required for sqlite3) RUN CGO_ENABLED=1 GOOS=linux go build -o /routewatch ./cmd/routewatch @@ -26,7 +50,8 @@ RUN tar --zstd -cf /routewatch-source.tar.zst \ . # Runtime stage -FROM debian:bookworm-slim +# debian:bookworm-slim, 2026-09-21 +FROM debian@sha256:3783cc01769c7b2b1b83a5c5ad96c815348e28ed7da68e2e3687004faa906251 # Install runtime dependencies # - ca-certificates: for HTTPS connections @@ -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" ] diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..3274443 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Jeffrey Paul + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md index ccf2218..cf77465 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,9 @@ # RouteWatch -RouteWatch is a real-time BGP routing table monitor that streams BGP UPDATE messages from the RIPE RIS Live service, maintains a live routing table in SQLite, and provides HTTP APIs for querying routing information. +RouteWatch is an MIT-licensed Go daemon by @sneak that monitors the BGP routing +table in real time: it streams BGP UPDATE messages from the RIPE RIS Live +service, maintains a live routing table in SQLite, and provides HTTP APIs for +querying routing information. ## Features @@ -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) diff --git a/TODO.md b/TODO.md index e4df8b9..b967d9a 100644 --- a/TODO.md +++ b/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) diff --git a/entrypoint.sh b/entrypoint.sh index 82acbbe..50d1aed 100644 --- a/entrypoint.sh +++ b/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 diff --git a/internal/config/config.go b/internal/config/config.go index 14ae95d..87a6ce3 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "runtime" + "strconv" "time" ) @@ -18,6 +19,12 @@ const ( // defaultRouteExpirationMinutes is the default route expiration timeout in minutes defaultRouteExpirationMinutes = 5 + + // defaultPort is the HTTP port used when PORT is not set + defaultPort = 8080 + + // maxPort is the highest TCP port number + maxPort = 65535 ) // Config holds configuration for the entire application @@ -25,6 +32,9 @@ type Config struct { // StateDir is the directory for all application state (database, snapshots) StateDir string + // Port is the TCP port the HTTP server listens on + Port int + // MaxRuntime is the maximum runtime (0 = run forever) MaxRuntime time.Duration @@ -43,8 +53,14 @@ func New() (*Config, error) { return nil, fmt.Errorf("failed to determine state directory: %w", err) } + port, err := getPort() + if err != nil { + return nil, err + } + return &Config{ StateDir: stateDir, + Port: port, MaxRuntime: 0, // Run forever by default EnableBatchedDatabaseWrites: true, // Enable batching by default RouteExpirationTimeout: defaultRouteExpirationMinutes * time.Minute, // For active route monitoring @@ -69,13 +85,20 @@ func getStateDirectory() (string, error) { return filepath.Join(home, "Library", "Application Support", AppIdentifier), nil case "linux", "freebsd", "openbsd", "netbsd": + // The XDG spec requires an absolute path; a relative one would put + // the database somewhere unexpected. + xdgData := os.Getenv("XDG_DATA_HOME") + if xdgData != "" && !filepath.IsAbs(xdgData) { + return "", fmt.Errorf("XDG_DATA_HOME must be an absolute path, got %q", xdgData) + } + // Unix-like: /var/lib/berlin.sneak.app.routewatch if root, else XDG_DATA_HOME if os.Geteuid() == 0 { return filepath.Join("/var/lib", AppIdentifier), nil } // Check XDG_DATA_HOME first - if xdgData := os.Getenv("XDG_DATA_HOME"); xdgData != "" { + if xdgData != "" { return filepath.Join(xdgData, AppIdentifier), nil } @@ -92,6 +115,22 @@ func getStateDirectory() (string, error) { } } +// getPort returns the HTTP port from PORT, or defaultPort when PORT is not set +func getPort() (int, error) { + value := os.Getenv("PORT") + if value == "" { + return defaultPort, nil + } + + // ParseUint, unlike Atoi, refuses a sign: the health check URL cannot use "+9090" + port, err := strconv.ParseUint(value, 10, 0) + if err != nil || port < 1 || port > maxPort { + return 0, fmt.Errorf("PORT must be a whole number from 1 to %d, got %q", maxPort, value) + } + + return int(port), nil +} + // EnsureDirectories creates all necessary directories if they don't exist func (c *Config) EnsureDirectories() error { // Ensure state directory exists diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..84f9414 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,62 @@ +package config + +import ( + "runtime" + "testing" +) + +func TestNewReadsPort(t *testing.T) { + tests := map[string]int{ + "": defaultPort, + "1": 1, + "9090": 9090, + "65535": 65535, + } + + for value, want := range tests { + t.Run(value, func(t *testing.T) { + t.Setenv("PORT", value) + t.Setenv("XDG_DATA_HOME", "") + + cfg, err := New() + if err != nil { + t.Fatalf("New() with PORT=%q: %v", value, err) + } + + if cfg.Port != want { + t.Errorf("New() with PORT=%q: Port = %d, want %d", value, cfg.Port, want) + } + }) + } +} + +func TestNewRefusesInvalidPort(t *testing.T) { + for _, value := range []string{"0", "65536", "-1", "+9090", "http", "80.5"} { + t.Run(value, func(t *testing.T) { + t.Setenv("PORT", value) + t.Setenv("XDG_DATA_HOME", "") + + if _, err := New(); err == nil { + t.Errorf("New() with PORT=%q returned no error", value) + } + }) + } +} + +func TestNewRefusesRelativeXDGDataHome(t *testing.T) { + if runtime.GOOS == "darwin" { + t.Skip("macOS does not read XDG_DATA_HOME") + } + + t.Setenv("PORT", "") + + t.Setenv("XDG_DATA_HOME", "relative/path") + if _, err := New(); err == nil { + t.Error("New() with a relative XDG_DATA_HOME returned no error") + } + + t.Setenv("XDG_DATA_HOME", "/var/lib") + if _, err := New(); err != nil { + t.Errorf("New() with XDG_DATA_HOME=/var/lib: %v", err) + } +} diff --git a/internal/database/counts.go b/internal/database/counts.go new file mode 100644 index 0000000..a305ba2 --- /dev/null +++ b/internal/database/counts.go @@ -0,0 +1,146 @@ +package database + +import ( + "context" + "fmt" + "sync" +) + +// liveCounts holds the running row counts that the stats endpoints report. They +// are seeded once at startup from the tables and then adjusted on every write, +// so a stats read serves them from memory instead of running a COUNT(*) over +// each table. Those scans, once the database passed a few GiB, took the whole +// request timeout and made /api/v1/stats return 500 (issue 27). +// +// A single mutex guards all fields so the stats reader takes a consistent +// snapshot at one instant and writers, which already run under the database +// write lock, adjust the counts after their transaction commits. +type liveCounts struct { + mu sync.RWMutex + asns int + prefixesV4 int + prefixesV6 int + peerings int + peers int + routesV4 int + routesV6 int +} + +// seed sets every count to the value read from the tables at startup. It runs +// before any writer, so it needs no coordination with the adjust methods. +func (c *liveCounts) seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6 int) { + c.mu.Lock() + defer c.mu.Unlock() + + c.asns = asns + c.prefixesV4 = prefixesV4 + c.prefixesV6 = prefixesV6 + c.peerings = peerings + c.peers = peers + c.routesV4 = routesV4 + c.routesV6 = routesV6 +} + +// addASNs adds n to the ASN count. +func (c *liveCounts) addASNs(n int) { + c.mu.Lock() + c.asns += n + c.mu.Unlock() +} + +// addPrefixes adds to the IPv4 and IPv6 prefix counts. +func (c *liveCounts) addPrefixes(v4, v6 int) { + c.mu.Lock() + c.prefixesV4 += v4 + c.prefixesV6 += v6 + c.mu.Unlock() +} + +// addPeerings adds n to the peering count. +func (c *liveCounts) addPeerings(n int) { + c.mu.Lock() + c.peerings += n + c.mu.Unlock() +} + +// addPeers adds n to the BGP peer count. +func (c *liveCounts) addPeers(n int) { + c.mu.Lock() + c.peers += n + c.mu.Unlock() +} + +// addRoutes adds to the IPv4 and IPv6 live-route counts. Deletions pass +// negative values. +func (c *liveCounts) addRoutes(v4, v6 int) { + c.mu.Lock() + c.routesV4 += v4 + c.routesV6 += v6 + c.mu.Unlock() +} + +// fill copies the counts into a Stats, including the derived totals, under a +// single read lock so the reader sees one consistent snapshot. +func (c *liveCounts) fill(s *Stats) { + c.mu.RLock() + defer c.mu.RUnlock() + + s.ASNs = c.asns + s.IPv4Prefixes = c.prefixesV4 + s.IPv6Prefixes = c.prefixesV6 + s.Prefixes = c.prefixesV4 + c.prefixesV6 + s.Peerings = c.peerings + s.Peers = c.peers + s.IPv4Routes = c.routesV4 + s.IPv6Routes = c.routesV6 + s.LiveRoutes = c.routesV4 + c.routesV6 +} + +// countRows returns the number of rows in the named table. It is used only at +// startup to seed the in-memory counters, so a full COUNT(*) is acceptable. +func (d *Database) countRows(ctx context.Context, table string) (int, error) { + var n int + // table is one of a fixed set of literals below, never external input. + if err := d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+table).Scan(&n); err != nil { + return 0, fmt.Errorf("failed to count %s: %w", table, err) + } + + return n, nil +} + +// seedCounts reads the current row counts from the tables into the in-memory +// counters. It runs once at startup, before the streamer begins writing. +func (d *Database) seedCounts(ctx context.Context) error { + asns, err := d.countRows(ctx, "asns") + if err != nil { + return err + } + prefixesV4, err := d.countRows(ctx, "prefixes_v4") + if err != nil { + return err + } + prefixesV6, err := d.countRows(ctx, "prefixes_v6") + if err != nil { + return err + } + peerings, err := d.countRows(ctx, "peerings") + if err != nil { + return err + } + peers, err := d.countRows(ctx, "bgp_peers") + if err != nil { + return err + } + routesV4, err := d.countRows(ctx, "live_routes_v4") + if err != nil { + return err + } + routesV6, err := d.countRows(ctx, "live_routes_v6") + if err != nil { + return err + } + + d.counts.seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6) + + return nil +} diff --git a/internal/database/counts_test.go b/internal/database/counts_test.go new file mode 100644 index 0000000..6db2859 --- /dev/null +++ b/internal/database/counts_test.go @@ -0,0 +1,335 @@ +package database + +import ( + "context" + "sync" + "testing" + "time" + + "git.eeqj.de/sneak/routewatch/internal/config" + "git.eeqj.de/sneak/routewatch/internal/logger" + "github.com/google/uuid" +) + +// mkV4Route builds an IPv4 live route with its range columns populated. +func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute { + t.Helper() + + start, end, err := CalculateIPv4Range(prefix) + if err != nil { + t.Fatalf("CalculateIPv4Range(%s): %v", prefix, err) + } + + return &LiveRoute{ + ID: uuid.New(), + Prefix: prefix, + MaskLength: 24, + IPVersion: ipVersionV4, + OriginASN: asn, + PeerIP: "192.0.2.1", + ASPath: []int{asn}, + NextHop: "192.0.2.254", + LastUpdated: ts, + V4IPStart: &start, + V4IPEnd: &end, + } +} + +// mkV6Route builds an IPv6 live route. +func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute { + return &LiveRoute{ + ID: uuid.New(), + Prefix: prefix, + MaskLength: 32, + IPVersion: ipVersionV6, + OriginASN: asn, + PeerIP: "2001:db8::1", + ASPath: []int{asn}, + NextHop: "2001:db8::ffff", + LastUpdated: ts, + } +} + +// TestLiveCountsTrackWritesInRealtime checks that the stats counts start at +// zero, reflect each write the moment it commits (no recompute, no timer), do +// not move when a route is merely re-announced, and drop when a route is +// deleted. These counts are what /api/v1/stats reports; before this change the +// endpoint recomputed them with a COUNT(*) over each table on every request. +func TestLiveCountsTrackWritesInRealtime(t *testing.T) { + cfg := &config.Config{StateDir: t.TempDir()} + + db, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + ctx := context.Background() + + empty, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext on empty database: %v", err) + } + if empty.ASNs != 0 || empty.Prefixes != 0 || empty.Peerings != 0 || + empty.Peers != 0 || empty.LiveRoutes != 0 { + t.Fatalf("empty database counts nonzero: %+v", empty) + } + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + if err := db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts}); err != nil { + t.Fatalf("GetOrCreateASNBatch: %v", err) + } + if err := db.UpdatePrefixesBatch(map[string]time.Time{ + "198.51.100.0/24": ts, + "2001:db8::/32": ts, + }); err != nil { + t.Fatalf("UpdatePrefixesBatch: %v", err) + } + if err := db.UpdatePeerBatch(map[string]PeerUpdate{ + "192.0.2.1": {PeerIP: "192.0.2.1", PeerASN: 64500, MessageType: "UPDATE", Timestamp: ts}, + }); err != nil { + t.Fatalf("UpdatePeerBatch: %v", err) + } + if err := db.RecordPeering(64500, 64501, ts); err != nil { + t.Fatalf("RecordPeering: %v", err) + } + + routes := []*LiveRoute{ + mkV4Route(t, "198.51.100.0/24", 64500, ts), + mkV4Route(t, "203.0.113.0/24", 64501, ts.Add(time.Minute)), + mkV6Route("2001:db8::/32", 64502, ts.Add(2*time.Minute)), + } + if err := db.UpsertLiveRouteBatch(routes); err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + stats, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + assertCounts(t, "after inserts", stats, wantCounts{ + asns: 2, prefixes: 2, peerings: 1, peers: 1, + ipv4Routes: 2, ipv6Routes: 1, liveRoutes: 3, + }) + + // Re-announcing the same routes is an update, not an insert: counts hold. + if err := db.UpsertLiveRouteBatch(routes); err != nil { + t.Fatalf("UpsertLiveRouteBatch (re-announce): %v", err) + } + stats, err = db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + assertCounts(t, "after re-announce", stats, wantCounts{ + asns: 2, prefixes: 2, peerings: 1, peers: 1, + ipv4Routes: 2, ipv6Routes: 1, liveRoutes: 3, + }) + + // A withdrawal removes one route. + if err := db.DeleteLiveRouteBatch([]LiveRouteDeletion{ + {Prefix: "203.0.113.0/24", OriginASN: 64501, PeerIP: "192.0.2.1", IPVersion: ipVersionV4}, + }); err != nil { + t.Fatalf("DeleteLiveRouteBatch: %v", err) + } + stats, err = db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + assertCounts(t, "after delete", stats, wantCounts{ + asns: 2, prefixes: 2, peerings: 1, peers: 1, + ipv4Routes: 1, ipv6Routes: 1, liveRoutes: 2, + }) +} + +// TestLiveCountsSeededFromDatabaseAtStartup writes rows, reopens the same +// database file, and checks the counts come back from the seed scan rather than +// starting at zero. +func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) { + cfg := &config.Config{StateDir: t.TempDir()} + + db, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + if err := db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts, 64502: ts}); err != nil { + t.Fatalf("GetOrCreateASNBatch: %v", err) + } + if err := db.UpsertLiveRouteBatch([]*LiveRoute{ + mkV4Route(t, "198.51.100.0/24", 64500, ts), + mkV6Route("2001:db8::/32", 64502, ts), + }); err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + if err := db.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + reopened, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to reopen database: %v", err) + } + defer func() { _ = reopened.Close() }() + + stats, err := reopened.GetStatsContext(context.Background()) + if err != nil { + t.Fatalf("GetStatsContext after reopen: %v", err) + } + if stats.ASNs != 3 { + t.Errorf("seeded ASNs = %d, want 3", stats.ASNs) + } + if stats.IPv4Routes != 1 || stats.IPv6Routes != 1 || stats.LiveRoutes != 2 { + t.Errorf("seeded routes = (v4 %d, v6 %d, total %d), want (1, 1, 2)", + stats.IPv4Routes, stats.IPv6Routes, stats.LiveRoutes) + } +} + +// TestStatsRouteTimestamps checks the oldest/newest route timestamps are read +// from the right rows across both tables and parse into time.Time. The old +// MIN/MAX union query read its result into *time.Time, which the driver could +// not parse, so it logged a warning every call and left both timestamps nil. +func TestStatsRouteTimestamps(t *testing.T) { + cfg := &config.Config{StateDir: t.TempDir()} + + db, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + ctx := context.Background() + + empty, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext on empty database: %v", err) + } + if empty.OldestRoute != nil || empty.NewestRoute != nil { + t.Fatalf("empty database timestamps = (%v, %v), want (nil, nil)", + empty.OldestRoute, empty.NewestRoute) + } + + base := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + oldest := base + newest := base.Add(2 * time.Minute) + if err := db.UpsertLiveRouteBatch([]*LiveRoute{ + mkV4Route(t, "198.51.100.0/24", 64500, base.Add(time.Minute)), + mkV4Route(t, "203.0.113.0/24", 64501, oldest), + mkV6Route("2001:db8::/32", 64502, newest), + }); err != nil { + t.Fatalf("UpsertLiveRouteBatch: %v", err) + } + + stats, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + if stats.OldestRoute == nil || !stats.OldestRoute.Equal(oldest) { + t.Errorf("OldestRoute = %v, want %v", stats.OldestRoute, oldest) + } + if stats.NewestRoute == nil || !stats.NewestRoute.Equal(newest) { + t.Errorf("NewestRoute = %v, want %v", stats.NewestRoute, newest) + } +} + +// TestLiveCountsConcurrentReadWrite runs writers and stats readers at once so +// the race detector proves the counters are safe under concurrent use. +func TestLiveCountsConcurrentReadWrite(t *testing.T) { + cfg := &config.Config{StateDir: t.TempDir()} + + db, err := New(cfg, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + + const writers = 4 + var wg sync.WaitGroup + + wg.Add(writers) + for w := range writers { + go func(base int) { + defer wg.Done() + for i := range 25 { + asn := 65000 + base*100 + i + route := mkV6Route("2001:db8::/32", asn, ts) + route.PeerIP = "2001:db8::" + uuid.NewString() + if err := db.UpsertLiveRoute(route); err != nil { + t.Errorf("UpsertLiveRoute: %v", err) + + return + } + } + }(w) + } + + var readerWG sync.WaitGroup + readerWG.Add(1) + stop := make(chan struct{}) + go func() { + defer readerWG.Done() + for { + select { + case <-stop: + return + default: + if _, err := db.GetStatsContext(context.Background()); err != nil { + t.Errorf("GetStatsContext: %v", err) + + return + } + } + } + }() + + wg.Wait() + close(stop) + readerWG.Wait() + + stats, err := db.GetStatsContext(context.Background()) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + if want := writers * 25; stats.IPv6Routes != want { + t.Errorf("IPv6Routes = %d, want %d", stats.IPv6Routes, want) + } +} + +type wantCounts struct { + asns int + prefixes int + peerings int + peers int + ipv4Routes int + ipv6Routes int + liveRoutes int +} + +func assertCounts(t *testing.T, when string, got Stats, want wantCounts) { + t.Helper() + + if got.ASNs != want.asns { + t.Errorf("%s: ASNs = %d, want %d", when, got.ASNs, want.asns) + } + if got.Prefixes != want.prefixes { + t.Errorf("%s: Prefixes = %d, want %d", when, got.Prefixes, want.prefixes) + } + if got.Peerings != want.peerings { + t.Errorf("%s: Peerings = %d, want %d", when, got.Peerings, want.peerings) + } + if got.Peers != want.peers { + t.Errorf("%s: Peers = %d, want %d", when, got.Peers, want.peers) + } + if got.IPv4Routes != want.ipv4Routes { + t.Errorf("%s: IPv4Routes = %d, want %d", when, got.IPv4Routes, want.ipv4Routes) + } + if got.IPv6Routes != want.ipv6Routes { + t.Errorf("%s: IPv6Routes = %d, want %d", when, got.IPv6Routes, want.ipv6Routes) + } + if got.LiveRoutes != want.liveRoutes { + t.Errorf("%s: LiveRoutes = %d, want %d", when, got.LiveRoutes, want.liveRoutes) + } +} diff --git a/internal/database/database.go b/internal/database/database.go index aac7744..4007b48 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -38,6 +38,22 @@ const ( maxIPv4 = 0xFFFFFFFF ) +// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every +// pooled connection gets them; the heap limits are process-wide and set once. +const ( + // sqliteCacheSizeKiB is the per-connection page cache; negative means KiB. + // -65536 = 64 MiB, so at most 640 MiB across the 10-connection pool. + sqliteCacheSizeKiB = -65536 + // sqliteBusyTimeoutMs is how long a connection waits on a locked database. + sqliteBusyTimeoutMs = 5000 + // sqliteSoftHeapLimitBytes (1 GiB) makes SQLite recycle its cache rather + // than allocate once its C heap passes this size. + sqliteSoftHeapLimitBytes = 1073741824 + // sqliteHardHeapLimitBytes (1.5 GiB) fails a statement with SQLITE_NOMEM + // instead of growing the C heap without bound. + sqliteHardHeapLimitBytes = 1610612736 +) + // Common errors var ( // ErrInvalidIP is returned when an IP address is malformed @@ -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) + inserted, err := upsertRouteRowV4(updV4, insV4, route, string(pathJSON)) + if err != nil { + return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) + } + if inserted { + newV4++ } - _, err = stmtV4.Exec( - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - *route.V4IPStart, - *route.V4IPEnd, - ) - } else { - // IPv6 routes - _, err = stmtV6.Exec( - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - ) + continue } + + inserted, err := upsertRouteRowV6(updV6, insV6, route, string(pathJSON)) if err != nil { return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) } + if inserted { + newV6++ + } } if err = tx.Commit(); err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } + d.counts.addRoutes(newV4, newV6) + return nil } @@ -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) } - if route.IPVersion == ipVersionV4 { - // Convert v4_ip_start and v4_ip_end to interface{} for SQL NULL handling - var v4Start, v4End interface{} - if route.V4IPStart != nil { - v4Start = *route.V4IPStart - } - if route.V4IPEnd != nil { - v4End = *route.V4IPEnd - } - - _, err = d.db.Exec(query, - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - v4Start, - v4End, - ) - } else { - _, err = d.db.Exec(query, - route.ID.String(), - route.Prefix, - route.MaskLength, - route.OriginASN, - route.PeerIP, - string(pathJSON), - route.NextHop, - route.LastUpdated, - ) + updateSQL, insertSQL := updateLiveRouteV4SQL, insertLiveRouteV4SQL + if route.IPVersion == ipVersionV6 { + updateSQL, insertSQL = updateLiveRouteV6SQL, insertLiveRouteV6SQL } - return err + // The write lock is held, so no other writer can insert this key between the + // update and the insert even though they are separate autocommit statements. + upd, err := d.db.Prepare(updateSQL) + if err != nil { + return fmt.Errorf("failed to prepare update statement: %w", err) + } + defer func() { _ = upd.Close() }() + + ins, err := d.db.Prepare(insertSQL) + if err != nil { + return fmt.Errorf("failed to prepare insert statement: %w", err) + } + defer func() { _ = ins.Close() }() + + var inserted bool + if route.IPVersion == ipVersionV4 { + inserted, err = upsertRouteRowV4(upd, ins, route, string(pathJSON)) + } else { + inserted, err = upsertRouteRowV6(upd, ins, route, string(pathJSON)) + } + if err != nil { + return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err) + } + + if inserted { + if route.IPVersion == ipVersionV4 { + d.counts.addRoutes(1, 0) + } else { + d.counts.addRoutes(0, 1) + } + } + + return nil } // DeleteLiveRoute deletes a live route @@ -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 + + // Literal per-table queries (rather than one formatted with the table name) + // so the delete carries no dynamically built SQL. A delete with no origin + // ASN can remove several rows. + var res sql.Result + switch { + case isV4 && originASN == 0: + res, err = d.db.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP) + case isV4: + res, err = d.db.Exec( + `DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, + prefix, originASN, peerIP) + case originASN == 0: + res, err = d.db.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP) + default: + res, err = d.db.Exec( + `DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, + prefix, originASN, peerIP) + } + if err != nil { + return err } - var query string - if originASN == 0 { - // Delete all routes for this prefix from this peer - query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND peer_ip = ?`, tableName) - _, err = d.db.Exec(query, prefix, peerIP) + affected, err := res.RowsAffected() + if err != nil { + return err + } + if isV4 { + d.counts.addRoutes(-int(affected), 0) } else { - // Delete specific route - query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, tableName) - _, err = d.db.Exec(query, prefix, originASN, peerIP) + d.counts.addRoutes(0, -int(affected)) } - return err + return nil } // GetPrefixDistribution returns the distribution of unique prefixes by mask length diff --git a/internal/database/database_test.go b/internal/database/database_test.go index 6636871..fa4bb48 100644 --- a/internal/database/database_test.go +++ b/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() diff --git a/internal/database/interface.go b/internal/database/interface.go index 7f3f8e9..038dfd5 100644 --- a/internal/database/interface.go +++ b/internal/database/interface.go @@ -18,6 +18,8 @@ type Stats struct { Peers int FileSizeBytes int64 LiveRoutes int + IPv4Routes int + IPv6Routes int OldestRoute *time.Time NewestRoute *time.Time IPv4PrefixDistribution []PrefixDistribution diff --git a/internal/ristypes/ris.go b/internal/ristypes/ris.go index ba36b91..a6831a4 100644 --- a/internal/ristypes/ris.go +++ b/internal/ristypes/ris.go @@ -63,24 +63,28 @@ type RISLiveMessage struct { // the actual BGP update data including AS path, communities, announcements, // and withdrawals. type RISMessage struct { - Type string `json:"type"` - Timestamp float64 `json:"timestamp"` - ParsedTimestamp time.Time `json:"-"` // Parsed from Timestamp field - Peer string `json:"peer"` - PeerASN string `json:"peer_asn"` - ID string `json:"id"` - Host string `json:"host"` - RRC string `json:"rrc,omitempty"` - MrtTime float64 `json:"mrt_time,omitempty"` - SocketTime float64 `json:"socket_time,omitempty"` - Path ASPath `json:"path,omitempty"` - Community [][]int `json:"community,omitempty"` - Origin string `json:"origin,omitempty"` - MED *int `json:"med,omitempty"` - LocalPref *int `json:"local_pref,omitempty"` - Announcements []RISAnnouncement `json:"announcements,omitempty"` - Withdrawals []string `json:"withdrawals,omitempty"` - Raw string `json:"raw,omitempty"` + Type string `json:"type"` + Timestamp float64 `json:"timestamp"` + ParsedTimestamp time.Time `json:"-"` // Parsed from Timestamp field + Peer string `json:"peer"` + PeerASN string `json:"peer_asn"` + ID string `json:"id"` + Host string `json:"host"` + RRC string `json:"rrc,omitempty"` + MrtTime float64 `json:"mrt_time,omitempty"` + SocketTime float64 `json:"socket_time,omitempty"` + Path ASPath `json:"path,omitempty"` + // Community and Raw are present in the feed but read by no handler. + // They are the largest fields on a message that lives in up to four + // handler queues, so json:"-" keeps them out of the decoded message + // to save queue memory. Do not decode them without a consumer. + Community [][]int `json:"-"` + Origin string `json:"origin,omitempty"` + MED *int `json:"med,omitempty"` + LocalPref *int `json:"local_pref,omitempty"` + Announcements []RISAnnouncement `json:"announcements,omitempty"` + Withdrawals []string `json:"withdrawals,omitempty"` + Raw string `json:"-"` } // RISAnnouncement represents a BGP route announcement within a RIS message. diff --git a/internal/ristypes/ris_test.go b/internal/ristypes/ris_test.go new file mode 100644 index 0000000..6a3b7b0 --- /dev/null +++ b/internal/ristypes/ris_test.go @@ -0,0 +1,67 @@ +package ristypes + +import ( + "bufio" + "encoding/json" + "os" + "testing" +) + +// messageExamplesPath is the captured RIS Live feed used as decode fixtures, +// one JSON message per line. +const messageExamplesPath = "../../docs/message-examples.json" + +// TestDecodeDropsCommunityAndRaw decodes every captured message the way the +// streamer does and checks that the fields no handler reads (Community, Raw) +// stay empty while the fields handlers use (Path, Announcements) still decode. +func TestDecodeDropsCommunityAndRaw(t *testing.T) { + f, err := os.Open(messageExamplesPath) + if err != nil { + t.Fatalf("open fixtures: %v", err) + } + defer f.Close() + + var messages, withPath, withAnnouncements int + + scanner := bufio.NewScanner(f) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for scanner.Scan() { + line := scanner.Bytes() + if len(line) == 0 { + continue + } + + var wrapper RISLiveMessage + if err := json.Unmarshal(line, &wrapper); err != nil { + t.Fatalf("unmarshal message %d: %v", messages+1, err) + } + messages++ + + msg := wrapper.Data + if msg.Community != nil { + t.Errorf("message %d: Community decoded, want empty: %v", messages, msg.Community) + } + if msg.Raw != "" { + t.Errorf("message %d: Raw decoded, want empty", messages) + } + + if len(msg.Path) > 0 { + withPath++ + } + if len(msg.Announcements) > 0 { + withAnnouncements++ + } + } + if err := scanner.Err(); err != nil { + t.Fatalf("scan fixtures: %v", err) + } + + if messages == 0 { + t.Fatal("no messages decoded from fixtures") + } + // The fixtures include announcement messages; a used field must still decode, + // otherwise an empty Community/Raw would prove nothing. + if withPath == 0 || withAnnouncements == 0 { + t.Fatalf("used fields did not decode: withPath=%d withAnnouncements=%d", withPath, withAnnouncements) + } +} diff --git a/internal/routewatch/app_integration_test.go b/internal/routewatch/app_integration_test.go index 619210d..56efc06 100644 --- a/internal/routewatch/app_integration_test.go +++ b/internal/routewatch/app_integration_test.go @@ -426,6 +426,9 @@ func (m *mockStore) Ping(ctx context.Context) error { } func TestRouteWatchLiveFeed(t *testing.T) { + if testing.Short() { + t.Skip("skipping live RIPE RIS network feed test in short mode; run without -short to include it") + } // Create mock database mockDB := newMockStore() @@ -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{ diff --git a/internal/routewatch/ashandler.go b/internal/routewatch/ashandler.go index c1ff34f..3eae6f3 100644 --- a/internal/routewatch/ashandler.go +++ b/internal/routewatch/ashandler.go @@ -10,9 +10,11 @@ import ( ) const ( - // asHandlerQueueSize is the queue capacity for ASN operations - // DO NOT set this higher than 100000 without explicit instructions - asHandlerQueueSize = 100000 + // asHandlerQueueSize is the queue capacity for ASN operations, about 4 + // seconds of feed at peak. The streamer drops rather than blocks when a + // queue is full, so this bounds memory. Batches still flush on a timer + // (asnBatchTimeout), so a queue smaller than asnBatchSize is fine. + asHandlerQueueSize = 20000 // asnBatchSize is the number of ASN operations to batch together asnBatchSize = 30000 diff --git a/internal/routewatch/peerhandler.go b/internal/routewatch/peerhandler.go index 7ffabab..2f7bf70 100644 --- a/internal/routewatch/peerhandler.go +++ b/internal/routewatch/peerhandler.go @@ -14,8 +14,11 @@ import ( ) const ( - // peerHandlerQueueSize is the queue capacity for peer tracking operations - peerHandlerQueueSize = 100000 + // peerHandlerQueueSize is the queue capacity for peer tracking operations, + // about 4 seconds of feed at peak. The streamer drops rather than blocks + // when a queue is full, so this bounds memory. Batches still flush on a + // timer (peerBatchTimeout). + peerHandlerQueueSize = 20000 // peerBatchSize is the number of peer updates to batch together peerBatchSize = 10000 diff --git a/internal/routewatch/peeringhandler.go b/internal/routewatch/peeringhandler.go index 050cb23..b4f3277 100644 --- a/internal/routewatch/peeringhandler.go +++ b/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 - asPaths map[string]time.Time // key is JSON-encoded AS path + mu sync.Mutex + asPaths map[string]time.Time // key is JSON-encoded AS path + droppedPaths int // paths dropped because the map was full stopCh chan struct{} } // NewPeeringHandler creates and initializes a new PeeringHandler with the -// provided database store and logger. It starts two background goroutines: -// one for periodic processing of accumulated AS paths into peering records, -// and one for pruning expired paths from memory. The handler begins -// processing immediately upon creation. +// provided database store and logger. It starts one background goroutine that +// periodically processes accumulated AS paths into peering records. The +// handler begins processing immediately upon creation. func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler { h := &PeeringHandler{ db: db, @@ -65,9 +61,8 @@ func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler stopCh: make(chan struct{}), } - // Start the periodic processing goroutines + // Start the periodic processing goroutine go h.processLoop() - go h.pruneLoop() return h } @@ -106,9 +101,18 @@ func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) { return } + key := string(pathJSON) h.mu.Lock() - h.asPaths[string(pathJSON)] = timestamp + if _, exists := h.asPaths[key]; exists { + // Already tracked: refresh its timestamp. + h.asPaths[key] = timestamp + } else if len(h.asPaths) >= maxTrackedPaths { + // Map is full; drop this new path and count it. + h.droppedPaths++ + } else { + h.asPaths[key] = timestamp + } h.mu.Unlock() } @@ -130,41 +134,6 @@ func (h *PeeringHandler) processLoop() { } } -// pruneLoop runs periodically to remove old AS paths -func (h *PeeringHandler) pruneLoop() { - ticker := time.NewTicker(pathPruneInterval) - defer ticker.Stop() - - for { - select { - case <-ticker.C: - h.prunePaths() - case <-h.stopCh: - return - } - } -} - -// prunePaths removes AS paths older than pathExpirationTime -func (h *PeeringHandler) prunePaths() { - cutoff := time.Now().Add(-pathExpirationTime) - var removed int - - h.mu.Lock() - for pathKey, timestamp := range h.asPaths { - if timestamp.Before(cutoff) { - delete(h.asPaths, pathKey) - removed++ - } - } - pathCount := len(h.asPaths) - h.mu.Unlock() - - if removed > 0 { - h.logger.Debug("Pruned old AS paths", "removed", removed, "remaining", pathCount) - } -} - // ProcessPeeringsNow triggers immediate processing of all accumulated AS // paths into peering records. This bypasses the normal periodic processing // schedule and is primarily intended for testing purposes. @@ -174,15 +143,16 @@ func (h *PeeringHandler) ProcessPeeringsNow() { // processPeerings extracts peerings from AS paths and writes to database func (h *PeeringHandler) processPeerings() { - // Take a snapshot of current AS paths - h.mu.RLock() - pathsCopy := make(map[string]time.Time, len(h.asPaths)) - for k, v := range h.asPaths { - pathsCopy[k] = v - } - h.mu.RUnlock() + // Take the accumulated paths and replace the map with a fresh empty one + // under the lock. Each path is processed exactly once and the memory is + // released, so the map never grows past a single interval's traffic. + h.mu.Lock() + paths := h.asPaths + h.asPaths = make(map[string]time.Time) + dropped := h.droppedPaths + h.mu.Unlock() - if len(pathsCopy) == 0 { + if len(paths) == 0 { return } @@ -192,7 +162,7 @@ func (h *PeeringHandler) processPeerings() { } peerings := make(map[peeringKey]time.Time) - for pathJSON, timestamp := range pathsCopy { + for pathJSON, timestamp := range paths { var path []int if err := json.Unmarshal([]byte(pathJSON), &path); err != nil { h.logger.Error("Failed to decode AS path", "error", err) @@ -241,15 +211,16 @@ func (h *PeeringHandler) processPeerings() { } h.logger.Info("Processed AS peerings", - "paths", len(pathsCopy), + "paths", len(paths), "unique_peerings", len(peerings), "success", successCount, + "dropped_paths", dropped, "duration", time.Since(start), ) } // Stop gracefully shuts down the handler by signaling the background -// goroutines to stop and performing a final synchronous processing of +// goroutine to stop and performing a final synchronous processing of // any remaining AS paths. This ensures no peering data is lost during // shutdown. func (h *PeeringHandler) Stop() { diff --git a/internal/routewatch/peeringhandler_test.go b/internal/routewatch/peeringhandler_test.go new file mode 100644 index 0000000..364fa46 --- /dev/null +++ b/internal/routewatch/peeringhandler_test.go @@ -0,0 +1,163 @@ +package routewatch + +import ( + "encoding/json" + "strconv" + "sync" + "testing" + "time" + + "git.eeqj.de/sneak/routewatch/internal/database" + "git.eeqj.de/sneak/routewatch/internal/logger" + "git.eeqj.de/sneak/routewatch/internal/ristypes" +) + +const ( + testASNA = 64500 + testASNB = 64501 + testASNC = 64502 +) + +// recordingStore wraps mockStore to count every RecordPeering call, so a +// test can tell how many times a peering was written across separate runs. +type recordingStore struct { + *mockStore + + mu sync.Mutex + calls int +} + +func (r *recordingStore) RecordPeering(asA, asB int, ts time.Time) error { + r.mu.Lock() + r.calls++ + r.mu.Unlock() + + return r.mockStore.RecordPeering(asA, asB, ts) +} + +func (r *recordingStore) callCount() int { + r.mu.Lock() + defer r.mu.Unlock() + + return r.calls +} + +// newTestHandler builds a PeeringHandler without starting the periodic +// processing goroutine, so tests drive processing explicitly. +func newTestHandler(db database.Store) *PeeringHandler { + return &PeeringHandler{ + db: db, + logger: logger.New(), + asPaths: make(map[string]time.Time), + stopCh: make(chan struct{}), + } +} + +func snapshot(h *PeeringHandler) (tracked, dropped int) { + h.mu.Lock() + defer h.mu.Unlock() + + return len(h.asPaths), h.droppedPaths +} + +func pathKey(t *testing.T, asns ...int) string { + t.Helper() + + b, err := json.Marshal(ristypes.ASPath(asns)) + if err != nil { + t.Fatalf("failed to marshal path: %v", err) + } + + return string(b) +} + +func handle(h *PeeringHandler, ts time.Time, asns ...int) { + h.HandleMessage(&ristypes.RISMessage{ + Path: ristypes.ASPath(asns), + ParsedTimestamp: ts, + }) +} + +// TestPeeringHandlerProcessesEachRunAndEmpties verifies that a processing run +// empties the path map (the swap) and that a path seen again after a run is +// recorded in the next run too. +func TestPeeringHandlerProcessesEachRunAndEmpties(t *testing.T) { + store := &recordingStore{mockStore: newMockStore()} + h := newTestHandler(store) + + now := time.Now().UTC() + + // Run 1: one path, one peering recorded, map emptied afterwards. + handle(h, now, testASNA, testASNB) + h.ProcessPeeringsNow() + + if tracked, _ := snapshot(h); tracked != 0 { + t.Fatalf("map not empty after first run: %d paths remain", tracked) + } + if got := store.callCount(); got != 1 { + t.Fatalf("want 1 RecordPeering call after first run, got %d", got) + } + + // Run 2: the same path again is recorded again (RecordPeering upserts). + handle(h, now.Add(time.Second), testASNA, testASNB) + h.ProcessPeeringsNow() + + if tracked, _ := snapshot(h); tracked != 0 { + t.Fatalf("map not empty after second run: %d paths remain", tracked) + } + if got := store.callCount(); got != 2 { + t.Fatalf("want 2 RecordPeering calls after second run, got %d", got) + } +} + +// TestPeeringHandlerCapDropsAndCounts verifies that a full map drops new paths +// and counts them, while a path already tracked is refreshed rather than +// dropped. +func TestPeeringHandlerCapDropsAndCounts(t *testing.T) { + store := &recordingStore{mockStore: newMockStore()} + h := newTestHandler(store) + + now := time.Now().UTC() + + // Fill the map to exactly maxTrackedPaths, including one real path key so + // the "already tracked" branch can be exercised. The filler keys are never + // processed in this test, so their contents do not matter. + existing := pathKey(t, testASNA, testASNB) + + h.mu.Lock() + h.asPaths[existing] = now + for i := 0; len(h.asPaths) < maxTrackedPaths; i++ { + h.asPaths[strconv.Itoa(i)] = now + } + h.mu.Unlock() + + // A new path is dropped and counted because the map is full. + handle(h, now.Add(time.Second), testASNA, testASNC) + + tracked, dropped := snapshot(h) + if tracked != maxTrackedPaths { + t.Fatalf("want map size %d after drop, got %d", maxTrackedPaths, tracked) + } + if dropped != 1 { + t.Fatalf("want dropped count 1, got %d", dropped) + } + + // A path already tracked is refreshed, not dropped. + refreshed := now.Add(2 * time.Second) + handle(h, refreshed, testASNA, testASNB) + + tracked, dropped = snapshot(h) + if tracked != maxTrackedPaths { + t.Fatalf("want map size %d after refresh, got %d", maxTrackedPaths, tracked) + } + if dropped != 1 { + t.Fatalf("want dropped count still 1 after refresh, got %d", dropped) + } + + h.mu.Lock() + gotTS := h.asPaths[existing] + h.mu.Unlock() + if !gotTS.Equal(refreshed) { + t.Fatalf("existing path timestamp not refreshed: want %v, got %v", refreshed, gotTS) + } +} diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 4223b20..9ce5675 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -14,9 +14,11 @@ import ( ) const ( - // prefixHandlerQueueSize is the queue capacity for prefix tracking operations - // DO NOT set this higher than 100000 without explicit instructions - prefixHandlerQueueSize = 100000 + // prefixHandlerQueueSize is the queue capacity for prefix tracking + // operations, about 4 seconds of feed at peak. The streamer drops rather + // than blocks when a queue is full, so this bounds memory. Batches still + // flush on a timer (prefixBatchTimeout). + prefixHandlerQueueSize = 20000 // prefixBatchSize is the number of prefix updates to batch together prefixBatchSize = 25000 diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 98ff6b2..195cb33 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -179,9 +179,11 @@ func (s *Server) handleStatusJSON() http.HandlerFunc { metrics := s.streamer.GetMetrics() - // Get database stats with timeout - statsChan := make(chan database.Stats) - errChan := make(chan error) + // Get database stats with timeout. The channels are buffered so the + // goroutine's send never blocks if the timeout wins and nothing here + // receives; otherwise it would block forever and leak. + statsChan := make(chan database.Stats, 1) + errChan := make(chan error, 1) go func() { dbStats, err := s.db.GetStatsContext(ctx) @@ -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, diff --git a/internal/server/handlers_test.go b/internal/server/handlers_test.go new file mode 100644 index 0000000..46756bd --- /dev/null +++ b/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 +} diff --git a/internal/server/server.go b/internal/server/server.go index afdd426..10f8649 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -4,9 +4,10 @@ package server import ( "context" "net/http" - "os" + "strconv" "time" + "git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/database" "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/streamer" @@ -33,16 +34,18 @@ type Server struct { db database.Store streamer *streamer.Streamer logger *logger.Logger + port int srv *http.Server asnFetcher ASNFetcher } // New creates a new HTTP server -func New(db database.Store, streamer *streamer.Streamer, logger *logger.Logger) *Server { +func New(db database.Store, streamer *streamer.Streamer, logger *logger.Logger, cfg *config.Config) *Server { s := &Server{ db: db, streamer: streamer, logger: logger, + port: cfg.Port, } s.setupRoutes() @@ -52,11 +55,6 @@ func New(db database.Store, streamer *streamer.Streamer, logger *logger.Logger) // Start starts the HTTP server func (s *Server) Start() error { - port := os.Getenv("PORT") - if port == "" { - port = "8080" - } - const ( readHeaderTimeout = 40 * time.Second readTimeout = 60 * time.Second @@ -65,7 +63,7 @@ func (s *Server) Start() error { ) s.srv = &http.Server{ - Addr: ":" + port, + Addr: ":" + strconv.Itoa(s.port), Handler: s.router, ReadHeaderTimeout: readHeaderTimeout, ReadTimeout: readTimeout, @@ -73,7 +71,7 @@ func (s *Server) Start() error { IdleTimeout: idleTimeout, } - s.logger.Info("Starting HTTP server", "port", port, "addr", s.srv.Addr) + s.logger.Info("Starting HTTP server", "port", s.port, "addr", s.srv.Addr) // Start in goroutine but log when actually listening go func() { diff --git a/internal/streamer/streamer.go b/internal/streamer/streamer.go index d345a9a..b8b75a7 100644 --- a/internal/streamer/streamer.go +++ b/internal/streamer/streamer.go @@ -106,6 +106,7 @@ type handlerInfo struct { type Streamer struct { logger *logger.Logger client *http.Client + url string handlers []*handlerInfo rawHandler RawMessageHandler mu sync.RWMutex @@ -124,6 +125,7 @@ type Streamer struct { func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer { return &Streamer{ logger: logger, + url: risLiveURL, client: &http.Client{ Timeout: 0, // No timeout for streaming Transport: &http.Transport{ @@ -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 diff --git a/internal/streamer/streamer_test.go b/internal/streamer/streamer_test.go index 1c788a4..5d9582b 100644 --- a/internal/streamer/streamer_test.go +++ b/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 +} diff --git a/script/test b/script/test index 87c1a57..bffb59e 100755 --- a/script/test +++ b/script/test @@ -7,9 +7,9 @@ ROOT="$(cd "$(dirname "$0")/.." && pwd -P)" main() { cd "$ROOT" - go test -timeout 30s -race -cover ./... || { + go test -short -timeout 30s -race -cover ./... || { echo "--- Rerunning with -v for details ---" - go test -timeout 30s -race -v ./... + go test -short -timeout 30s -race -v ./... exit 1 } }