Keep prefixes that do not parse out of the prefix tables (closes #114 )

The prefix handler now drops an announcement or withdrawal whose prefix does not parse, and logs it as an error, so the prefix never reaches the prefix tables or the prefix count in the stats. Before, no route was stored for such a prefix, but it was stored as a permanent row in the prefix tables and counted. Model: opus-5-5
2026-10-20 01:30:51 +00:00
16 changed files with 86 additions and 579 deletions
@@ -73,8 +73,7 @@
 # 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; logs, test binaries, coverage profiles, source archives, and
+# .tmp file; test binaries, coverage profiles and source archives.
 # the binaries and libraries a build for another platform writes.
 /bin
 /routewatch
 /asinfo-gen
@@ -83,24 +82,15 @@
 /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. The daemon's database
+# context, so these would otherwise ship inside it.
 # 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
@@ -67,14 +67,10 @@ node_modules/
 go.work
 go.work.sum
-# Local databases, temporary files and debug output. The daemon's
+# Local databases, temporary files and debug output.
 # 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
@@ -5,22 +5,6 @@ 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
@@ -211,11 +195,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
+      deletes on withdrawal or on an announcement whose AS path is empty or ends
-      holds AS 0 anywhere, deletes all of a peer's routes when its BGP session
+      in AS 0, deletes all of a peer's routes when its BGP session goes down).
-      goes down). Routes are kept per peer IP, so when one peer IP has sessions
+      Routes are kept per peer IP, so when one peer IP has sessions on two
-      on two collectors, a session going down on one also removes the routes the
+      collectors, a session going down on one also removes the routes the other
-      other still carries, until they are announced again.
+      still carries, until they are announced again.
    - **PeeringHandler**: Extracts AS peering relationships from AS paths
 5. **HTTP Request Flow**
@@ -169,20 +169,6 @@ 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
@@ -243,7 +229,7 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 		counts: &liveCounts{},
 	}
-	err = database.Initialize(walBytes)
+	err = database.Initialize()
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize database: %w", err)
 	}
@@ -259,16 +245,8 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
 	return database, nil
 }
-// The messages Initialize logs the start-up WAL checkpoint with.
+// Initialize creates the database schema if it doesn't exist.
-const (
+func (d *Database) Initialize() error {
 	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
@@ -288,33 +266,19 @@ func (d *Database) Initialize(walBytes int64) error {
 		}
 	}
-	// The start-up checkpoint copies the WAL a previous run left into the
+	// Run WAL checkpoint on startup to consolidate any existing WAL
-	// database file and empties the WAL file; its frame counts then read 0,
+	var walPages, checkpointedPages, movedPages int
 	// so walBytes is logged with them.
 	var busy, walFrames, checkpointedFrames int
 	err := d.writer.QueryRowContext(context.Background(),
 		"PRAGMA wal_checkpoint(TRUNCATE)",
-	).Scan(&busy, &walFrames, &checkpointedFrames)
+	).Scan(&walPages, &checkpointedPages, &movedPages)
-
+	if err != nil {
 	switch {
 	case err != nil:
 		d.logger.Warn("Failed to checkpoint WAL on startup", "error", err)
-	case busy != 0:
+	} else {
-		// Another process using the database kept the checkpoint from
+		d.logger.Info("WAL checkpoint on startup",
-		// copying or emptying all of the WAL.
+			"wal_pages", walPages,
-		d.logger.Warn(startupCheckpointIncompleteMessage,
+			"checkpointed", checkpointedPages,
-			"wal_bytes", walBytes,
+			"moved", movedPages,
 			"busy", busy,
 			"wal_frames", walFrames,
 			"checkpointed_frames", checkpointedFrames,
 		)
 	default:
 		d.logger.Info(startupCheckpointMessage,
 			"wal_bytes", walBytes,
 			"busy", busy,
 			"wal_frames", walFrames,
 			"checkpointed_frames", checkpointedFrames,
 		)
 	}
@@ -891,7 +855,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. A prefix that does not parse is not stored.
+// first and then the IPv6 prefixes.
 func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 	if len(prefixes) == 0 {
 		return nil
@@ -910,11 +874,6 @@ 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 {
@@ -1,10 +1,8 @@
 package database
 import (
 	"bytes"
 	"context"
 	"database/sql"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -1054,176 +1052,6 @@ 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.
@@ -1,61 +0,0 @@
 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)
 		}
 	}
 }
@@ -4,7 +4,6 @@ package ristypes
 import (
 	"encoding/json"
 	"slices"
 	"time"
 )
@@ -25,8 +24,7 @@ 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]]. A path
+// which an AS set is an array of its members, such as [1, 2, [3, 4]].
 // 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
@@ -50,41 +48,23 @@ func (p *ASPath) Origin() int {
 	return p.Sequence[len(p.Sequence)-1]
 }
-// UnmarshalJSON implements the json.Unmarshaler interface for ASPath. A path
+// UnmarshalJSON implements the json.Unmarshaler interface for ASPath.
 // 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 {
-		return ASPath{Sequence: sequence}, nil
+		*p = ASPath{Sequence: sequence}
 		return nil
 	}
 	var raw []any
 	err = json.Unmarshal(data, &raw)
 	if err != nil {
-		return ASPath{}, err
+		return err
 	}
 	path := ASPath{Sequence: make([]int, 0, len(raw))}
@@ -110,7 +90,9 @@ func decodeASPath(data []byte) (ASPath, error) {
 		}
 	}
-	return path, nil
+	*p = path
 	return nil
 }
 // RISLiveMessage represents the outer wrapper message from the RIPE RIS Live stream.
@@ -145,33 +145,3 @@ 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)
 			}
 		})
 	}
 }
@@ -1,8 +1,6 @@
 package routewatch
 import (
 	"maps"
 	"slices"
 	"testing"
 	"git.eeqj.de/sneak/routewatch/internal/logger"
@@ -44,25 +42,3 @@ 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)))
 	}
 }
@@ -53,8 +53,7 @@ type ASNFetcher struct {
 	currentInterval  time.Duration
 	consecutiveFails int
-	// hourly stats tracking; now is the clock, time.Now except in tests
+	// hourly stats tracking
 	now          func() time.Time
 	statsMu      sync.Mutex
 	successTimes []time.Time
 	errorTimes   []time.Time
@@ -70,7 +69,6 @@ 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),
 	}
@@ -123,7 +121,7 @@ func (f *ASNFetcher) GetStats() server.ASNFetcherStats {
 	f.intervalMu.Unlock()
 	// Prune old entries and count
-	cutoff := f.now().Add(-statsWindow)
+	cutoff := time.Now().Add(-statsWindow)
 	f.successTimes = pruneOldTimes(f.successTimes, cutoff)
 	f.errorTimes = pruneOldTimes(f.errorTimes, cutoff)
@@ -135,15 +133,16 @@ func (f *ASNFetcher) GetStats() server.ASNFetcherStats {
 	}
 }
-// pruneOldTimes drops the times at or before cutoff from the front of times,
+// pruneOldTimes removes times older than cutoff and returns the pruned slice.
 // 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 {
-	for len(times) > 0 && !times[0].After(cutoff) {
+	result := make([]time.Time, 0, len(times))
-		times = times[1:]
+	for _, t := range times {
 		if t.After(cutoff) {
 			result = append(result, t)
 		}
 	}
-	return times
+	return result
 }
 // getInterval returns the current fetch interval.
@@ -171,11 +170,9 @@ func (f *ASNFetcher) recordSuccess() {
 	}
 	f.intervalMu.Unlock()
-	// Record success time for stats, dropping those older than statsWindow
+	// Record success time for stats
 	f.statsMu.Lock()
-	now := f.now()
+	f.successTimes = append(f.successTimes, time.Now())
 	f.successTimes = pruneOldTimes(f.successTimes, now.Add(-statsWindow))
 	f.successTimes = append(f.successTimes, now)
 	f.statsMu.Unlock()
 }
@@ -197,11 +194,9 @@ func (f *ASNFetcher) recordFailure() {
 	}
 	f.intervalMu.Unlock()
-	// Record error time for stats, dropping those older than statsWindow
+	// Record error time for stats
 	f.statsMu.Lock()
-	now := f.now()
+	f.errorTimes = append(f.errorTimes, time.Now())
 	f.errorTimes = pruneOldTimes(f.errorTimes, now.Add(-statsWindow))
 	f.errorTimes = append(f.errorTimes, now)
 	f.statsMu.Unlock()
 }
@@ -203,35 +203,3 @@ 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)
 	}
 }
@@ -153,43 +153,6 @@ 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.
@@ -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 holds AS 0 anywhere, and all routes of
+// announcement whose AS path is empty or ends in AS 0, and all routes of a peer
-// a peer are deleted when its BGP session goes down.
+// are deleted when its BGP session goes down.
 type PrefixHandler struct {
 	db      database.Store
 	logger  *logger.Logger
@@ -169,10 +169,19 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// each peer's last update for a prefix: a message that lists a prefix as
 	// both withdrawn and announced keeps the route (RFC 4271 section 4.3).
 	// A withdrawal names only the prefix and peer: the message's AS path, and
-	// so its origin AS, belongs to the announced prefixes.
+	// so its origin AS, belongs to the announced prefixes. An update whose
-	for _, prefix := range msg.Withdrawals {
+	// prefix does not parse is logged and dropped.
 	for _, withdrawn := range msg.Withdrawals {
 		prefix, err := canonicalPrefix(withdrawn)
 		if err != nil {
 			h.logger.Error("Dropped a withdrawal whose prefix does not parse",
 				"prefix", withdrawn, "peer", msg.Peer, "error", err)
 			continue
 		}
 		h.batch = append(h.batch, prefixUpdate{
-			prefix:      canonicalPrefix(prefix),
+			prefix:      prefix,
 			peer:        msg.Peer,
 			messageType: "withdrawal",
 			timestamp:   timestamp,
@@ -184,9 +193,18 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 	for _, announcement := range msg.Announcements {
-		for _, prefix := range announcement.Prefixes {
+		for _, announced := range announcement.Prefixes {
 			prefix, err := canonicalPrefix(announced)
 			if err != nil {
 				h.logger.Error(
 					"Dropped an announcement whose prefix does not parse",
 					"prefix", announced, "peer", msg.Peer, "error", err)
 				continue
 			}
 			h.batch = append(h.batch, prefixUpdate{
-				prefix:      canonicalPrefix(prefix),
+				prefix:      prefix,
 				originASN:   originASN,
 				peer:        msg.Peer,
 				messageType: "announcement",
@@ -373,10 +391,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
+		// prefix: a withdrawal, or an announcement whose path is empty or ends
-		// holds AS 0 anywhere (the decoder empties such a path). Such an
+		// in AS 0. Such an announcement replaces the peer's route, as any
-		// announcement replaces the peer's route, as any announcement does,
+		// announcement does, with none (RFC 7607 treats AS 0 in a path as a
-		// with none (RFC 7607 treats AS 0 in a path as a withdrawal).
+		// withdrawal).
 		if update.messageType == "announcement" &&
 			(update.originASN > 0 || update.endsInSet) {
 			// Create live route for batch upsert
@@ -560,15 +578,15 @@ func (h *PrefixHandler) writeBatch(
 // the IP lookup builds when it looks a prefix up. The feed sends IPv6
 // announcements compressed ("2001:db8::/32") but IPv6 withdrawals uncompressed
 // ("2001:db8:0:0:0:0:0:0/32"); stored as received, a withdrawal would not match
-// the route its announcement stored. A prefix that does not parse is returned
+// the route its announcement stored. It returns an error for a prefix that does
-// unchanged, and the batch flush reports it.
+// not parse.
-func canonicalPrefix(prefix string) string {
+func canonicalPrefix(prefix string) (string, error) {
 	p, err := netip.ParsePrefix(prefix)
 	if err != nil {
-		return prefix
+		return "", err
 	}
-	return p.Masked().String()
+	return p.Masked().String(), nil
 }
 // parseCIDR returns the mask length and the IP version of a prefix string
@@ -381,12 +381,11 @@ 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, an empty one, or one
+// an AS path that gives no origin: one whose last AS is 0, or an empty one. As
-// that holds AS 0 before its end, which decodes as an empty path. As any later
+// any later announcement from the peer does, the second replaces the peer's
-// announcement from the peer does, the second replaces the peer's route, here
+// route, here with none, whether it comes in a later batch or follows a
-// with none, whether it comes in a later batch or follows a withdrawal in one
+// withdrawal in one batch. The in-memory route count and prefix distribution
-// batch. The in-memory route count and prefix distribution must match the
+// must match the route table.
 // route table.
 func TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin(
 	t *testing.T,
 ) {
@@ -422,12 +421,6 @@ 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)},
@@ -447,12 +440,12 @@ func TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin(
 	}
 }
-// TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse runs an announcement of
+// TestPrefixHandlerDropsAPrefixThatDoesNotParse 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) {
+func TestPrefixHandlerDropsAPrefixThatDoesNotParse(t *testing.T) {
 	t.Parallel()
 	cfg := &config.Config{StateDir: t.TempDir()}
@@ -1,54 +0,0 @@
 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)
 			}
 		})
 	}
 }
@@ -19,8 +19,8 @@ const (
 	Author = "@sneak"
 	// AuthorURL is the author's website
 	AuthorURL = "https://sneak.berlin"
-	// License is the program license, as LICENSE grants it
+	// License is the program license
-	License = "MIT"
+	License = "WTFPL"
 	// RepoURL is the git repository URL
 	RepoURL = "https://git.eeqj.de/sneak/routewatch"
 )