11 Commits
Author SHA1 Message Date
clawbot 71ee2aeda7 Log the start-up checkpoint's figures under their own names (closes #117)
check / check (push) Waiting to run
Initialize scanned the row of PRAGMA wal_checkpoint(TRUNCATE), which is
the busy flag, the frames in the WAL and the frames copied, into
wal_pages, checkpointed and moved; and a completed TRUNCATE reads 0 for
both counts. The line now logs busy, wal_frames and checkpointed_frames,
plus wal_bytes, the WAL file's size, which New reads before it opens the
database because the open writes a frame to the WAL. The line is a
warning when busy is 1. Tests open a database a second time while the
first stays open, with and without a read open on the first, and check
the line's level and figures.

Model: opus-5-5
2026-10-20 04:32:08 +02:00
clawbot 8d16bf9ea6 Add the Rationale section to README (closes #121)
check / check (push) Canceled after 0s
REPO_POLICIES.md requires a Rationale section saying why the program
exists. RIS Live sends only the changes to the routing table, so it
cannot answer which AS an address is routed to now; routewatch applies
the changes to its own tables and answers from them. The section says
those tables hold only the routes announced since routewatch first
started, and miss the updates it drops under load or while it is
stopped.

Model: opus-5-5
2026-10-20 04:24:41 +02:00
clawbot 651a9e22f6 Treat a path with AS 0 anywhere as a withdrawal (closes #141)
check / check (push) Canceled after 0s
An AS path that holds AS 0 anywhere, a member of an AS set included, now
decodes as an empty path. RFC 7607 makes such a path malformed and its
announcements withdrawals. An empty path already gets that handling:
the prefix handler deletes the peer's route for the prefix, and the AS
and peering handlers record nothing from it. Before, a path such as
[64496, 0, 64497] stored the route with origin AS64497, recorded both
ASes, and recorded the peerings between the path's other ASes.

Model: opus-5-5
2026-10-20 04:07:41 +02:00
clawbot 9280e2eb2c Keep the database files, logs and binaries out of the build context (closes #120)
check / check (push) Canceled after 0s
The daemon's database is db.sqlite with its -wal and -shm files, which
the .db patterns in .dockerignore and .gitignore did not match, so a
database left in the clone shipped in the image's source archive.
.dockerignore now also excludes the logs and the .exe, .dll, .so and
.dylib files .gitignore already ignores. A test opens a database and
checks that .dockerignore has an any-depth pattern for each of its
files.

Model: opus-5-5
2026-10-20 03:58:56 +02:00
clawbot 100689a4fb Keep prefixes that do not parse out of the prefix tables (closes #114)
check / check (push) Canceled after 0s
UpdatePrefixesBatch now skips a prefix that does not parse, so it never
becomes a row in prefixes_v4 or prefixes_v6 and never counts in the prefix
total in the stats. Before, the prefix handler stored no route for such a
prefix and logged it once per prefix and peer, but still passed it to
UpdatePrefixesBatch, which stored it.

Model: opus-5-5
2026-10-20 03:51:10 +02:00
clawbot e13240cb04 Show the MIT licence in the page footers (closes #119)
check / check (push) Canceled after 0s
version.License, which the home and status page footers show, still
said WTFPL after LICENSE and the README moved to MIT. It now says MIT.
A test renders both pages and expects each footer to name MIT and not
WTFPL.

Model: opus-5-5
2026-10-20 03:46:44 +02:00
clawbot 55ecb22846 Bound the WHOIS fetcher's success and error time lists (closes #118)
check / check (push) Canceled after 0s
recordSuccess and recordFailure appended a time on every fetch, and only
GetStats, run when a stats endpoint is requested, dropped the times older
than statsWindow, so a daemon nobody polled kept every one. Each now drops
those times before it appends. The drop takes them off the front of the
list instead of copying the rest: the times are in the order they were
recorded, and it now runs on every fetch.

The fetcher reads the time through a now field so a test can set the
clock. The test records 100,000 successes and failures a second apart
without asking for the stats and finds one hour's worth of each kept.

Model: opus-5-5
2026-10-20 03:38:51 +02:00
clawbot bae5a14d76 Reset the peering handler's dropped_paths figure after each log line (closes #116)
check / check (push) Canceled after 0s
processPeerings replaced the path map every run but kept the count of
paths dropped while the map was full, so every later "Processed AS
peerings" line reported all drops since start next to that run's own
figures. The count is now set to 0 where the map is replaced, under the
same lock. A test fills the map, has one path dropped, runs the
processing and finds the count back at 0.

Model: opus-5-5
2026-10-20 03:28:06 +02:00
clawbot 467c295f94 Handle an announcement without an origin like a replacement (closes #113)
check / check (push) Canceled after 0s
An announcement whose AS path is empty or ends in AS 0 now deletes the
peer's route for the prefix, as a withdrawal does. Before, it was neither
stored nor applied: the peer's old route stayed live, and within one
batch it cancelled the peer's earlier withdrawal of the prefix.

Model: opus-5-5
2026-10-20 03:25:41 +02:00
clawbot 2f238236bd Read the drop counter atomically and set the WHOIS fetcher before serving (closes #115)
check / check (push) Canceled after 0s
The read loop added to a handler's drop count atomically while the
statistics endpoint and the metrics log read it as a plain field, a data
race whenever messages were dropped. The count is now an atomic.Uint64,
read with Load. A test reads the handler statistics while the read loop
drops messages for a handler that never finishes one.

Run created the WHOIS fetcher and gave it to the HTTP server after the
server had started, while request handlers could read it. The fetcher is
now set before the server starts.

Model: opus-5-5
2026-10-20 03:18:00 +02:00
clawbot f489321e3a Answer HEAD requests and give 405 a plain message (closes #111)
check / check (push) Canceled after 0s
Every route was registered for GET only, so HEAD answered 405, and on
the JSON routes that 405 said "Internal server error". chi's GetHead
now serves HEAD with the GET handler; /ip and /ip/ register HEAD
themselves, as GetHead does not reach them. The /api/v1 and /ip route
groups answer a path they lack with a JSON 404 "Not found", and the
other methods the router knows (POST, PUT, PATCH, DELETE, OPTIONS,
TRACE, CONNECT) with a JSON 405 "Method not allowed" whose Allow header
names GET and HEAD. JSONValidationMiddleware gives a 4xx sent without a
JSON body the status's standard text instead of "Internal server error".

Model: opus-5-5
2026-10-20 03:06:23 +02:00
25 changed files with 913 additions and 76 deletions
+12 -2
View File
@@ -73,7 +73,8 @@
# This repository's host-built artifacts: `make build` writes bin/, `go
# build` in the root the binaries, `make run` log.txt, `make asupdate` its
# .tmp file; test binaries, coverage profiles and source archives.
# .tmp file; logs, test binaries, coverage profiles, source archives, and
# the binaries and libraries a build for another platform writes.
/bin
/routewatch
/asinfo-gen
@@ -82,15 +83,24 @@
/out
/pkg/asinfo/asdata.json
**/*.tar.zst
**/*.log
**/*.test
**/*.out
**/*.tmp
**/*.exe
**/*.dll
**/*.so
**/*.dylib
# Local databases. The image carries a source archive of the whole build
# context, so these would otherwise ship inside it.
# context, so these would otherwise ship inside it. The daemon's database
# is db.sqlite, with its -wal and -shm files while it runs.
**/*.db
**/*.db-journal
**/*.db-wal
**/*.sqlite
**/*.sqlite-wal
**/*.sqlite-shm
# A local Go workspace points at directories outside the build context.
/go.work
+5 -1
View File
@@ -67,10 +67,14 @@ node_modules/
go.work
go.work.sum
# Local databases, temporary files and debug output.
# Local databases, temporary files and debug output. The daemon's
# database is db.sqlite, with its -wal and -shm files while it runs.
*.db
*.db-journal
*.db-wal
*.sqlite
*.sqlite-wal
*.sqlite-shm
*.tmp
log.txt
out
+28 -7
View File
@@ -5,6 +5,22 @@ 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.
## Rationale
RIPE RIS Live streams the BGP updates that RIPE's route collectors receive, as
they happen, from the moment a client connects. It reports changes, not the
routing table they add up to, so it cannot say which AS an IP address is routed
to now or which routes a prefix has. RouteWatch applies the feed's updates to
live routing tables in its own SQLite database and answers IP-to-AS lookups and
prefix pages from them.
The feed sends no table when a client connects, so these tables hold only the
routes the feed has announced since RouteWatch first started: an IP-to-AS lookup
finds nothing for an address whose routes have not been announced since then.
The tables also miss the updates RouteWatch drops when it falls behind under
load and those sent while it is stopped, so a lookup or a prefix page can show a
route the feed has withdrawn.
## Features
- Real-time streaming of BGP updates from RIPE RIS Live
@@ -36,6 +52,8 @@ DEBUG=routewatch ./routewatch
## HTTP Endpoints
Every route answers `HEAD` as it answers `GET`, without the body.
### Web Interface
- `GET /` - Home page (HTML): headline figures, and lookups by AS number and by
@@ -90,8 +108,11 @@ and 4 seconds for the database reads behind `/status.json` and `/api/v1/stats`.
An AS number, prefix or mask length that does not parse answers 400, as does a
mask length out of range for its IP version. An AS the database does not hold,
or a prefix with no live route, answers 404. The HTML pages give these errors as
plain text. A 500 answer says only `Internal server error`, and a host name that
does not resolve only `could not resolve host name`; the detail goes to the log.
plain text. Under `/api/v1` and `/ip/`, a path no route matches answers 404 with
the message `Not found`, and `POST`, `PUT`, `PATCH`, `DELETE`, `OPTIONS`,
`TRACE` or `CONNECT` answers 405 with `Method not allowed`. A 500 answer says
only `Internal server error`, and a host name that does not resolve only
`could not resolve host name`; the detail goes to the log.
Every answer carries `Strict-Transport-Security`, `Content-Security-Policy`,
`X-Frame-Options`, `X-Content-Type-Options`, `Referrer-Policy` and
@@ -190,11 +211,11 @@ routewatch/
- **ASHandler**: Tracks all ASNs seen in AS paths
- **PeerHandler**: Records BGP peer information
- **PrefixHandler**: Maintains live routing table (upserts on announcement,
deletes on withdrawal or on an announcement whose AS path is empty or ends
in AS 0, deletes all of a peer's routes when its BGP session goes down).
Routes are kept per peer IP, so when one peer IP has sessions on two
collectors, a session going down on one also removes the routes the other
still carries, until they are announced again.
deletes on withdrawal or on an announcement whose AS path is empty or
holds AS 0 anywhere, deletes all of a peer's routes when its BGP session
goes down). Routes are kept per peer IP, so when one peer IP has sessions
on two collectors, a session going down on one also removes the routes the
other still carries, until they are announced again.
- **PeeringHandler**: Extracts AS peering relationships from AS paths
5. **HTTP Request Flow**
+54 -13
View File
@@ -169,6 +169,20 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
return nil, fmt.Errorf("failed to create database directory: %w", err)
}
// The size of the WAL file the previous run left, which Initialize logs
// with the start-up checkpoint. It is read before the database is opened,
// because opening it writes a frame to the WAL. There is no WAL file when
// the database is new or the last run closed the database, which deletes
// it.
var walBytes int64
wal, err := os.Stat(dbPath + walFileSuffix)
if err == nil {
walBytes = wal.Size()
} else if !errors.Is(err, os.ErrNotExist) {
logger.Warn("Failed to read the WAL file's size on startup", "error", err)
}
// 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
@@ -229,7 +243,7 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
counts: &liveCounts{},
}
err = database.Initialize()
err = database.Initialize(walBytes)
if err != nil {
return nil, fmt.Errorf("failed to initialize database: %w", err)
}
@@ -245,8 +259,16 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
return database, nil
}
// Initialize creates the database schema if it doesn't exist.
func (d *Database) Initialize() error {
// The messages Initialize logs the start-up WAL checkpoint with.
const (
startupCheckpointMessage = "WAL checkpoint on startup"
startupCheckpointIncompleteMessage = "WAL checkpoint on startup incomplete"
)
// Initialize creates the database schema if it doesn't exist. walBytes is the
// size of the WAL file the previous run left, which New reads before it opens
// the database.
func (d *Database) Initialize(walBytes int64) error {
// Set SQLite pragmas for performance. Per-connection settings (cache_size,
// synchronous, busy_timeout, journal_mode) live in the DSN, and New sets the
// writing connection's cache_size; temp_store is left at its default so
@@ -266,19 +288,33 @@ func (d *Database) Initialize() error {
}
}
// Run WAL checkpoint on startup to consolidate any existing WAL
var walPages, checkpointedPages, movedPages int
// The start-up checkpoint copies the WAL a previous run left into the
// database file and empties the WAL file; its frame counts then read 0,
// so walBytes is logged with them.
var busy, walFrames, checkpointedFrames int
err := d.writer.QueryRowContext(context.Background(),
"PRAGMA wal_checkpoint(TRUNCATE)",
).Scan(&walPages, &checkpointedPages, &movedPages)
if err != nil {
).Scan(&busy, &walFrames, &checkpointedFrames)
switch {
case err != nil:
d.logger.Warn("Failed to checkpoint WAL on startup", "error", err)
} else {
d.logger.Info("WAL checkpoint on startup",
"wal_pages", walPages,
"checkpointed", checkpointedPages,
"moved", movedPages,
case busy != 0:
// Another process using the database kept the checkpoint from
// copying or emptying all of the WAL.
d.logger.Warn(startupCheckpointIncompleteMessage,
"wal_bytes", walBytes,
"busy", busy,
"wal_frames", walFrames,
"checkpointed_frames", checkpointedFrames,
)
default:
d.logger.Info(startupCheckpointMessage,
"wal_bytes", walBytes,
"busy", busy,
"wal_frames", walFrames,
"checkpointed_frames", checkpointedFrames,
)
}
@@ -855,7 +891,7 @@ func insertPrefixesInTable(
// UpdatePrefixesBatch stores, in a single transaction, the prefixes that are
// not stored yet, each first seen at its time in prefixes: the IPv4 prefixes
// first and then the IPv6 prefixes.
// first and then the IPv6 prefixes. A prefix that does not parse is not stored.
func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
if len(prefixes) == 0 {
return nil
@@ -874,6 +910,11 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
prefixesV6 := make(map[string]time.Time)
for prefix, timestamp := range prefixes {
_, err := netip.ParsePrefix(prefix)
if err != nil {
continue
}
if detectIPVersion(prefix) == ipVersionV4 {
prefixesV4[prefix] = timestamp
} else {
+172
View File
@@ -1,8 +1,10 @@
package database
import (
"bytes"
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"log/slog"
@@ -1052,6 +1054,176 @@ func walFileSize(t *testing.T, db *Database) int64 {
return wal.Size()
}
// TestStartupCheckpointLogsItsFigures opens a new database, writes a route,
// and opens the same file again while the first stays open, so that its WAL
// file stays. The first open finds no WAL file: it logs a size of 0 and no
// warning. The second logs the size of the WAL file the first left, and the
// checkpoint's busy flag and frame counts, each under its own name; the
// counts read 0 once the checkpoint has emptied the file.
// Initialize logged the busy flag as wal_pages and the counts under each
// other's names (https://git.eeqj.de/sneak/routewatch/issues/117).
func TestStartupCheckpointLogsItsFigures(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, lines := openLogged(t, cfg)
for _, line := range lines {
if line[slog.LevelKey] == slog.LevelWarn.String() {
t.Errorf("opening a new database logged a warning: %v", line)
}
}
line := findLine(t, lines, startupCheckpointMessage)
if line["wal_bytes"] != 0.0 {
t.Errorf("new database: start-up checkpoint logged wal_bytes = %v, "+
"want 0", line["wal_bytes"])
}
err := db.UpsertLiveRouteBatch([]*LiveRoute{
mkV4Route(t, "192.0.2.0/24", lockTestASN, time.Now().UTC()),
})
if err != nil {
t.Fatalf("failed to store a route: %v", err)
}
walBytes := walFileSize(t, db)
if walBytes == 0 {
t.Fatal("WAL file empty after the route write, want frames in it")
}
_, lines = openLogged(t, cfg)
line = findLine(t, lines, startupCheckpointMessage)
for _, key := range []string{"busy", "wal_frames", "checkpointed_frames"} {
if line[key] != 0.0 {
t.Errorf("start-up checkpoint logged %s = %v, want 0", key, line[key])
}
}
if line["wal_bytes"] != float64(walBytes) {
t.Errorf("start-up checkpoint logged wal_bytes = %v, want the %d "+
"bytes the first database left", line["wal_bytes"], walBytes)
}
}
// TestStartupCheckpointWarnsWhenIncomplete keeps a read open on a database
// while a route is written after the read began, then opens the same file
// again. The second open's start-up checkpoint cannot copy the frames written
// after the read began, and gives up once the busy timeout has passed. It
// checks that the checkpoint is logged as a warning with busy 1 and fewer
// frames copied than the WAL holds.
func TestStartupCheckpointWarnsWhenIncomplete(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, _ := openLogged(t, cfg)
ctx := context.Background()
reader, err := db.db.Conn(ctx)
if err != nil {
t.Fatalf("failed to open the reading connection: %v", err)
}
defer func() { _ = reader.Close() }()
// A plain BEGIN starts a transaction that only reads, unlike the BEGIN
// IMMEDIATE of _txlock. It keeps the snapshot its first read takes until
// it ends.
_, err = reader.ExecContext(ctx, "BEGIN")
if err != nil {
t.Fatalf("failed to begin the read transaction: %v", err)
}
defer func() { _, _ = reader.ExecContext(ctx, "ROLLBACK") }()
var routes int
err = reader.QueryRowContext(ctx,
"SELECT count(*) FROM live_routes_v4").Scan(&routes)
if err != nil {
t.Fatalf("failed to read the live routes: %v", err)
}
err = db.UpsertLiveRouteBatch([]*LiveRoute{
mkV4Route(t, "192.0.2.0/24", lockTestASN, time.Now().UTC()),
})
if err != nil {
t.Fatalf("failed to store a route after the read: %v", err)
}
_, lines := openLogged(t, cfg)
line := findLine(t, lines, startupCheckpointIncompleteMessage)
if line[slog.LevelKey] != slog.LevelWarn.String() {
t.Errorf("start-up checkpoint logged at level %v, want %s",
line[slog.LevelKey], slog.LevelWarn)
}
if line["busy"] != 1.0 {
t.Errorf("start-up checkpoint logged busy = %v, want 1", line["busy"])
}
walFrames, _ := line["wal_frames"].(float64)
checkpointedFrames, _ := line["checkpointed_frames"].(float64)
if checkpointedFrames >= walFrames {
t.Errorf("start-up checkpoint logged checkpointed_frames = %v and "+
"wal_frames = %v with a read open, want fewer copied than the WAL "+
"holds", line["checkpointed_frames"], line["wal_frames"])
}
}
// openLogged opens the database of cfg with a logger that writes JSON, and
// returns it with the lines it logged while it opened, decoded.
func openLogged(t *testing.T, cfg *config.Config) (*Database, []map[string]any) {
t.Helper()
var logged bytes.Buffer
db, err := New(cfg,
&logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logged, nil))})
if err != nil {
t.Fatalf("failed to open the database: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
var lines []map[string]any
decoder := json.NewDecoder(&logged)
for decoder.More() {
var line map[string]any
err := decoder.Decode(&line)
if err != nil {
t.Fatalf("failed to decode the log: %v", err)
}
lines = append(lines, line)
}
return db, lines
}
// findLine returns the first of lines logged with message.
func findLine(
t *testing.T,
lines []map[string]any,
message string,
) map[string]any {
t.Helper()
for _, line := range lines {
if line[slog.MessageKey] == message {
return line
}
}
t.Fatalf("no line %q in the log", message)
return nil
}
// TestGetRandomPrefixesByLengthListsPrefixOnce stores one prefix of each
// family learned from two peers in the same second and checks that the
// prefix-length sample lists each of them once.
@@ -0,0 +1,61 @@
package database
import (
"os"
"path/filepath"
"strings"
"testing"
"git.eeqj.de/sneak/routewatch/internal/config"
"git.eeqj.de/sneak/routewatch/internal/logger"
)
// TestDockerignoreExcludesDatabaseFiles checks that .dockerignore keeps the
// database file, and the WAL and shared-memory files SQLite writes next to it,
// out of the build context wherever they sit in the clone. The image ships the
// whole build context as a source archive.
func TestDockerignoreExcludesDatabaseFiles(t *testing.T) {
t.Parallel()
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
data, err := os.ReadFile("../../.dockerignore")
if err != nil {
t.Fatalf("failed to read .dockerignore: %v", err)
}
// Docker reads `**/` as any number of directories and the rest of the
// pattern as filepath.Match does, so `**/` followed by a pattern that
// matches a file's name excludes that file at any depth.
var anyDepth []string
for line := range strings.Lines(string(data)) {
pattern, ok := strings.CutPrefix(strings.TrimSpace(line), "**/")
if ok {
anyDepth = append(anyDepth, pattern)
}
}
name := filepath.Base(db.path)
for _, file := range []string{name, name + "-wal", name + "-shm"} {
excluded := false
for _, pattern := range anyDepth {
matched, err := filepath.Match(pattern, file)
if err != nil {
t.Fatalf("bad .dockerignore pattern **/%s: %v", pattern, err)
}
excluded = excluded || matched
}
if !excluded {
t.Errorf(".dockerignore has no **/ pattern that excludes %s", file)
}
}
}
+27 -9
View File
@@ -4,6 +4,7 @@ package ristypes
import (
"encoding/json"
"slices"
"time"
)
@@ -24,7 +25,8 @@ const (
)
// ASPath is a BGP AS path as the feed writes it: an array of AS numbers in
// which an AS set is an array of its members, such as [1, 2, [3, 4]].
// which an AS set is an array of its members, such as [1, 2, [3, 4]]. A path
// that holds AS 0 decodes as an empty one (UnmarshalJSON).
type ASPath struct {
// Sequence is the path's AS numbers in order. A set of one member that
// ends the path counts as that AS and is its last element. The members of
@@ -48,23 +50,41 @@ func (p *ASPath) Origin() int {
return p.Sequence[len(p.Sequence)-1]
}
// UnmarshalJSON implements the json.Unmarshaler interface for ASPath.
// UnmarshalJSON implements the json.Unmarshaler interface for ASPath. A path
// that holds AS 0 anywhere, a member of an AS set included, decodes as an empty
// path: RFC 7607 makes such a path malformed and its announcements
// withdrawals. On an announcement with an empty path the prefix handler
// deletes the peer's route, and the AS and peering handlers record nothing.
func (p *ASPath) UnmarshalJSON(data []byte) error {
path, err := decodeASPath(data)
if err != nil {
return err
}
if slices.Contains(path.Sequence, 0) || slices.Contains(path.Set, 0) {
path = ASPath{}
}
*p = path
return nil
}
// decodeASPath decodes an AS path as the feed writes it.
func decodeASPath(data []byte) (ASPath, error) {
// Most paths have no AS set and decode as a plain array of integers.
var sequence []int
err := json.Unmarshal(data, &sequence)
if err == nil {
*p = ASPath{Sequence: sequence}
return nil
return ASPath{Sequence: sequence}, nil
}
var raw []any
err = json.Unmarshal(data, &raw)
if err != nil {
return err
return ASPath{}, err
}
path := ASPath{Sequence: make([]int, 0, len(raw))}
@@ -90,9 +110,7 @@ func (p *ASPath) UnmarshalJSON(data []byte) error {
}
}
*p = path
return nil
return path, nil
}
// RISLiveMessage represents the outer wrapper message from the RIPE RIS Live stream.
+30
View File
@@ -145,3 +145,33 @@ func TestASPathKeepsATrailingSetApart(t *testing.T) {
})
}
}
// TestASPathWithASZeroDecodesAsAnEmptyPath decodes AS paths that hold AS 0
// before their end, at their end, or in an AS set. RFC 7607 makes such a path
// malformed, so each decodes as an empty path.
func TestASPathWithASZeroDecodesAsAnEmptyPath(t *testing.T) {
t.Parallel()
for _, path := range []string{
"[64496, 0, 64497]",
"[64496, 0]",
"[64496, [0, 64497], 64498]",
"[64496, [64497, 0]]",
} {
t.Run(path, func(t *testing.T) {
t.Parallel()
var got ristypes.ASPath
err := json.Unmarshal([]byte(path), &got)
if err != nil {
t.Fatalf("unmarshal %s: %v", path, err)
}
if len(got.Sequence) > 0 || len(got.Set) > 0 {
t.Errorf("%s decoded to sequence %v and set %v, want an empty path",
path, got.Sequence, got.Set)
}
})
}
}
+6 -5
View File
@@ -116,17 +116,18 @@ func (rw *RouteWatch) Run(ctx context.Context) error {
return err
}
// Start ASN WHOIS fetcher for background updates. The HTTP handlers read
// the server's fetcher without a lock, so it is set before they run.
rw.asnFetcher = NewASNFetcher(rw.db, rw.logger.Logger)
rw.asnFetcher.Start(ctx)
rw.server.SetASNFetcher(rw.asnFetcher)
// Start HTTP server
err = rw.server.Start(ctx)
if err != nil {
return err
}
// Start ASN WHOIS fetcher for background updates
rw.asnFetcher = NewASNFetcher(rw.db, rw.logger.Logger)
rw.asnFetcher.Start(ctx)
rw.server.SetASNFetcher(rw.asnFetcher)
// Start database maintenance goroutine
rw.dbMaintainer = NewDBMaintainer(rw.db, rw.logger.Logger)
rw.dbMaintainer.Start(ctx)
@@ -1,6 +1,8 @@
package routewatch
import (
"maps"
"slices"
"testing"
"git.eeqj.de/sneak/routewatch/internal/logger"
@@ -42,3 +44,25 @@ func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
asn.LastSeen, second.ParsedTimestamp)
}
}
// TestASHandlerRecordsNothingFromAPathWithASZero handles an update whose AS
// path holds AS 0 between two other ASes. RFC 7607 makes such a path
// malformed, so neither AS is recorded.
func TestASHandlerRecordsNothingFromAPathWithASZero(t *testing.T) {
t.Parallel()
store := newMockStore()
// Built without NewASHandler's flush goroutine, so the test flushes itself.
h := &ASHandler{db: store, logger: logger.New()}
msg := peerAnnouncement(testPeerIP, testV4Prefix)
msg.Path = decodePath(t, testASNA, 0, testASNB)
h.HandleMessage(msg)
h.flushBatch()
if len(store.ASNs) > 0 {
t.Errorf("ASes recorded = %v, want none",
slices.Sorted(maps.Keys(store.ASNs)))
}
}
+18 -13
View File
@@ -53,7 +53,8 @@ type ASNFetcher struct {
currentInterval time.Duration
consecutiveFails int
// hourly stats tracking
// hourly stats tracking; now is the clock, time.Now except in tests
now func() time.Time
statsMu sync.Mutex
successTimes []time.Time
errorTimes []time.Time
@@ -69,6 +70,7 @@ func NewASNFetcher(db database.Store, logger *slog.Logger) *ASNFetcher {
stopCh: make(chan struct{}),
done: make(chan struct{}),
currentInterval: baseInterval,
now: time.Now,
successTimes: make([]time.Time, 0),
errorTimes: make([]time.Time, 0),
}
@@ -121,7 +123,7 @@ func (f *ASNFetcher) GetStats() server.ASNFetcherStats {
f.intervalMu.Unlock()
// Prune old entries and count
cutoff := time.Now().Add(-statsWindow)
cutoff := f.now().Add(-statsWindow)
f.successTimes = pruneOldTimes(f.successTimes, cutoff)
f.errorTimes = pruneOldTimes(f.errorTimes, cutoff)
@@ -133,16 +135,15 @@ func (f *ASNFetcher) GetStats() server.ASNFetcherStats {
}
}
// pruneOldTimes removes times older than cutoff and returns the pruned slice.
// pruneOldTimes drops the times at or before cutoff from the front of times,
// which holds times in the order they were recorded, and returns the rest. It
// copies nothing, so it is cheap enough to run on every append.
func pruneOldTimes(times []time.Time, cutoff time.Time) []time.Time {
result := make([]time.Time, 0, len(times))
for _, t := range times {
if t.After(cutoff) {
result = append(result, t)
}
for len(times) > 0 && !times[0].After(cutoff) {
times = times[1:]
}
return result
return times
}
// getInterval returns the current fetch interval.
@@ -170,9 +171,11 @@ func (f *ASNFetcher) recordSuccess() {
}
f.intervalMu.Unlock()
// Record success time for stats
// Record success time for stats, dropping those older than statsWindow
f.statsMu.Lock()
f.successTimes = append(f.successTimes, time.Now())
now := f.now()
f.successTimes = pruneOldTimes(f.successTimes, now.Add(-statsWindow))
f.successTimes = append(f.successTimes, now)
f.statsMu.Unlock()
}
@@ -194,9 +197,11 @@ func (f *ASNFetcher) recordFailure() {
}
f.intervalMu.Unlock()
// Record error time for stats
// Record error time for stats, dropping those older than statsWindow
f.statsMu.Lock()
f.errorTimes = append(f.errorTimes, time.Now())
now := f.now()
f.errorTimes = pruneOldTimes(f.errorTimes, now.Add(-statsWindow))
f.errorTimes = append(f.errorTimes, now)
f.statsMu.Unlock()
}
@@ -203,3 +203,35 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
t.Fatal("the fetcher goroutine did not return after the lookup ended")
}
}
// TestRecordKeepsOnlyTheStatsWindow records a success and a failure every
// second, by a clock the test sets, far longer than statsWindow, and never
// asks for the stats. The fetcher must keep only the times within the last
// statsWindow.
func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) {
t.Parallel()
const (
records = 100_000
step = time.Second
want = int(statsWindow / step)
)
clock := time.Now()
f := NewASNFetcher(newMockStore(), logger.New().Logger)
f.now = func() time.Time { return clock }
for range records {
clock = clock.Add(step)
f.recordSuccess()
f.recordFailure()
}
if len(f.successTimes) != want || len(f.errorTimes) != want {
t.Errorf("%d success and %d error times kept, want %d of each, "+
"those of the last %s", len(f.successTimes), len(f.errorTimes),
want, statsWindow)
}
}
+4 -2
View File
@@ -46,7 +46,7 @@ type PeeringHandler struct {
// and the latest time an update carried it.
mu sync.Mutex
asPaths map[string]database.SeenTimes
droppedPaths int // paths dropped because the map was full
droppedPaths int // paths dropped since the last run because the map was full
// writing is how many paths processPeerings took from asPaths and has
// not finished writing the peerings of.
writing int
@@ -190,12 +190,14 @@ func (h *PeeringHandler) processLoop() {
func (h *PeeringHandler) processPeerings() {
// 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.
// released, so the map never grows past a single interval's traffic. The
// count of dropped paths starts again with the new map.
h.mu.Lock()
paths := h.asPaths
h.asPaths = make(map[string]database.SeenTimes)
h.writing = len(paths)
dropped := h.droppedPaths
h.droppedPaths = 0
h.mu.Unlock()
if len(paths) == 0 {
@@ -153,6 +153,43 @@ func TestPeeringHandlerLeavesATrailingASSetOutOfThePeerings(t *testing.T) {
}
}
// TestPeeringHandlerRecordsNothingFromAPathWithASZero handles AS paths that
// hold AS 0. RFC 7607 makes such a path malformed, so no peering is recorded
// from it, not even one between two other ASes of the path.
func TestPeeringHandlerRecordsNothingFromAPathWithASZero(t *testing.T) {
t.Parallel()
tests := []struct {
name string
path []any
}{
{name: "AS 0 between two ASes", path: []any{testASNA, 0, testASNB}},
{
name: "AS 0 after two ASes",
path: []any{testASNA, testASNB, 0, testASNC},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
store := newMockStore()
h := newTestHandler(store)
h.HandleMessage(&ristypes.RISMessage{
Path: decodePath(t, tt.path...),
ParsedTimestamp: time.Now().UTC(),
})
h.ProcessPeeringsNow()
if len(store.Peerings) > 0 {
t.Errorf("peerings recorded = %v, want none",
slices.Sorted(maps.Keys(store.Peerings)))
}
})
}
}
// 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.
@@ -264,6 +301,36 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
}
}
// TestPeeringHandlerRunResetsTheDroppedCount fills the map, has the handler
// drop one path, and runs the processing. The count of dropped paths is the
// run's own, so it is 0 again after the run.
func TestPeeringHandlerRunResetsTheDroppedCount(t *testing.T) {
t.Parallel()
h := newTestHandler(newMockStore())
now := time.Now().UTC()
// Paths of one AS carry no peering, so the run records nothing.
h.mu.Lock()
for asn := 1; len(h.asPaths) < maxTrackedPaths; asn++ {
h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now}
}
h.mu.Unlock()
handle(h, now, testASNA, testASNB)
if _, dropped := snapshot(h); dropped != 1 {
t.Fatalf("want dropped count 1 with the map full, got %d", dropped)
}
h.ProcessPeeringsNow()
if _, dropped := snapshot(h); dropped != 0 {
t.Errorf("want dropped count 0 after a run, got %d", dropped)
}
}
// TestPeeringHandlerStoresTheEarliestAndLatestTimesOfANewPeering handles two
// updates whose path carries a peering the database does not hold yet, the
// second 5 seconds after the first, in one run into a real database. The
+7 -7
View File
@@ -45,8 +45,8 @@ const (
// PrefixHandler tracks BGP prefixes and maintains a live routing table in the
// database. Routes are added on announcement and deleted on withdrawal or on an
// announcement whose AS path is empty or ends in AS 0, and all routes of a peer
// are deleted when its BGP session goes down.
// announcement whose AS path is empty or holds AS 0 anywhere, and all routes of
// a peer are deleted when its BGP session goes down.
type PrefixHandler struct {
db database.Store
logger *logger.Logger
@@ -373,10 +373,10 @@ func (h *PrefixHandler) flushBatch() {
// An announcement is stored when its path gives an origin, or when it
// ends in an AS set of several members: that route is stored with
// origin 0 (none). Any other update deletes the peer's route for the
// prefix: a withdrawal, or an announcement whose path is empty or ends
// in AS 0. Such an announcement replaces the peer's route, as any
// announcement does, with none (RFC 7607 treats AS 0 in a path as a
// withdrawal).
// prefix: a withdrawal, or an announcement whose path is empty or
// holds AS 0 anywhere (the decoder empties such a path). Such an
// announcement replaces the peer's route, as any announcement does,
// with none (RFC 7607 treats AS 0 in a path as a withdrawal).
if update.messageType == "announcement" &&
(update.originASN > 0 || update.endsInSet) {
// Create live route for batch upsert
@@ -388,7 +388,7 @@ func (h *PrefixHandler) flushBatch() {
// Parse CIDR to get IP version
_, ipVersion, err := parseCIDR(update.prefix)
if err != nil {
h.logger.Error("Failed to parse CIDR for withdrawal",
h.logger.Error("Failed to parse CIDR for route deletion",
"prefix", update.prefix, "error", err)
continue
@@ -381,11 +381,12 @@ func TestPrefixHandlerDoesNotStoreAnAnnouncementWithAnEmptyPath(t *testing.T) {
// TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin runs, through
// the prefix handler into a real database, a peer's announcement of a prefix
// with origin AS testASNB and then the peer's announcement of the prefix with
// an AS path that gives no origin: one whose last AS is 0, or an empty one. As
// any later announcement from the peer does, the second replaces the peer's
// route, here with none, whether it comes in a later batch or follows a
// withdrawal in one batch. The in-memory route count and prefix distribution
// must match the route table.
// an AS path that gives no origin: one whose last AS is 0, an empty one, or one
// that holds AS 0 before its end, which decodes as an empty path. As any later
// announcement from the peer does, the second replaces the peer's route, here
// with none, whether it comes in a later batch or follows a withdrawal in one
// batch. The in-memory route count and prefix distribution must match the
// route table.
func TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin(
t *testing.T,
) {
@@ -421,6 +422,12 @@ func TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin(
name: "an empty path",
batch: []*ristypes.RISMessage{noOrigin(ristypes.ASPath{})},
},
{
name: "a path with AS 0 before its end",
batch: []*ristypes.RISMessage{
noOrigin(decodePath(t, testASNA, 0, testASNB)),
},
},
{
name: "a withdrawal and then a path whose last AS is 0",
batch: []*ristypes.RISMessage{withdrawal, noOrigin(endsInZero)},
@@ -440,6 +447,64 @@ func TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin(
}
}
// TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse runs an announcement of
// "not-a-prefix" and a withdrawal of "also/bad" through the prefix handler into
// a real database. Neither is stored in the prefix tables: the prefix count in
// the stats stays 0, both as kept in memory and as counted from the tables when
// the database is opened again.
func TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse(t *testing.T) {
t.Parallel()
cfg := &config.Config{StateDir: t.TempDir()}
db, err := database.New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
// Built without NewPrefixHandler's flush timer, so the test flushes itself.
h := &PrefixHandler{db: db, logger: logger.New()}
withdrawal := &ristypes.RISMessage{
Type: ristypes.MessageTypeUpdate,
Peer: testPeerIP,
ParsedTimestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
Withdrawals: []string{"also/bad"},
}
h.HandleMessage(peerAnnouncement(testPeerIP, "not-a-prefix"))
h.HandleMessage(withdrawal)
h.flushBatch()
checkNoPrefixes := func(when string) {
t.Helper()
stats, err := db.GetStatsContext(context.Background())
if err != nil {
t.Fatalf("GetStatsContext %s: %v", when, err)
}
if stats.Prefixes != 0 {
t.Errorf("prefix count %s = %d, want 0", when, stats.Prefixes)
}
}
checkNoPrefixes("after the flush")
err = db.Close()
if err != nil {
t.Fatalf("Close: %v", err)
}
db, err = database.New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to open the database again: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
checkNoPrefixes("after opening the database again")
}
// TestStateMessagesGoToThePrefixHandlerNotThePeerHandler checks that STATE
// messages go to the prefix handler, which acts on a down, and not to the peer
// handler, which would record a peer whose session never comes up as active.
+9 -3
View File
@@ -142,8 +142,9 @@ func deadlinePassed(ctx context.Context) bool {
//
// If a response is not valid JSON or is empty when JSON is expected, the middleware
// returns a properly formatted JSON error response. If the request's time limit
// has passed, it answers 504 with the error message "Request timeout". For other
// errors, it returns "Internal server error" with status 500 if the original
// has passed, it answers 504 with the error message "Request timeout". A 4xx
// keeps its status, with the status's standard text as the message. For other
// errors, it returns "Internal server error", with status 500 if the original
// status was 200.
//
// Non-JSON responses are passed through unchanged.
@@ -203,8 +204,13 @@ func JSONValidationMiddleware(next http.Handler) http.Handler {
w.WriteHeader(statusCode)
errorMsg := internalErrorMessage
if statusCode == http.StatusGatewayTimeout {
switch {
case statusCode == http.StatusGatewayTimeout:
errorMsg = requestTimeoutMessage
case statusCode >= http.StatusBadRequest &&
statusCode < http.StatusInternalServerError:
errorMsg = http.StatusText(statusCode)
}
response := map[string]any{
@@ -58,6 +58,34 @@ func TestResponsesCarrySecurityHeaders(t *testing.T) {
}
}
// TestJSONValidationKeeps4xxWithoutBody wraps a handler that answers 400
// without a body in JSONValidationMiddleware and checks that the JSON error
// body keeps the 400 and gives the status's own text. Before the fix it said
// "Internal server error"
// (https://git.eeqj.de/sneak/routewatch/issues/111).
func TestJSONValidationKeeps4xxWithoutBody(t *testing.T) {
t.Parallel()
handler := JSONValidationMiddleware(http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusBadRequest)
}))
rec := httptest.NewRecorder()
handler.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodGet, "/", nil))
const name = "400 without a body"
if rec.Code != http.StatusBadRequest {
t.Errorf("%s: status %d, want %d; body %s",
name, rec.Code, http.StatusBadRequest, rec.Body)
}
checkJSONErrorCode(t, name, rec, http.StatusBadRequest)
checkJSONErrorMessage(t, name, rec, http.StatusText(http.StatusBadRequest))
}
// TestPagesCarryTheCSPNonce checks that every script and style block of each
// page carries the nonce its Content-Security-Policy header names, so that the
// browser runs and applies them, that each answer has a nonce of its own, and
+34 -1
View File
@@ -1,18 +1,33 @@
package server
import (
"net/http"
"time"
"github.com/go-chi/chi/v5"
"github.com/go-chi/chi/v5/middleware"
)
// The error messages the JSON routes under /api/v1 and /ip answer with when no
// route matches the path, and when none takes the method.
const (
notFoundMessage = "Not found"
methodNotAllowedMessage = "Method not allowed"
// jsonRoutesAllow is the Allow header of their 405 answer: every one of
// them takes GET and HEAD, and no other method.
jsonRoutesAllow = "GET, HEAD"
)
// setupRoutes configures the HTTP routes
func (s *Server) setupRoutes() {
r := chi.NewRouter()
// Middleware
r.Use(DecodedPathMiddleware)
// GetHead serves HEAD with the GET handler. It looks the route up by the
// path DecodedPathMiddleware sets, so it comes after it.
r.Use(middleware.GetHead)
r.Use(SecurityHeadersMiddleware)
r.Use(middleware.RequestID)
r.Use(RealIPMiddleware(s.trustedProxies))
@@ -40,16 +55,34 @@ func (s *Server) setupRoutes() {
r.Get("/prefixlength/{length}", s.handlePrefixLength())
r.Get("/prefixlength6/{length}", s.handlePrefixLength6())
// The JSON routes answer a path or method they do not take with the JSON
// error body, as they answer everything else.
jsonNotFound := func(w http.ResponseWriter, _ *http.Request) {
s.writeJSONError(w, http.StatusNotFound, notFoundMessage)
}
jsonMethodNotAllowed := func(w http.ResponseWriter, _ *http.Request) {
// chi's own 405 answer sets Allow; one set in its place must too.
w.Header().Set("Allow", jsonRoutesAllow)
s.writeJSONError(w, http.StatusMethodNotAllowed, methodNotAllowedMessage)
}
// IP info JSON endpoints (replaces old /ip redirect)
r.Route("/ip", func(r chi.Router) {
r.Use(JSONValidationMiddleware)
r.Get("/", s.handleIPInfo()) // Client IP
r.NotFound(jsonNotFound)
r.MethodNotAllowed(jsonMethodNotAllowed)
r.Get("/", s.handleIPInfo()) // Client IP
// GetHead does not reach this route: mounting the group registers
// /ip and /ip/ for every method, so HEAD arrives here as HEAD.
r.Head("/", s.handleIPInfo())
r.Get("/{addr}", s.handleIPInfo()) // Specified IP
})
// API routes
r.Route("/api/v1", func(r chi.Router) {
r.Use(JSONValidationMiddleware)
r.NotFound(jsonNotFound)
r.MethodNotAllowed(jsonMethodNotAllowed)
r.Get("/stats", s.handleStats())
r.Get("/ip/{ip}", s.handleIPLookup())
r.Get("/as/{asn}", s.handleASDetailJSON())
+100
View File
@@ -0,0 +1,100 @@
package server
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
)
// TestRoutesAnswerHEAD sends HEAD for a page and JSON routes through the
// server's router and checks that each answers as GET does. /ip and /ip/, the
// lookup of the caller's own address, are in the list because chi's GetHead
// does not reach them. Before the fix every route answered HEAD with 405
// (https://git.eeqj.de/sneak/routewatch/issues/111).
func TestRoutesAnswerHEAD(t *testing.T) {
t.Parallel()
s := newServerWithTestRoutes(t)
for _, path := range []string{"/", statsPath, "/ip", "/ip/"} {
rec := httptest.NewRecorder()
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodHead, path, nil))
if rec.Code != http.StatusOK {
t.Errorf("HEAD %s: status %d, want %d; body %s",
path, rec.Code, http.StatusOK, rec.Body)
}
}
}
// TestJSONRoutesAnswerUnmatchedRequestsWithJSONErrors sends a method the JSON
// routes do not take, and a path none of them matches, to each group of JSON
// routes through the server's router. Each must answer with the JSON error
// body, and a 405 must name the methods the routes take in its Allow header.
// Before the fix a 405 said "Internal server error" and a 404 was chi's plain
// text answer (https://git.eeqj.de/sneak/routewatch/issues/111).
func TestJSONRoutesAnswerUnmatchedRequestsWithJSONErrors(t *testing.T) {
t.Parallel()
s := newServerWithTestRoutes(t)
tests := []struct {
method string
path string
wantCode int
wantMsg string
}{
{http.MethodPost, statsPath, http.StatusMethodNotAllowed, methodNotAllowedMessage},
{http.MethodPost, "/ip/", http.StatusMethodNotAllowed, methodNotAllowedMessage},
{http.MethodGet, "/api/v1/nothing", http.StatusNotFound, notFoundMessage},
{http.MethodGet, "/ip/192.0.2.1/nothing", http.StatusNotFound, notFoundMessage},
}
for _, tt := range tests {
name := tt.method + " " + tt.path
rec := httptest.NewRecorder()
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), tt.method, tt.path, nil))
if rec.Code != tt.wantCode {
t.Errorf("%s: status %d, want %d; body %s",
name, rec.Code, tt.wantCode, rec.Body)
}
checkJSONErrorCode(t, name, rec, tt.wantCode)
checkJSONErrorMessage(t, name, rec, tt.wantMsg)
allow := rec.Header().Get("Allow")
if tt.wantCode == http.StatusMethodNotAllowed && allow != jsonRoutesAllow {
t.Errorf("%s: Allow %q, want %q", name, allow, jsonRoutesAllow)
}
}
}
// checkJSONErrorMessage checks that rec holds the JSON error body with the
// given message.
func checkJSONErrorMessage(
t *testing.T, name string, rec *httptest.ResponseRecorder, want string,
) {
t.Helper()
var body struct {
Error struct {
Msg string `json:"msg"`
} `json:"error"`
}
err := json.Unmarshal(rec.Body.Bytes(), &body)
if err != nil {
t.Errorf("%s: decoding the answer %q: %v", name, rec.Body, err)
return
}
if body.Error.Msg != want {
t.Errorf("%s: error message %q, want %q", name, body.Error.Msg, want)
}
}
+2 -1
View File
@@ -139,7 +139,8 @@ func (s *Server) Stop(ctx context.Context) error {
return nil
}
// SetASNFetcher sets the ASN WHOIS fetcher for on-demand lookups.
// SetASNFetcher sets the ASN WHOIS fetcher for on-demand lookups. Call it
// before Start: the handlers read the fetcher without a lock.
func (s *Server) SetASNFetcher(fetcher ASNFetcher) {
s.asnFetcher = fetcher
}
+5 -5
View File
@@ -106,12 +106,12 @@ type RawMessageHandler func(line string)
// handlerMetrics tracks performance metrics for a handler
type handlerMetrics struct {
processedCount uint64 // Total messages processed
droppedCount uint64 // Total messages dropped
droppedCount atomic.Uint64 // Total messages dropped; not under mu
totalTime time.Duration // Total processing time (for average calculation)
minTime time.Duration // Minimum processing time
maxTime time.Duration // Maximum processing time
queueHighWaterMark int // Maximum queue length seen
mu sync.Mutex // Protects the metrics
mu sync.Mutex // Protects the other metrics
}
// handlerInfo wraps a handler with its queue and metrics
@@ -371,7 +371,7 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
QueueCapacity: cap(info.queue),
QueueHighWaterMark: info.metrics.queueHighWaterMark,
ProcessedCount: info.metrics.processedCount,
DroppedCount: info.metrics.droppedCount,
DroppedCount: info.metrics.droppedCount.Load(),
MinProcessTime: info.metrics.minTime,
MaxProcessTime: info.metrics.maxTime,
}
@@ -488,7 +488,7 @@ func (s *Streamer) logMetrics() {
"queue_len", len(info.queue),
"queue_cap", cap(info.queue),
"processed", info.metrics.processedCount,
"dropped", info.metrics.droppedCount,
"dropped", info.metrics.droppedCount.Load(),
"avg_time", avgTime,
"min_time", info.metrics.minTime,
"max_time", info.metrics.maxTime,
@@ -858,7 +858,7 @@ func (s *Streamer) dispatchMessage(
}
if !s.queueMessage(info, msg) {
atomic.AddUint64(&info.metrics.droppedCount, 1)
info.metrics.droppedCount.Add(1)
s.totalDropped.Add(1)
continue
@@ -621,6 +621,68 @@ func TestStreamReadsLongLine(t *testing.T) {
}
}
// blockedHandler wants UPDATE messages, has room on its queue for one, and
// does not return from HandleMessage until release is closed.
type blockedHandler struct {
release chan struct{}
}
func (blockedHandler) WantsMessage(messageType string) bool {
return messageType == ristypes.MessageTypeUpdate
}
func (blockedHandler) QueueCapacity() int { return 1 }
func (h blockedHandler) HandleMessage(*ristypes.RISMessage) { <-h.release }
// TestHandlerStatsReadWhileMessagesAreDropped has the server send a few UPDATE
// lines to a blockedHandler, so the read loop drops all but the one the
// handler holds and the one on its queue, while the test reads the handler's
// statistics. The drop count must reach the number dropped, and the race
// detector must find the count read and written safely.
func TestHandlerStatsReadWhileMessagesAreDropped(t *testing.T) {
t.Parallel()
const (
lines = 5
wantDropped = lines - 2
maxWait = 5 * time.Second
)
srv := httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, r *http.Request) {
_, _ = io.WriteString(w, strings.Repeat(updateLine, lines))
flush(w)
<-r.Context().Done()
}))
defer srv.Close()
s := New(logger.New(), metrics.New())
s.url = srv.URL
handler := blockedHandler{release: make(chan struct{})}
s.RegisterHandler(handler)
err := s.Start(t.Context())
if err != nil {
t.Fatalf("Start returned error: %v", err)
}
defer s.Stop(t.Context())
defer close(handler.release)
// The test polls instead of waiting for a signal from the read loop: a
// signal would order the drops before the reads here, and the race
// detector would have nothing to report.
deadline := time.Now().Add(maxWait)
for s.GetHandlerStats()[0].DroppedCount < wantDropped {
if time.Now().After(deadline) {
t.Fatalf("fewer than %d messages counted as dropped within %s",
wantDropped, maxWait)
}
}
}
// TestPeerCountCountsEveryPeerHeardFrom has the server send an UPDATE from
// stateDownPeer and a KEEPALIVE from another peer, with no OPEN from either,
// and then a message that ends stateDownPeer's session: a NOTIFICATION, or a
+54
View File
@@ -0,0 +1,54 @@
package templates_test
import (
"bytes"
"html/template"
"strings"
"testing"
"git.eeqj.de/sneak/routewatch/internal/templates"
)
// TestFooterLicense renders the home and status pages and expects each footer
// to name the licence in LICENSE, MIT, and not WTFPL.
func TestFooterLicense(t *testing.T) {
t.Parallel()
const (
license = "MIT"
wrongLicense = "WTFPL"
)
pages := map[string]*template.Template{
"index": templates.IndexTemplate(),
"status": templates.StatusTemplate(),
}
for name, tmpl := range pages {
t.Run(name, func(t *testing.T) {
t.Parallel()
var page bytes.Buffer
err := tmpl.Execute(&page, struct{ CSPNonce string }{})
if err != nil {
t.Fatalf("rendering the page: %v", err)
}
_, footer, found := strings.Cut(page.String(), "<footer")
if !found {
t.Fatal("the page has no footer")
}
footer, _, _ = strings.Cut(footer, "</footer>")
if !strings.Contains(footer, license) {
t.Errorf("footer does not name %s: %q", license, footer)
}
if strings.Contains(footer, wrongLicense) {
t.Errorf("footer names %s: %q", wrongLicense, footer)
}
})
}
}
+2 -2
View File
@@ -19,8 +19,8 @@ const (
Author = "@sneak"
// AuthorURL is the author's website
AuthorURL = "https://sneak.berlin"
// License is the program license
License = "WTFPL"
// License is the program license, as LICENSE grants it
License = "MIT"
// RepoURL is the git repository URL
RepoURL = "https://git.eeqj.de/sneak/routewatch"
)