Bound each shutdown stage by the stop deadline (closes #100)

Shutdown passes its context to every stage: the streamer, the four
handlers, the WHOIS fetcher, the database maintainer and the HTTP server
each stop waiting when it ends. A handler then skips its last write and
logs how many updates it did not write, and the HTTP server closes the
connections still open. The database is checkpointed and closed last,
and left open once the deadline has passed, since a write may still
hold it. The stop hook gives Shutdown 8 seconds and fx 9, inside the 10
seconds upaas and docker stop allow, so a stop signal exits 0.

Model: opus-5-5

README.md

@@ -54,8 +54,9 @@ DEBUG=routewatch ./routewatch
   picked at random, each with its newest route (HTML)
 - `GET /prefixlength6/{length}` - The same for IPv6
 - `GET /ip/{addr}` - The same JSON answer as `/api/v1/ip/{ip}`
-- `GET /ip/` - The same, for the caller's own address, or the one a proxy names
+- `GET /ip/` - The same, for the caller's own address: the address the request
-  in `True-Client-IP`, `X-Real-IP` or `X-Forwarded-For`
+  came from or, when that is a trusted proxy (`TRUSTED_PROXIES`, see
+  Configuration), the client address in its `X-Forwarded-For`
 
 ### API v1
 
@@ -84,7 +85,17 @@ 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.
+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.
+
+Every answer carries `Strict-Transport-Security`, `Content-Security-Policy`,
+`X-Frame-Options`, `X-Content-Type-Options`, `Referrer-Policy` and
+`Permissions-Policy`. The Content-Security-Policy lets a page load content only
+from the server, and does not allow inline content with `'unsafe-inline'`: the
+pages keep their CSS and JavaScript in inline `<style>` and `<script>` blocks,
+so each answer's policy names a random nonce made for that answer, and the
+browser applies or runs only the blocks that carry it. The pages therefore use
+no `style` attributes, which a nonce cannot cover.
 
 An origin AS of 0 (`origin_asn` in a route, `asn` in an IP lookup, the key `0`
 in a prefix's `origins`) means the route has no known origin: its AS path ends
@@ -191,7 +202,11 @@ routewatch/
   batches for performance
 - **Backpressure**: Probabilistic message dropping when queues exceed 50%
   capacity
-- **Graceful Shutdown**: 60-second timeout, flushes all pending batches
+- **Graceful Shutdown**: on SIGTERM or SIGINT the streamer stops, the handlers
+  write the batches they hold, the HTTP server stops and the database is
+  checkpointed and closed, all within 8 seconds, and the daemon exits with
+  status 0. What is not written by then is lost, and the handler logs how many
+  updates (see "Running under upaas").
 - **Reconnection**: Any end of the stream other than shutdown, including RIS
   Live ending the response normally, marks the feed disconnected and reconnects
   with exponential backoff (5s-320s). The backoff resets after a stream that
@@ -247,10 +262,20 @@ Configuration is handled via environment variables and OS-specific paths:
 | `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 |
+| `TRUSTED_PROXIES`  | (empty)                          | Addresses and CIDR ranges of trusted proxies; see below     |
 
 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.
 
+`TRUSTED_PROXIES` lists the reverse proxies whose `X-Forwarded-For` the server
+believes, comma-separated, such as `10.0.0.0/8,192.0.2.1`. For a request from
+one of them, the request log and `/ip/` use the last address in
+`X-Forwarded-For` that is not a trusted proxy: each proxy adds the address it
+received the request from to the end, so the entries before it are whatever the
+client sent. For a request from any other address they use the address the
+request came from, whatever forwarding headers it carries. `X-Real-IP` and
+`True-Client-IP` are not read.
+
 State directory (database location):
 
 - macOS: `~/Library/Application Support/berlin.sneak.app.routewatch/`
@@ -336,15 +361,36 @@ 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
+- Environment: `TRUSTED_PROXIES`, set to the address the reverse proxy in front
-  `MALLOC_ARENA_MAX` at the image's values. `DEBUG=routewatch` is optional and
+  of the app connects from, as the container sees it; without it the request log
-  adds the `System stats` memory line to the log.
+  and `/ip/` give the proxy's address for every request. Nothing else is needed:
+  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`.
+- Stop grace: the container needs 9 seconds between SIGTERM and SIGKILL. upaas
+  gives 10 (`stopTimeoutSeconds` in its `internal/docker/client.go`), as
+  `docker stop` does by default. The daemon allows 8 seconds for the handlers to
+  write what they hold and for the requests in progress to finish, then stops
+  the HTTP server, closes the database and exits with status 0. A longer grace
+  changes nothing. What is not written after 8 seconds is lost, and the streamer
+  and each handler log how much. When the prefix handler has fallen behind the
+  feed, a stop loses far more: its batch is full at 250,000 updates and the
+  streamer's worker for it waits for room, so the streamer's stop takes the
+  whole 8 seconds and every handler then skips its last write. Lost then are the
+  messages left on the four handlers' queues, what the AS, peer and peering
+  handlers hold, the prefix handler's batch, and the write it has in progress,
+  up to another 250,000 updates, which take about 20 seconds and are cut off by
+  the exit; withdrawals and peer downs are among them. A route withdrawn in them
+  stays in the live route table until its peer announces or withdraws the prefix
+  again or its session goes down. When the 8 seconds run out the database is not
+  closed, and SQLite rolls back a write the exit cut off when the database next
+  opens.
 
 ## Development
 

cmd/streamdumper/main.go

@@ -35,7 +35,7 @@ func main() {
 	if err != nil {
 		log.Fatal("Failed to start streamer:", err)
 	}
-	defer s.Stop()
+	defer s.Stop(context.Background())
 
 	// Set up context with cancellation
 	ctx, cancel := context.WithCancel(context.Background())

internal/config/config.go

@@ -4,10 +4,12 @@ package config
 import (
 	"errors"
 	"fmt"
+	"net/netip"
 	"os"
 	"path/filepath"
 	"runtime"
 	"strconv"
+	"strings"
 	"time"
 )
 
@@ -32,6 +34,8 @@ var (
 	errRelativeXDGDataHome = errors.New("XDG_DATA_HOME must be an absolute path")
 	errUnsupportedOS       = errors.New("unsupported operating system")
 	errInvalidPort         = errors.New("PORT must be a whole number")
+	errInvalidProxy        = errors.New(
+		"TRUSTED_PROXIES must list IP addresses and CIDR ranges, comma-separated")
 )
 
 // Config holds configuration for the entire application
@@ -42,6 +46,10 @@ type Config struct {
 	// Port is the TCP port the HTTP server listens on
 	Port int
 
+	// TrustedProxies are the reverse proxies whose X-Forwarded-For header
+	// the HTTP server believes
+	TrustedProxies []netip.Prefix
+
 	// MaxRuntime is the maximum runtime (0 = run forever)
 	MaxRuntime time.Duration
 
@@ -68,9 +76,15 @@ func New() (*Config, error) {
 		return nil, err
 	}
 
+	trustedProxies, err := getTrustedProxies()
+	if err != nil {
+		return nil, err
+	}
+
 	return &Config{
-		StateDir: stateDir,
+		StateDir:       stateDir,
-		Port:     port,
+		Port:           port,
+		TrustedProxies: trustedProxies,
 		// Run forever by default
 		MaxRuntime: 0,
 		// Enable batching by default
@@ -145,6 +159,38 @@ func getPort() (int, error) {
 	return int(port), nil
 }
 
+// getTrustedProxies returns the addresses and CIDR ranges listed in
+// TRUSTED_PROXIES, such as "10.0.0.0/8,192.0.2.1", or none when it is not set.
+// An address is returned as a range holding only that address.
+func getTrustedProxies() ([]netip.Prefix, error) {
+	value := os.Getenv("TRUSTED_PROXIES")
+	if value == "" {
+		return nil, nil
+	}
+
+	var proxies []netip.Prefix
+
+	for entry := range strings.SplitSeq(value, ",") {
+		entry = strings.TrimSpace(entry)
+
+		addr, err := netip.ParseAddr(entry)
+		if err == nil {
+			proxies = append(proxies, netip.PrefixFrom(addr, addr.BitLen()))
+
+			continue
+		}
+
+		prefix, err := netip.ParsePrefix(entry)
+		if err != nil {
+			return nil, fmt.Errorf("%w, got %q", errInvalidProxy, entry)
+		}
+
+		proxies = append(proxies, prefix)
+	}
+
+	return proxies, nil
+}
+
 // EnsureDirectories creates all necessary directories if they don't exist
 func (c *Config) EnsureDirectories() error {
 	// Ensure state directory exists

internal/config/config_internal_test.go

@@ -1,7 +1,9 @@
 package config
 
 import (
+	"net/netip"
 	"runtime"
+	"slices"
 	"testing"
 )
 
@@ -44,6 +46,55 @@ func TestNewRefusesInvalidPort(t *testing.T) {
 	}
 }
 
+func TestNewReadsTrustedProxies(t *testing.T) {
+	t.Setenv("PORT", "")
+	t.Setenv("XDG_DATA_HOME", "")
+
+	t.Setenv("TRUSTED_PROXIES", "")
+
+	cfg, err := New()
+	if err != nil {
+		t.Fatalf("New() without TRUSTED_PROXIES: %v", err)
+	}
+
+	if len(cfg.TrustedProxies) != 0 {
+		t.Errorf("New() without TRUSTED_PROXIES: TrustedProxies = %v, want none",
+			cfg.TrustedProxies)
+	}
+
+	t.Setenv("TRUSTED_PROXIES", "10.0.0.0/8, 192.0.2.1,2001:db8::/32")
+
+	cfg, err = New()
+	if err != nil {
+		t.Fatalf("New() with TRUSTED_PROXIES: %v", err)
+	}
+
+	want := []netip.Prefix{
+		netip.MustParsePrefix("10.0.0.0/8"),
+		netip.MustParsePrefix("192.0.2.1/32"),
+		netip.MustParsePrefix("2001:db8::/32"),
+	}
+	if !slices.Equal(cfg.TrustedProxies, want) {
+		t.Errorf("TrustedProxies = %v, want %v", cfg.TrustedProxies, want)
+	}
+}
+
+func TestNewRefusesInvalidTrustedProxies(t *testing.T) {
+	invalid := []string{"proxy.example", "10.0.0.0/33", "192.0.2.1:80", "192.0.2.1,"}
+	for _, value := range invalid {
+		t.Run(value, func(t *testing.T) {
+			t.Setenv("PORT", "")
+			t.Setenv("XDG_DATA_HOME", "")
+			t.Setenv("TRUSTED_PROXIES", value)
+
+			_, err := New()
+			if err == nil {
+				t.Errorf("New() with TRUSTED_PROXIES=%q returned no error", value)
+			}
+		})
+	}
+}
+
 func TestNewRefusesRelativeXDGDataHome(t *testing.T) {
 	if runtime.GOOS == "darwin" {
 		t.Skip("macOS does not read XDG_DATA_HOME")

internal/routewatch/app.go

@@ -137,7 +137,12 @@ func (rw *RouteWatch) Run(ctx context.Context) error {
 	return nil
 }
 
-// Shutdown performs graceful shutdown of all services
+// Shutdown stops all services in order: the streamer, the handlers, which
+// write what they hold, the WHOIS fetcher, the database maintainer and the
+// HTTP server, and last the database, which it checkpoints and closes. Each
+// stage waits at most until ctx ends, so Shutdown returns soon after: a
+// handler logs what it could not write by then, and the HTTP server closes
+// the connections of requests still running.
 func (rw *RouteWatch) Shutdown(ctx context.Context) {
 	rw.mu.Lock()
 	if rw.shutdown {
@@ -150,51 +155,48 @@ func (rw *RouteWatch) Shutdown(ctx context.Context) {
 	rw.mu.Unlock()
 
 	// Stop the streamer first: it returns once the handlers have been handed
-	// every message it queued. Then each handler writes what it holds.
+	// every message it queued, or when ctx ends. Then each handler writes
-	rw.streamer.Stop()
+	// what it holds.
+	rw.streamer.Stop(ctx)
 
 	if rw.dbHandler != nil {
 		rw.logger.Info("Flushing database handler")
-		rw.dbHandler.Stop()
+		rw.dbHandler.Stop(ctx)
 	}
 
 	if rw.peerHandler != nil {
 		rw.logger.Info("Flushing peer handler")
-		rw.peerHandler.Stop()
+		rw.peerHandler.Stop(ctx)
 	}
 
 	if rw.prefixHandler != nil {
 		rw.logger.Info("Flushing prefix handler")
-		rw.prefixHandler.Stop()
+		rw.prefixHandler.Stop(ctx)
 	}
 
 	if rw.peeringHandler != nil {
 		rw.logger.Info("Flushing peering handler")
-		rw.peeringHandler.Stop()
+		rw.peeringHandler.Stop(ctx)
 	}
 
 	// Stop ASN WHOIS fetcher
 	if rw.asnFetcher != nil {
-		rw.asnFetcher.Stop()
+		rw.asnFetcher.Stop(ctx)
 	}
 
 	// Stop database maintainer
 	if rw.dbMaintainer != nil {
-		rw.dbMaintainer.Stop()
+		rw.dbMaintainer.Stop(ctx)
 	}
 
-	// Stop HTTP server with a timeout of its own, however much of ctx is left
+	err := rw.server.Stop(ctx)
-	const serverStopTimeout = 5 * time.Second
-
-	stopCtx, cancel := context.WithTimeout(
-		context.WithoutCancel(ctx), serverStopTimeout)
-	defer cancel()
-
-	err := rw.server.Stop(stopCtx)
 	if err != nil {
-		rw.logger.Error("Failed to stop HTTP server gracefully", "error", err)
+		rw.logger.Error("Stopped the HTTP server before every request finished",
+			"error", err)
 	}
 
+	rw.closeDatabase(ctx)
+
 	// Log final metrics
 	metrics := rw.streamer.GetMetrics()
 	rw.logger.Info("Final metrics",
@@ -206,6 +208,27 @@ func (rw *RouteWatch) Shutdown(ctx context.Context) {
 	)
 }
 
+// closeDatabase checkpoints and closes the database, unless ctx has ended: a
+// write may then still be running, and both would wait for it. SQLite rolls
+// back a write cut off by the exit when the database next opens.
+func (rw *RouteWatch) closeDatabase(ctx context.Context) {
+	if ctx.Err() != nil {
+		rw.logger.Error("Stop deadline passed; the database is not closed")
+
+		return
+	}
+
+	err := rw.db.Checkpoint(ctx)
+	if err != nil {
+		rw.logger.Error("Failed to checkpoint the database", "error", err)
+	}
+
+	err = rw.db.Close()
+	if err != nil {
+		rw.logger.Error("Failed to close the database", "error", err)
+	}
+}
+
 // getModule provides all fx dependencies
 func getModule() []fx.Option {
 	return []fx.Option{

internal/routewatch/app_integration_internal_test.go

@@ -4,8 +4,11 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
+	"slices"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -643,3 +646,175 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
 		}
 	}
 }
+
+// freePort returns a TCP port that was free a moment ago, for a server the
+// test starts.
+func freePort(t *testing.T) int {
+	t.Helper()
+
+	var listenConfig net.ListenConfig
+
+	listener, err := listenConfig.Listen(t.Context(), "tcp", ":0")
+	if err != nil {
+		t.Fatalf("failed to find a free port: %v", err)
+	}
+	defer func() { _ = listener.Close() }()
+
+	addr, ok := listener.Addr().(*net.TCPAddr)
+	if !ok {
+		t.Fatalf("listener address %v is not a TCP address", listener.Addr())
+	}
+
+	return addr.Port
+}
+
+// TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked starts the HTTP server
+// and has the AS handler start a write that the store holds until the test
+// ends, past Shutdown's deadline. Shutdown must return within its deadline
+// plus a small margin, with the HTTP server stopped. Before the fix Shutdown
+// waited for the write.
+func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
+	t.Parallel()
+
+	const (
+		deadline = 100 * time.Millisecond
+		// margin is how long Shutdown may take after its deadline, to stop
+		// the HTTP server.
+		margin = 2 * time.Second
+	)
+
+	store := newBlockingStore()
+	defer store.releaseWrites()
+
+	log := logger.New()
+	port := freePort(t)
+	cfg := &config.Config{Port: port}
+	s := streamer.New(log, metrics.New())
+
+	rw := New(Dependencies{
+		DB:       store,
+		Streamer: s,
+		Server:   server.New(store, s, log, cfg),
+		Logger:   log,
+		Config:   cfg,
+	})
+
+	err := rw.server.Start(t.Context())
+	if err != nil {
+		t.Fatalf("failed to start the HTTP server: %v", err)
+	}
+
+	rw.dbHandler = NewASHandler(store, log)
+	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
+	startBlockedWrite(t, store, rw.dbHandler, []*ristypes.RISMessage{{Path: fill}})
+
+	ctx, cancel := context.WithTimeout(t.Context(), deadline)
+	defer cancel()
+
+	shutDown := make(chan struct{})
+
+	go func() {
+		rw.Shutdown(ctx)
+		close(shutDown)
+	}()
+
+	select {
+	case <-shutDown:
+	case <-time.After(deadline + margin):
+		t.Fatalf("Shutdown did not return within %s of its deadline while "+
+			"a write was blocked", margin)
+	}
+
+	var dialer net.Dialer
+
+	addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port))
+
+	conn, err := dialer.DialContext(t.Context(), "tcp", addr)
+	if err == nil {
+		_ = conn.Close()
+
+		t.Errorf("the HTTP server still accepts connections on %s after "+
+			"Shutdown returned", addr)
+	}
+}
+
+// closeOrderStore is a mockStore that records, in order, the AS writes,
+// checkpoints and closes it is given.
+type closeOrderStore struct {
+	*mockStore
+
+	callsMu sync.Mutex
+	calls   []string
+}
+
+// The calls closeOrderStore records.
+const (
+	callWrite      = "AS write"
+	callCheckpoint = "checkpoint"
+	callClose      = "close"
+)
+
+func (s *closeOrderStore) GetOrCreateASNBatch(asns map[int]time.Time) error {
+	s.record(callWrite)
+
+	return s.mockStore.GetOrCreateASNBatch(asns)
+}
+
+func (s *closeOrderStore) Checkpoint(_ context.Context) error {
+	s.record(callCheckpoint)
+
+	return nil
+}
+
+func (s *closeOrderStore) Close() error {
+	s.record(callClose)
+
+	return nil
+}
+
+func (s *closeOrderStore) record(call string) {
+	s.callsMu.Lock()
+	defer s.callsMu.Unlock()
+
+	s.calls = append(s.calls, call)
+}
+
+// TestShutdownCheckpointsAndClosesTheDatabaseLast hands the AS handler one
+// message and shuts down with time to spare. The handler's last write must
+// come first, then a checkpoint, then the close. Before the fix Shutdown
+// neither checkpointed nor closed the database.
+func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
+	t.Parallel()
+
+	const deadline = 10 * time.Second
+
+	store := &closeOrderStore{mockStore: newMockStore()}
+	log := logger.New()
+	cfg := &config.Config{}
+	s := streamer.New(log, metrics.New())
+
+	rw := New(Dependencies{
+		DB:       store,
+		Streamer: s,
+		Server:   server.New(store, s, log, cfg),
+		Logger:   log,
+		Config:   cfg,
+	})
+	rw.dbHandler = NewASHandler(store, log)
+	rw.dbHandler.HandleMessage(&ristypes.RISMessage{
+		Path: ristypes.ASPath{Sequence: []int{testASNA}},
+	})
+
+	ctx, cancel := context.WithTimeout(t.Context(), deadline)
+	defer cancel()
+
+	rw.Shutdown(ctx)
+
+	store.callsMu.Lock()
+	defer store.callsMu.Unlock()
+
+	want := []string{callWrite, callCheckpoint, callClose}
+	if !slices.Equal(store.calls, want) {
+		t.Errorf("Shutdown made the calls %q, want %q", store.calls, want)
+	}
+}

internal/routewatch/ashandler.go

@@ -1,6 +1,7 @@
 package routewatch
 
 import (
+	"context"
 	"slices"
 	"sync"
 	"time"
@@ -39,18 +40,23 @@ type ASHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// Batching. mu guards batch only: the flush goroutine takes the batch
+	// Batching. mu guards batch and writing only: the flush goroutine takes
-	// under mu and writes it after releasing mu, so HandleMessage waits for
+	// the batch under mu and writes it after releasing mu, so HandleMessage
-	// a database write only while batch is at asnBatchMaxSize.
+	// waits for a database write only while batch is at asnBatchMaxSize.
 	mu    sync.Mutex
 	batch []asnOp
+	// writing is how many operations the flush goroutine took from batch
+	// and has not finished writing.
+	writing int
 	// batchTaken is signalled when the flush goroutine takes batch; its L is
 	// &mu.
 	batchTaken  sync.Cond
 	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
 	lastFlush   time.Time     // used only by the flush goroutine
-	stopCh      chan struct{}
+	// stopCh hands Stop's context to the flush goroutine. It holds one, so
-	wg          sync.WaitGroup
+	// Stop does not wait for a write in progress to hand it over.
+	stopCh chan context.Context
+	done   chan struct{} // closed when the flush goroutine returns
 }
 
 type asnOp struct {
@@ -70,12 +76,12 @@ func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
 		batch:       make([]asnOp, 0, asnBatchSize),
 		batchFullCh: make(chan struct{}, 1),
 		lastFlush:   time.Now(),
-		stopCh:      make(chan struct{}),
+		stopCh:      make(chan context.Context, 1),
+		done:        make(chan struct{}),
 	}
 	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
-	h.wg.Add(1)
 	go h.flushLoop()
 
 	return h
@@ -147,20 +153,32 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// Stop gracefully shuts down the ASHandler by signaling the background flush
+// Stop stops the flush goroutine, which writes the operations the handler holds
-// goroutine to terminate and waiting for it to complete. Any pending ASN
+// before it returns, and waits for it until ctx ends. Once ctx has ended the
-// operations in the current batch are flushed to the database before Stop
+// flush goroutine skips that last write, and a write still running is lost
-// returns. This method should be called during application shutdown to ensure
+// when the process exits, so Stop logs how many operations are not written.
-// no data is lost.
+func (h *ASHandler) Stop(ctx context.Context) {
-func (h *ASHandler) Stop() {
+	h.stopCh <- ctx
-	close(h.stopCh)
+
-	h.wg.Wait()
+	select {
+	case <-h.done:
+	case <-ctx.Done():
+	}
+
+	h.mu.Lock()
+	unwritten := h.writing + len(h.batch)
+	h.mu.Unlock()
+
+	if unwritten > 0 {
+		h.logger.Error("Stop deadline passed before the AS handler wrote "+
+			"its updates", "updates", unwritten)
+	}
 }
 
 // flushLoop runs in a goroutine, the only one that writes the batch: it
 // flushes when the batch is full, periodically, and once more on Stop.
 func (h *ASHandler) flushLoop() {
-	defer h.wg.Done()
+	defer close(h.done)
 
 	ticker := time.NewTicker(asnBatchTimeout)
 	defer ticker.Stop()
@@ -173,9 +191,11 @@ func (h *ASHandler) flushLoop() {
 			}
 		case <-h.batchFullCh:
 			h.flushBatch()
-		case <-h.stopCh:
+		case ctx := <-h.stopCh:
-			// Final flush
+			// Final flush, unless the stop deadline has passed
-			h.flushBatch()
+			if ctx.Err() == nil {
+				h.flushBatch()
+			}
 
 			return
 		}
@@ -183,7 +203,8 @@ func (h *ASHandler) flushLoop() {
 }
 
 // takeBatch returns the pending operations and leaves an empty batch in their
-// place for HandleMessage to fill.
+// place for HandleMessage to fill. They count in writing until flushBatch has
+// written them.
 func (h *ASHandler) takeBatch() []asnOp {
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -193,6 +214,7 @@ func (h *ASHandler) takeBatch() []asnOp {
 	}
 
 	batch := h.batch
+	h.writing = len(batch)
 	h.batch = make([]asnOp, 0, asnBatchSize)
 	h.batchTaken.Broadcast()
 
@@ -222,5 +244,9 @@ func (h *ASHandler) flushBatch() {
 			"error", err, "count", len(asnMap))
 	}
 
+	h.mu.Lock()
+	h.writing = 0
+	h.mu.Unlock()
+
 	h.lastFlush = time.Now()
 }

internal/routewatch/asnfetcher.go

@@ -43,7 +43,7 @@ type ASNFetcher struct {
 	logger         *slog.Logger
 	immediateQueue chan int
 	stopCh         chan struct{}
-	wg             sync.WaitGroup
+	done           chan struct{} // closed when the fetcher goroutine returns
 
 	// fetchMu ensures only one fetch runs at a time
 	fetchMu sync.Mutex
@@ -67,6 +67,7 @@ func NewASNFetcher(db database.Store, logger *slog.Logger) *ASNFetcher {
 		logger:          logger.With("component", "asn_fetcher"),
 		immediateQueue:  make(chan int, immediateQueueSize),
 		stopCh:          make(chan struct{}),
+		done:            make(chan struct{}),
 		currentInterval: baseInterval,
 		successTimes:    make([]time.Time, 0),
 		errorTimes:      make([]time.Time, 0),
@@ -76,8 +77,6 @@ func NewASNFetcher(db database.Store, logger *slog.Logger) *ASNFetcher {
 // Start begins the background ASN fetcher goroutine. It runs until Stop:
 // the end of ctx does not stop it.
 func (f *ASNFetcher) Start(ctx context.Context) {
-	f.wg.Add(1)
-
 	go f.run(context.WithoutCancel(ctx))
 
 	f.logger.Info("ASN fetcher started",
@@ -87,11 +86,17 @@ func (f *ASNFetcher) Start(ctx context.Context) {
 	)
 }
 
-// Stop gracefully shuts down the fetcher.
+// Stop stops the fetcher and waits, until ctx ends, for a lookup in progress to
-func (f *ASNFetcher) Stop() {
+// finish.
+func (f *ASNFetcher) Stop(ctx context.Context) {
 	close(f.stopCh)
-	f.wg.Wait()
+
-	f.logger.Info("ASN fetcher stopped")
+	select {
+	case <-f.done:
+		f.logger.Info("ASN fetcher stopped")
+	case <-ctx.Done():
+		f.logger.Warn("Stop deadline passed before the ASN fetcher stopped")
+	}
 }
 
 // QueueImmediate queues an ASN for immediate WHOIS lookup.
@@ -197,7 +202,7 @@ func (f *ASNFetcher) recordFailure() {
 
 // run is the main background loop.
 func (f *ASNFetcher) run(ctx context.Context) {
-	defer f.wg.Done()
+	defer close(f.done)
 
 	timer := time.NewTimer(f.getInterval())
 	defer timer.Stop()

internal/routewatch/batchwrite_internal_test.go

@@ -1,6 +1,10 @@
 package routewatch
 
 import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"log/slog"
 	"slices"
 	"sync"
 	"testing"
@@ -22,8 +26,9 @@ const (
 )
 
 // blockingStore is a mockStore whose batch writes for the AS, peer and prefix
-// handlers wait until the test calls releaseWrites. writeStarted is closed when
+// handlers, delete of a peer's routes and peering write wait until the test
-// the first of them starts. It records what each write was given.
+// calls releaseWrites. writeStarted is closed when the first of them starts. It
+// records what each batch write was given.
 type blockingStore struct {
 	*mockStore
 
@@ -85,6 +90,18 @@ func (s *blockingStore) UpdatePeerBatch(peers map[string]database.PeerUpdate) er
 	return nil
 }
 
+func (s *blockingStore) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
+	s.waitForRelease()
+
+	return s.mockStore.DeleteLiveRoutesOfPeer(peerIP)
+}
+
+func (s *blockingStore) RecordPeering(asA, asB int, ts time.Time) error {
+	s.waitForRelease()
+
+	return s.mockStore.RecordPeering(asA, asB, ts)
+}
+
 func (s *blockingStore) waitForRelease() {
 	s.startOnce.Do(func() { close(s.writeStarted) })
 	<-s.release
@@ -96,6 +113,40 @@ func (s *blockingStore) releaseWrites() {
 	s.releaseOnce.Do(func() { close(s.release) })
 }
 
+// stopLog holds what a handler's Stop logs it did not write.
+//
+//nolint:tagliatelle // field names are the keys the handlers log
+type stopLog struct {
+	Msg       string `json:"msg"`
+	Updates   int    `json:"updates"`
+	PeersDown int    `json:"peers_down"`
+	Paths     int    `json:"paths"`
+}
+
+// newErrorLogger returns a logger that writes only errors, as JSON lines, to
+// logs.
+func newErrorLogger(logs *bytes.Buffer) *logger.Logger {
+	return &logger.Logger{Logger: slog.New(slog.NewJSONHandler(logs,
+		&slog.HandlerOptions{Level: slog.LevelError}))}
+}
+
+// lastLogged decodes the last line in logs, which is the one Stop logs: a
+// handler logs nothing after it.
+func lastLogged(t *testing.T, logs *bytes.Buffer) stopLog {
+	t.Helper()
+
+	lines := bytes.Split(bytes.TrimSpace(logs.Bytes()), []byte("\n"))
+
+	var logged stopLog
+
+	err := json.Unmarshal(lines[len(lines)-1], &logged)
+	if err != nil {
+		t.Fatalf("want the error Stop logs, got %q: %v", logs.String(), err)
+	}
+
+	return logged
+}
+
 // returnsInTime reports whether f returns within handleMessageTimeout.
 func returnsInTime(f func()) bool {
 	done := make(chan struct{})
@@ -151,14 +202,14 @@ func handleWhileAWriteIsBlocked(
 	store *blockingStore,
 	h interface {
 		HandleMessage(msg *ristypes.RISMessage)
-		Stop()
+		Stop(ctx context.Context)
 	},
 	fill []*ristypes.RISMessage,
 	extra *ristypes.RISMessage,
 ) {
 	t.Helper()
 
-	defer h.Stop()
+	defer h.Stop(t.Context())
 	defer store.releaseWrites()
 
 	startBlockedWrite(t, store, h, fill)
@@ -178,14 +229,14 @@ func handleAtMaxBatchSize(
 	store *blockingStore,
 	h interface {
 		HandleMessage(msg *ristypes.RISMessage)
-		Stop()
+		Stop(ctx context.Context)
 	},
 	fill, toMax []*ristypes.RISMessage,
 	extra *ristypes.RISMessage,
 ) {
 	t.Helper()
 
-	defer h.Stop()
+	defer h.Stop(t.Context())
 	defer store.releaseWrites()
 
 	startBlockedWrite(t, store, h, fill)
@@ -217,6 +268,173 @@ func handleAtMaxBatchSize(
 	}
 }
 
+// stopWhileAWriteIsBlocked starts a blocked write with the messages in fill,
+// hands h the message extra, and stops h with a deadline that passes while
+// store holds the write. Stop must return all the same. Then it lets the write
+// finish and waits until done, closed when h's flush goroutine returns. Past
+// the deadline that goroutine skips its last write, so extra is not written,
+// and Stop logs as not written both extra and what the held write took.
+func stopWhileAWriteIsBlocked(
+	t *testing.T,
+	store *blockingStore,
+	h interface {
+		HandleMessage(msg *ristypes.RISMessage)
+		Stop(ctx context.Context)
+	},
+	done <-chan struct{},
+	fill []*ristypes.RISMessage,
+	extra *ristypes.RISMessage,
+) {
+	t.Helper()
+
+	const stopDeadline = 100 * time.Millisecond
+
+	defer store.releaseWrites()
+
+	startBlockedWrite(t, store, h, fill)
+	h.HandleMessage(extra)
+
+	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
+	defer cancel()
+
+	if !returnsInTime(func() { h.Stop(ctx) }) {
+		t.Fatalf("Stop with a deadline of %s did not return while a write "+
+			"was blocked", stopDeadline)
+	}
+
+	store.releaseWrites()
+
+	select {
+	case <-done:
+	case <-time.After(handleMessageTimeout):
+		t.Fatal("the flush goroutine did not return after the write finished")
+	}
+}
+
+func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	const extraPrefix = "198.51.100.0/24"
+
+	var logs bytes.Buffer
+
+	store := newBlockingStore()
+	h := NewPrefixHandler(store, newErrorLogger(&logs))
+	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
+	// The peer's session goes down first, so the write held is the delete of
+	// its routes, and Stop must log that peer down too.
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		[]*ristypes.RISMessage{sessionDown(), prefixAnnouncement(fill)},
+		prefixAnnouncement([]string{extraPrefix}))
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+
+	if store.prefixes[extraPrefix] {
+		t.Errorf("%s was written after the stop deadline", extraPrefix)
+	}
+
+	logged := lastLogged(t, &logs)
+	if logged.Updates != prefixBatchSize+1 || logged.PeersDown != 1 {
+		t.Errorf("Stop logged %d updates and %d peer downs not written, "+
+			"want %d and 1", logged.Updates, logged.PeersDown, prefixBatchSize+1)
+	}
+}
+
+func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	// Each message gives an update for its origin AS and one for each AS
+	// in its path: fill asnBatchSize+1, extra 2.
+	const wantUpdates = asnBatchSize + 1 + 2
+
+	var logs bytes.Buffer
+
+	store := newBlockingStore()
+	h := NewASHandler(store, newErrorLogger(&logs))
+	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		[]*ristypes.RISMessage{{Path: fill}},
+		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+
+	if store.asns[testASNB] {
+		t.Errorf("AS%d was written after the stop deadline", testASNB)
+	}
+
+	if logged := lastLogged(t, &logs); logged.Updates != wantUpdates {
+		t.Errorf("Stop logged %d updates not written, want %d",
+			logged.Updates, wantUpdates)
+	}
+}
+
+func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	const otherPeerIP = "2001:db8:ffff::2"
+
+	var logs bytes.Buffer
+
+	store := newBlockingStore()
+	h := NewPeerHandler(store, newErrorLogger(&logs))
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
+		&ristypes.RISMessage{Peer: otherPeerIP})
+
+	store.mu.Lock()
+	defer store.mu.Unlock()
+
+	if store.peers[otherPeerIP] {
+		t.Errorf("peer %s was written after the stop deadline", otherPeerIP)
+	}
+
+	if logged := lastLogged(t, &logs); logged.Updates != peerBatchSize+1 {
+		t.Errorf("Stop logged %d updates not written, want %d",
+			logged.Updates, peerBatchSize+1)
+	}
+}
+
+// TestPeeringHandlerStopCountsTheWriteInProgress hands a running peering
+// handler one AS path and stops it with a deadline that passes while store
+// holds Stop's last write, of that path's peering. Stop must log that path as
+// not written.
+func TestPeeringHandlerStopCountsTheWriteInProgress(t *testing.T) {
+	t.Parallel()
+
+	const stopDeadline = 100 * time.Millisecond
+
+	var logs bytes.Buffer
+
+	store := newBlockingStore()
+	defer store.releaseWrites()
+
+	h := NewPeeringHandler(store, newErrorLogger(&logs))
+	handle(h, time.Now().UTC(), testASNA, testASNB)
+
+	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
+	defer cancel()
+
+	if !returnsInTime(func() { h.Stop(ctx) }) {
+		t.Fatalf("Stop with a deadline of %s did not return while a write "+
+			"was blocked", stopDeadline)
+	}
+
+	store.releaseWrites()
+
+	select {
+	case <-h.done:
+	case <-time.After(handleMessageTimeout):
+		t.Fatal("the processing goroutine did not return after the write " +
+			"finished")
+	}
+
+	if logged := lastLogged(t, &logs); logged.Paths != 1 {
+		t.Errorf("Stop logged %d paths not written, want 1", logged.Paths)
+	}
+}
+
 // prefixAnnouncement returns a message in which testPeerIP announces prefixes
 // with origin AS testASNA.
 func prefixAnnouncement(prefixes []string) *ristypes.RISMessage {

internal/routewatch/cli.go

@@ -14,8 +14,14 @@ import (
 )
 
 const (
-	// shutdownTimeout is the maximum time allowed for graceful shutdown
+	// shutdownTimeout is how long Shutdown has after a stop signal. upaas,
-	shutdownTimeout = 60 * time.Second
+	// like docker stop by default, kills the container 10 seconds after the
+	// stop signal (README "Running under upaas").
+	shutdownTimeout = 8 * time.Second
+	// stopTimeout is fx's limit on the stop hook, past which fx exits with
+	// status 1: shutdownTimeout and time for Shutdown to close the HTTP
+	// server after its deadline.
+	stopTimeout = 9 * time.Second
 	// debugInterval is how often to log debug stats
 	debugInterval = 60 * time.Second
 	// bytesPerMB is bytes per megabyte
@@ -63,7 +69,7 @@ func logDebugStats(logger *logger.Logger) {
 func CLIEntry() {
 	app := fx.New(
 		fx.Options(getModule()...),
-		fx.StopTimeout(shutdownTimeout), // Allow 60 seconds for graceful shutdown
+		fx.StopTimeout(stopTimeout),
 		fx.Invoke(func(
 			lc fx.Lifecycle,
 			rw *RouteWatch,
@@ -106,6 +112,10 @@ func CLIEntry() {
 				},
 				OnStop: func(ctx context.Context) error {
 					logger.Info("Shutting down RouteWatch")
+
+					ctx, cancel := context.WithTimeout(ctx, shutdownTimeout)
+					defer cancel()
+
 					rw.Shutdown(ctx)
 
 					return nil

internal/routewatch/dbmaintainer.go

@@ -46,7 +46,7 @@ type DBMaintainer struct {
 	db     database.Store
 	logger *slog.Logger
 	stopCh chan struct{}
-	wg     sync.WaitGroup
+	done   chan struct{} // closed when the maintenance goroutine returns
 
 	// Stats tracking
 	statsMu             sync.Mutex
@@ -67,14 +67,13 @@ func NewDBMaintainer(db database.Store, logger *slog.Logger) *DBMaintainer {
 		db:     db,
 		logger: logger.With("component", "db_maintainer"),
 		stopCh: make(chan struct{}),
+		done:   make(chan struct{}),
 	}
 }
 
 // Start begins the background maintenance goroutine. It runs until Stop:
 // the end of ctx does not stop it.
 func (m *DBMaintainer) Start(ctx context.Context) {
-	m.wg.Add(1)
-
 	go m.run(context.WithoutCancel(ctx))
 
 	m.logger.Info("Database maintainer started",
@@ -84,16 +83,23 @@ func (m *DBMaintainer) Start(ctx context.Context) {
 	)
 }
 
-// Stop gracefully shuts down the maintainer.
+// Stop stops the maintainer and waits, until ctx ends, for a maintenance task
-func (m *DBMaintainer) Stop() {
+// in progress to finish.
+func (m *DBMaintainer) Stop(ctx context.Context) {
 	close(m.stopCh)
-	m.wg.Wait()
+
-	m.logger.Info("Database maintainer stopped")
+	select {
+	case <-m.done:
+		m.logger.Info("Database maintainer stopped")
+	case <-ctx.Done():
+		m.logger.Warn("Stop deadline passed before the database maintainer " +
+			"stopped")
+	}
 }
 
 // run is the main background loop.
 func (m *DBMaintainer) run(ctx context.Context) {
-	defer m.wg.Done()
+	defer close(m.done)
 
 	// Use different timers for each task
 	checkpointTimer := time.NewTimer(checkpointInterval)

internal/routewatch/peerhandler.go

@@ -5,6 +5,7 @@ package routewatch
 // peer state in the database.
 
 import (
+	"context"
 	"strconv"
 	"sync"
 	"time"
@@ -42,18 +43,24 @@ type PeerHandler struct {
 	db     database.Store
 	logger *logger.Logger
 
-	// Batching. mu guards peerBatch only: the flush goroutine takes the batch
+	// Batching. mu guards peerBatch and writing only: the flush goroutine
-	// under mu and writes it after releasing mu, so HandleMessage waits for
+	// takes the batch under mu and writes it after releasing mu, so
-	// a database write only while peerBatch is at peerBatchMaxSize.
+	// HandleMessage waits for a database write only while peerBatch is at
+	// peerBatchMaxSize.
 	mu        sync.Mutex
 	peerBatch []peerUpdate
+	// writing is how many updates the flush goroutine took from peerBatch
+	// and has not finished writing.
+	writing int
 	// batchTaken is signalled when the flush goroutine takes peerBatch; its L
 	// is &mu.
 	batchTaken  sync.Cond
 	batchFullCh chan struct{} // wakes the flush goroutine when peerBatch is full
 	lastFlush   time.Time     // used only by the flush goroutine
-	stopCh      chan struct{}
+	// stopCh hands Stop's context to the flush goroutine. It holds one, so
-	wg          sync.WaitGroup
+	// Stop does not wait for a write in progress to hand it over.
+	stopCh chan context.Context
+	done   chan struct{} // closed when the flush goroutine returns
 }
 
 type peerUpdate struct {
@@ -74,12 +81,12 @@ func NewPeerHandler(db database.Store, logger *logger.Logger) *PeerHandler {
 		peerBatch:   make([]peerUpdate, 0, peerBatchSize),
 		batchFullCh: make(chan struct{}, 1),
 		lastFlush:   time.Now(),
-		stopCh:      make(chan struct{}),
+		stopCh:      make(chan context.Context, 1),
+		done:        make(chan struct{}),
 	}
 	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
-	h.wg.Add(1)
 	go h.flushLoop()
 
 	return h
@@ -144,16 +151,32 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// Stop gracefully stops the handler and flushes remaining batches
+// Stop stops the flush goroutine, which writes the updates the handler holds
-func (h *PeerHandler) Stop() {
+// before it returns, and waits for it until ctx ends. Once ctx has ended the
-	close(h.stopCh)
+// flush goroutine skips that last write, and a write still running is lost
-	h.wg.Wait()
+// when the process exits, so Stop logs how many updates are not written.
+func (h *PeerHandler) Stop(ctx context.Context) {
+	h.stopCh <- ctx
+
+	select {
+	case <-h.done:
+	case <-ctx.Done():
+	}
+
+	h.mu.Lock()
+	unwritten := h.writing + len(h.peerBatch)
+	h.mu.Unlock()
+
+	if unwritten > 0 {
+		h.logger.Error("Stop deadline passed before the peer handler wrote "+
+			"its updates", "updates", unwritten)
+	}
 }
 
 // flushLoop runs in a goroutine, the only one that writes the batch: it
 // flushes when the batch is full, periodically, and once more on Stop.
 func (h *PeerHandler) flushLoop() {
-	defer h.wg.Done()
+	defer close(h.done)
 
 	ticker := time.NewTicker(peerBatchTimeout)
 	defer ticker.Stop()
@@ -166,9 +189,11 @@ func (h *PeerHandler) flushLoop() {
 			}
 		case <-h.batchFullCh:
 			h.flushBatch()
-		case <-h.stopCh:
+		case ctx := <-h.stopCh:
-			// Final flush
+			// Final flush, unless the stop deadline has passed
-			h.flushBatch()
+			if ctx.Err() == nil {
+				h.flushBatch()
+			}
 
 			return
 		}
@@ -176,7 +201,8 @@ func (h *PeerHandler) flushLoop() {
 }
 
 // takeBatch returns the pending updates and leaves an empty batch in their
-// place for HandleMessage to fill.
+// place for HandleMessage to fill. They count in writing until flushBatch has
+// written them.
 func (h *PeerHandler) takeBatch() []peerUpdate {
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -186,6 +212,7 @@ func (h *PeerHandler) takeBatch() []peerUpdate {
 	}
 
 	batch := h.peerBatch
+	h.writing = len(batch)
 	h.peerBatch = make([]peerUpdate, 0, peerBatchSize)
 	h.batchTaken.Broadcast()
 
@@ -227,5 +254,9 @@ func (h *PeerHandler) flushBatch() {
 			"error", err, "count", len(dbPeerMap))
 	}
 
+	h.mu.Lock()
+	h.writing = 0
+	h.mu.Unlock()
+
 	h.lastFlush = time.Now()
 }

internal/routewatch/peeringhandler.go

@@ -1,6 +1,7 @@
 package routewatch
 
 import (
+	"context"
 	"encoding/json"
 	"sync"
 	"time"
@@ -45,9 +46,14 @@ type PeeringHandler struct {
 	mu           sync.Mutex
 	asPaths      map[string]time.Time // key is JSON-encoded AS path
 	droppedPaths int                  // paths dropped because the map was full
+	// writing is how many paths processPeerings took from asPaths and has
+	// not finished writing the peerings of.
+	writing int
 
-	stopCh chan struct{}
+	// stopCh hands Stop's context to the processing goroutine. It holds one,
-	wg     sync.WaitGroup
+	// so Stop does not wait for a write in progress to hand it over.
+	stopCh chan context.Context
+	done   chan struct{} // closed when the processing goroutine returns
 }
 
 // NewPeeringHandler creates and initializes a new PeeringHandler with the
@@ -59,12 +65,11 @@ func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler
 		db:      db,
 		logger:  logger,
 		asPaths: make(map[string]time.Time),
-		stopCh:  make(chan struct{}),
+		stopCh:  make(chan context.Context, 1),
+		done:    make(chan struct{}),
 	}
 
 	// Start the periodic processing goroutine
-	h.wg.Add(1)
-
 	go h.processLoop()
 
 	return h
@@ -132,18 +137,33 @@ func (h *PeeringHandler) ProcessPeeringsNow() {
 	h.processPeerings()
 }
 
-// Stop gracefully shuts down the handler by signaling the background
+// Stop stops the processing goroutine, which writes the peerings of the AS
-// goroutine to stop and waiting while it processes the remaining AS paths,
+// paths the handler holds before it returns, and waits for it until ctx ends.
-// so no peering data is lost during shutdown.
+// Once ctx has ended the processing goroutine skips that last write, and a
-func (h *PeeringHandler) Stop() {
+// write still running is lost when the process exits, so Stop logs how many
-	close(h.stopCh)
+// paths' peerings are not written.
-	h.wg.Wait()
+func (h *PeeringHandler) Stop(ctx context.Context) {
+	h.stopCh <- ctx
+
+	select {
+	case <-h.done:
+	case <-ctx.Done():
+	}
+
+	h.mu.Lock()
+	unwritten := h.writing + len(h.asPaths)
+	h.mu.Unlock()
+
+	if unwritten > 0 {
+		h.logger.Error("Stop deadline passed before the peering handler wrote "+
+			"its peerings", "paths", unwritten)
+	}
 }
 
 // processLoop runs periodically to process AS paths into peerings, and once
 // more on Stop.
 func (h *PeeringHandler) processLoop() {
-	defer h.wg.Done()
+	defer close(h.done)
 
 	ticker := time.NewTicker(peeringProcessInterval)
 	defer ticker.Stop()
@@ -152,9 +172,11 @@ func (h *PeeringHandler) processLoop() {
 		select {
 		case <-ticker.C:
 			h.processPeerings()
-		case <-h.stopCh:
+		case ctx := <-h.stopCh:
-			// Final processing
+			// Final processing, unless the stop deadline has passed
-			h.processPeerings()
+			if ctx.Err() == nil {
+				h.processPeerings()
+			}
 
 			return
 		}
@@ -169,6 +191,7 @@ func (h *PeeringHandler) processPeerings() {
 	h.mu.Lock()
 	paths := h.asPaths
 	h.asPaths = make(map[string]time.Time)
+	h.writing = len(paths)
 	dropped := h.droppedPaths
 	h.mu.Unlock()
 
@@ -234,6 +257,10 @@ func (h *PeeringHandler) processPeerings() {
 		}
 	}
 
+	h.mu.Lock()
+	h.writing = 0
+	h.mu.Unlock()
+
 	h.logger.Info("Processed AS peerings",
 		"paths", len(paths),
 		"unique_peerings", len(peerings),

internal/routewatch/peeringhandler_internal_test.go

@@ -53,7 +53,6 @@ func newTestHandler(db database.Store) *PeeringHandler {
 		db:      db,
 		logger:  logger.New(),
 		asPaths: make(map[string]time.Time),
-		stopCh:  make(chan struct{}),
 	}
 }
 
@@ -197,7 +196,7 @@ func TestPeeringHandlerStopWritesTheRemainingPaths(t *testing.T) {
 	h := NewPeeringHandler(store, logger.New())
 
 	handle(h, time.Now().UTC(), testASNA, testASNB)
-	h.Stop()
+	h.Stop(t.Context())
 
 	if got := store.callCount(); got != 1 {
 		t.Errorf("want 1 RecordPeering call when Stop returns, got %d", got)

internal/routewatch/prefixhandler.go

@@ -1,6 +1,7 @@
 package routewatch
 
 import (
+	"context"
 	"net"
 	"net/netip"
 	"slices"
@@ -50,9 +51,9 @@ type PrefixHandler struct {
 	logger  *logger.Logger
 	metrics *metrics.Tracker
 
-	// Batching. mu guards batch, held and peersDown only: the flush goroutine
+	// Batching. mu guards batch, held, peersDown, writing and deleting only:
-	// takes them under mu and writes them after releasing mu, so
+	// the flush goroutine takes them under mu and writes them after releasing
-	// HandleMessage waits for a database write only while batch is at
+	// mu, so HandleMessage waits for a database write only while batch is at
 	// prefixBatchMaxSize.
 	mu    sync.Mutex
 	batch []prefixUpdate
@@ -66,13 +67,21 @@ type PrefixHandler struct {
 	// since the batch was taken, and those whose route delete failed in the
 	// last flush.
 	peersDown []string
+	// writing is how many updates the flush goroutine took and has not
+	// finished writing.
+	writing int
+	// deleting is how many peers from peersDown the flush goroutine took and
+	// has not yet deleted the routes of.
+	deleting int
 	// batchTaken is signalled when the flush goroutine takes batch; its L is
 	// &mu.
 	batchTaken  sync.Cond
 	batchFullCh chan struct{} // wakes the flush goroutine when batch is full
 	lastFlush   time.Time     // used only by the flush goroutine
-	stopCh      chan struct{}
+	// stopCh hands Stop's context to the flush goroutine. It holds one, so
-	wg          sync.WaitGroup
+	// Stop does not wait for a write in progress to hand it over.
+	stopCh chan context.Context
+	done   chan struct{} // closed when the flush goroutine returns
 }
 
 type prefixUpdate struct {
@@ -95,12 +104,12 @@ func NewPrefixHandler(db database.Store, logger *logger.Logger) *PrefixHandler {
 		batch:       make([]prefixUpdate, 0, prefixBatchSize),
 		batchFullCh: make(chan struct{}, 1),
 		lastFlush:   time.Now(),
-		stopCh:      make(chan struct{}),
+		stopCh:      make(chan context.Context, 1),
+		done:        make(chan struct{}),
 	}
 	h.batchTaken.L = &h.mu
 
 	// Start the flush timer goroutine
-	h.wg.Add(1)
 	go h.flushLoop()
 
 	return h
@@ -202,10 +211,39 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 }
 
-// Stop gracefully stops the handler and flushes remaining batches
+// Stop stops the flush goroutine, which writes the updates and peer downs the
-func (h *PrefixHandler) Stop() {
+// handler holds before it returns, and waits for it until ctx ends. Once ctx
-	close(h.stopCh)
+// has ended the flush goroutine skips that last write, and a write still
-	h.wg.Wait()
+// running is lost when the process exits. A peer whose route delete fails in
+// the last write is not retried either. Stop logs how many updates are not
+// written and how many peer downs have not had their routes deleted.
+func (h *PrefixHandler) Stop(ctx context.Context) {
+	h.stopCh <- ctx
+
+	select {
+	case <-h.done:
+	case <-ctx.Done():
+	}
+
+	h.mu.Lock()
+	unwritten := h.writing + len(h.batch) + len(h.held)
+	peersDown := h.deleting + len(h.peersDown)
+	h.mu.Unlock()
+
+	if unwritten == 0 && peersDown == 0 {
+		return
+	}
+
+	if ctx.Err() != nil {
+		h.logger.Error("Stop deadline passed before the prefix handler wrote "+
+			"its updates", "updates", unwritten, "peers_down", peersDown)
+
+		return
+	}
+
+	h.logger.Error("The prefix handler stopped without deleting the routes "+
+		"of peers whose session went down", "updates", unwritten,
+		"peers_down", peersDown)
 }
 
 // peerSessionDown notes that the peer's BGP session went down, so that the next
@@ -230,7 +268,7 @@ func (h *PrefixHandler) peerSessionDown(peer string) {
 // flushLoop runs in a goroutine, the only one that writes the batch: it
 // flushes when the batch is full, periodically, and once more on Stop.
 func (h *PrefixHandler) flushLoop() {
-	defer h.wg.Done()
+	defer close(h.done)
 
 	ticker := time.NewTicker(prefixBatchTimeout)
 	defer ticker.Stop()
@@ -243,9 +281,11 @@ func (h *PrefixHandler) flushLoop() {
 			}
 		case <-h.batchFullCh:
 			h.flushBatch()
-		case <-h.stopCh:
+		case ctx := <-h.stopCh:
-			// Final flush
+			// Final flush, unless the stop deadline has passed
-			h.flushBatch()
+			if ctx.Err() == nil {
+				h.flushBatch()
+			}
 
 			return
 		}
@@ -254,7 +294,9 @@ func (h *PrefixHandler) flushLoop() {
 
 // takeBatch returns the pending updates, the held ones first, and the peers in
 // peersDown, and leaves an empty batch, no held updates and no peers in their
-// place for HandleMessage to fill.
+// place for HandleMessage to fill. The updates count in writing until
+// flushBatch has written them, and the peers in deleting until their routes
+// are deleted.
 func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -268,6 +310,8 @@ func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) {
 		batch = slices.Concat(h.held, h.batch)
 	}
 
+	h.writing = len(batch)
+	h.deleting = len(peersDown)
 	h.held = nil
 	h.batch = make([]prefixUpdate, 0, prefixBatchSize)
 	h.peersDown = nil
@@ -354,6 +398,10 @@ func (h *PrefixHandler) flushBatch() {
 
 	successCount := h.writeBatch(routesToUpsert, routesToDelete, prefixesToUpdate)
 
+	h.mu.Lock()
+	h.writing = 0
+	h.mu.Unlock()
+
 	elapsed := time.Since(startTime)
 	h.logger.Debug("Flushed prefix batch",
 		"batch_size", batchSize,
@@ -369,7 +417,8 @@ func (h *PrefixHandler) flushBatch() {
 // deleteLiveRoutesOfPeers deletes all live routes of each of peers, the peers
 // whose BGP session went down, and returns those whose delete failed. Most
 // downs come from peers that have no routes, so only a delete that removed
-// routes is logged.
+// routes is logged. A peer whose delete succeeded no longer counts in deleting;
+// a failed one counts until keepForNextFlush puts it back in peersDown.
 func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) []string {
 	var failed []string
 
@@ -386,6 +435,10 @@ func (h *PrefixHandler) deleteLiveRoutesOfPeers(peers []string) []string {
 			continue
 		}
 
+		h.mu.Lock()
+		h.deleting--
+		h.mu.Unlock()
+
 		if deleted > 0 {
 			h.logger.Info(
 				"Deleted the live routes of a peer whose session went down",
@@ -439,8 +492,12 @@ func (h *PrefixHandler) keepForNextFlush(
 		kept = slices.Clone(kept[len(kept)-prefixBatchSize:])
 	}
 
-	// takeBatch left held empty, and only this flush fills it.
+	// takeBatch left held empty, and only this flush fills it. The kept
+	// updates now count in held, not in writing, and the failed peers in
+	// peersDown, not in deleting.
 	h.held = kept
+	h.writing = len(toWrite)
+	h.deleting = 0
 
 	for _, peer := range failedPeers {
 		if !slices.Contains(h.peersDown, peer) {

internal/routewatch/prefixhandler_internal_test.go

@@ -1,11 +1,13 @@
 package routewatch
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
 	"net/netip"
 	"slices"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -766,6 +768,34 @@ func TestPrefixHandlerKeepsAtMostABatchOfUpdatesWhileADeleteKeepsFailing(
 	}
 }
 
+// TestPrefixHandlerStopLogsAFailedDeleteWithoutTheDeadline stops a prefix
+// handler, with no deadline, after a peer's session went down and the peer
+// announced a prefix, into a database whose every delete of a peer's routes
+// fails. Stop must log the peer down and the announcement as not written, and
+// must not say that the stop deadline passed.
+func TestPrefixHandlerStopLogsAFailedDeleteWithoutTheDeadline(t *testing.T) {
+	t.Parallel()
+
+	var logs bytes.Buffer
+
+	h := NewPrefixHandler(peerDeleteAlwaysFailsStore{Database: newTestDatabase(t)},
+		newErrorLogger(&logs))
+	h.HandleMessage(sessionDown())
+	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
+	h.Stop(t.Context())
+
+	logged := lastLogged(t, &logs)
+	if strings.Contains(logged.Msg, "deadline") {
+		t.Errorf("Stop logged %q although its deadline had not passed",
+			logged.Msg)
+	}
+
+	if logged.Updates != 1 || logged.PeersDown != 1 {
+		t.Errorf("Stop logged %d updates and %d peer downs not written, "+
+			"want 1 and 1", logged.Updates, logged.PeersDown)
+	}
+}
+
 // newTestDatabase opens a database in a new temporary directory. It is closed
 // when the test ends.
 func newTestDatabase(t *testing.T) *database.Database {

internal/server/handlers.go

@@ -41,6 +41,12 @@ const (
 	jsonStatusError = "error"
 )
 
+// Error messages an answer sends in place of the error that caused it.
+const (
+	invalidPrefixMessage   = "Invalid prefix"
+	hostNotResolvedMessage = "could not resolve host name"
+)
+
 // IP versions as database.LiveRoute.IPVersion records them.
 const (
 	ipVersionV4 = 4
@@ -68,7 +74,9 @@ func (s *Server) handleHealthCheck() http.HandlerFunc {
 		// Check database connectivity with lightweight ping
 		err := s.db.Ping(ctx)
 		if err != nil {
-			checks["database"] = "error: " + err.Error()
+			s.logger.Error("Health check database ping failed", "error", err)
+
+			checks["database"] = jsonStatusError
 			healthy = false
 		} else {
 			checks["database"] = "ok"
@@ -108,12 +116,12 @@ func (s *Server) handleHealthCheck() http.HandlerFunc {
 
 // handleIndex returns a handler that serves the home page.
 func (s *Server) handleIndex() http.HandlerFunc {
-	return func(w http.ResponseWriter, _ *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "text/html; charset=utf-8")
 
 		tmpl := templates.IndexTemplate()
 
-		err := tmpl.Execute(w, nil)
+		err := tmpl.Execute(w, pageData{CSPNonce: cspNonce(r.Context())})
 		if err != nil {
 			s.logger.Error("Failed to render index template", "error", err)
 			http.Error(w, "Internal Server Error", http.StatusInternalServerError)
@@ -121,6 +129,12 @@ func (s *Server) handleIndex() http.HandlerFunc {
 	}
 }
 
+// pageData is what the pages that show no data are rendered with: the nonce
+// their script and style blocks carry.
+type pageData struct {
+	CSPNonce string
+}
+
 // writeJSONError writes a standardized JSON error response with the given
 // status code and error message.
 func (s *Server) writeJSONError(w http.ResponseWriter, statusCode int, message string) {
@@ -251,7 +265,7 @@ func (s *Server) handleStatusJSON() http.HandlerFunc {
 
 		if err != nil {
 			s.logger.Error("Failed to get database stats", "error", err)
-			s.writeJSONError(w, http.StatusInternalServerError, err.Error())
+			s.writeJSONError(w, http.StatusInternalServerError, internalErrorMessage)
 
 			return
 		}
@@ -541,7 +555,7 @@ func (s *Server) handleStats() http.HandlerFunc {
 
 		if err != nil {
 			s.logger.Error("Failed to get database stats", "error", err)
-			s.writeJSONError(w, http.StatusInternalServerError, err.Error())
+			s.writeJSONError(w, http.StatusInternalServerError, internalErrorMessage)
 
 			return
 		}
@@ -622,12 +636,12 @@ func (s *Server) statsResponse(
 // handleStatusHTML returns a handler that serves the HTML status page,
 // which displays real-time statistics fetched via JavaScript.
 func (s *Server) handleStatusHTML() http.HandlerFunc {
-	return func(w http.ResponseWriter, _ *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "text/html; charset=utf-8")
 
 		tmpl := templates.StatusTemplate()
 
-		err := tmpl.Execute(w, nil)
+		err := tmpl.Execute(w, pageData{CSPNonce: cspNonce(r.Context())})
 		if err != nil {
 			s.logger.Error("Failed to render template", "error", err)
 			http.Error(w, "Internal Server Error", http.StatusInternalServerError)
@@ -667,7 +681,7 @@ func (s *Server) handleIPInfo() http.HandlerFunc {
 		}
 
 		// Collect all IPs to look up
-		ipsToLookup, resolveErrors := resolveLookupTargets(ctx, target)
+		ipsToLookup, resolveErrors := s.resolveLookupTargets(ctx, target)
 		if deadlinePassed(ctx) {
 			s.writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
 
@@ -695,7 +709,8 @@ func (s *Server) handleIPInfo() http.HandlerFunc {
 			}
 
 			if err != nil {
-				response.Errors = append(response.Errors, ip+": "+err.Error())
+				response.Errors = append(response.Errors,
+					ip+": "+s.lookupErrorMessage(ip, err))
 
 				continue
 			}
@@ -736,6 +751,19 @@ func (s *Server) handleIPInfo() http.HandlerFunc {
 	}
 }
 
+// lookupErrorMessage returns what an IP lookup answer says about err, the
+// failed lookup of ip: err's own text when ip is not a valid address or has no
+// route, otherwise a fixed message, with err in the log.
+func (s *Server) lookupErrorMessage(ip string, err error) string {
+	if errors.Is(err, database.ErrNoRoute) || errors.Is(err, database.ErrInvalidIP) {
+		return err.Error()
+	}
+
+	s.logger.Error("Failed to look up IP", "ip", ip, "error", err)
+
+	return internalErrorMessage
+}
+
 // lookupTarget returns what an IP lookup request asks for: the ip or addr URL
 // parameter, or else the client's own address.
 func lookupTarget(r *http.Request) string {
@@ -766,9 +794,12 @@ func (s *Server) queueWHOISRefresh(asns map[int]bool) {
 
 // resolveLookupTargets splits a comma-separated list of IP addresses and
 // hostnames and returns the IP addresses to look up, resolving each hostname.
-// A hostname that does not resolve is reported in the returned errors instead.
+// A hostname that does not resolve is reported in the returned errors instead,
+// with the resolver's own error only in the log, since it names the DNS server.
 // It stops once ctx's deadline has passed; the caller then answers 504.
-func resolveLookupTargets(ctx context.Context, target string) ([]string, []string) {
+func (s *Server) resolveLookupTargets(
+	ctx context.Context, target string,
+) ([]string, []string) {
 	var ipsToLookup, resolveErrors []string
 
 	for t := range strings.SplitSeq(target, ",") {
@@ -791,7 +822,8 @@ func resolveLookupTargets(ctx context.Context, target string) ([]string, []strin
 		}
 
 		if err != nil {
-			resolveErrors = append(resolveErrors, t+": "+err.Error())
+			s.logger.Warn("Host name lookup failed", "host", t, "error", err)
+			resolveErrors = append(resolveErrors, t+": "+hostNotResolvedMessage)
 
 			continue
 		}
@@ -819,7 +851,7 @@ func lookupPTRNames(ctx context.Context, ip string) []string {
 }
 
 // extractClientIP extracts the client IP from the request.
-// Works with chi's RealIP middleware which sets RemoteAddr.
+// Works with RealIPMiddleware, which sets RemoteAddr.
 func extractClientIP(r *http.Request) string {
 	// RemoteAddr is in the form "IP:port" or just "IP" for unix sockets
 	host, _, err := net.SplitHostPort(r.RemoteAddr)
@@ -854,7 +886,9 @@ func (s *Server) handleASDetailJSON() http.HandlerFunc {
 			if errors.Is(err, database.ErrNoRoute) {
 				s.writeJSONError(w, http.StatusNotFound, err.Error())
 			} else {
-				s.writeJSONError(w, http.StatusInternalServerError, err.Error())
+				s.logger.Error("Failed to get AS details", "asn", asn, "error", err)
+				s.writeJSONError(w, http.StatusInternalServerError,
+					internalErrorMessage)
 			}
 
 			return
@@ -933,7 +967,7 @@ func (s *Server) handlePrefixDetailJSON() http.HandlerFunc {
 		// tables store, so that any valid way of writing it finds the route.
 		parsed, err := netip.ParsePrefix(prefixParam + "/" + lenParam)
 		if err != nil {
-			s.writeJSONError(w, http.StatusBadRequest, err.Error())
+			s.writeJSONError(w, http.StatusBadRequest, invalidPrefixMessage)
 
 			return
 		}
@@ -951,7 +985,10 @@ func (s *Server) handlePrefixDetailJSON() http.HandlerFunc {
 			if errors.Is(err, database.ErrNoRoute) {
 				s.writeJSONError(w, http.StatusNotFound, err.Error())
 			} else {
-				s.writeJSONError(w, http.StatusInternalServerError, err.Error())
+				s.logger.Error("Failed to get prefix details",
+					"prefix", prefix, "error", err)
+				s.writeJSONError(w, http.StatusInternalServerError,
+					internalErrorMessage)
 			}
 
 			return
@@ -1002,7 +1039,7 @@ func (s *Server) handleASDetail() http.HandlerFunc {
 				http.Error(w, "AS not found", http.StatusNotFound)
 			} else {
 				s.logger.Error("Failed to get AS details", "error", err)
-				http.Error(w, "Internal server error", http.StatusInternalServerError)
+				http.Error(w, internalErrorMessage, http.StatusInternalServerError)
 			}
 
 			return
@@ -1039,6 +1076,7 @@ func (s *Server) handleASDetail() http.HandlerFunc {
 			IPv6Count    int
 			Peers        []database.ASPeer
 			PeerCount    int
+			CSPNonce     string
 		}{
 			ASN:          asInfo,
 			IPv4Prefixes: ipv4Prefixes,
@@ -1048,6 +1086,7 @@ func (s *Server) handleASDetail() http.HandlerFunc {
 			IPv6Count:    len(ipv6Prefixes),
 			Peers:        peers,
 			PeerCount:    len(peers),
+			CSPNonce:     cspNonce(r.Context()),
 		}
 
 		// Check if context is still valid before writing response
@@ -1058,9 +1097,7 @@ func (s *Server) handleASDetail() http.HandlerFunc {
 
 		w.Header().Set("Content-Type", "text/html; charset=utf-8")
 
-		tmpl := templates.ASDetailTemplate()
+		err = templates.ASDetailTemplate().Execute(w, data)
-
-		err = tmpl.Execute(w, data)
 		if err != nil {
 			s.logger.Error("Failed to render AS detail template", "error", err)
 			http.Error(w, "Internal Server Error", http.StatusInternalServerError)
@@ -1085,7 +1122,7 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
 		// tables store, so that any valid way of writing it finds the route.
 		parsed, err := netip.ParsePrefix(prefixParam + "/" + lenParam)
 		if err != nil {
-			http.Error(w, "Invalid prefix", http.StatusBadRequest)
+			http.Error(w, invalidPrefixMessage, http.StatusBadRequest)
 
 			return
 		}
@@ -1104,7 +1141,7 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
 				http.Error(w, "Prefix not found", http.StatusNotFound)
 			} else {
 				s.logger.Error("Failed to get prefix details", "error", err)
-				http.Error(w, "Internal server error", http.StatusInternalServerError)
+				http.Error(w, internalErrorMessage, http.StatusInternalServerError)
 			}
 
 			return
@@ -1131,6 +1168,7 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
 			Origins     []*ASNInfo
 			PeerCount   int
 			OriginCount int
+			CSPNonce    string
 		}{
 			Prefix:      prefix,
 			MaskLength:  maskLength,
@@ -1139,6 +1177,7 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
 			Origins:     origins,
 			PeerCount:   len(routes),
 			OriginCount: len(origins),
+			CSPNonce:    cspNonce(r.Context()),
 		}
 
 		// Check if context is still valid before writing response
@@ -1151,9 +1190,7 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
 
 		w.Header().Set("Content-Type", "text/html; charset=utf-8")
 
-		tmpl := templates.PrefixDetailTemplate()
+		err = templates.PrefixDetailTemplate().Execute(w, data)
-
-		err = tmpl.Execute(w, data)
 		if err != nil {
 			s.logger.Error("Failed to render prefix detail template", "error", err)
 			http.Error(w, "Internal Server Error", http.StatusInternalServerError)
@@ -1303,7 +1340,7 @@ func (s *Server) prefixLengthHandler(ipVersion, maxMaskLength int) http.HandlerF
 
 		if err != nil {
 			s.logger.Error("Failed to get prefixes by length", "error", err)
-			http.Error(w, "Internal server error", http.StatusInternalServerError)
+			http.Error(w, internalErrorMessage, http.StatusInternalServerError)
 
 			return
 		}
@@ -1324,6 +1361,7 @@ func (s *Server) prefixLengthHandler(ipVersion, maxMaskLength int) http.HandlerF
 			"IPVersion":  ipVersion,
 			"Prefixes":   prefixesWithASDescriptions(prefixes),
 			"Count":      len(prefixes),
+			"CSPNonce":   cspNonce(r.Context()),
 		}
 
 		// Check if context is still valid before writing response

internal/server/handlers_internal_test.go

@@ -28,6 +28,14 @@ const (
 	testPeerV6   = "2001:db8::1"
 )
 
+// Paths several tests request.
+const (
+	asPagePath           = "/as/64500"
+	prefixLengthPagePath = "/prefixlength/24"
+	statusJSONPath       = "/status.json"
+	statsPath            = "/api/v1/stats"
+)
+
 // The names the stats handler tests report the two stats handlers by.
 const (
 	statusJSONHandlerName = "status.json"
@@ -429,7 +437,7 @@ func TestPagesSayOriginZeroIsNoKnownOrigin(t *testing.T) {
 
 	s := New(db, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
 
-	for _, path := range []string{"/prefix/" + testPrefixV4, "/prefixlength/24"} {
+	for _, path := range []string{"/prefix/" + testPrefixV4, prefixLengthPagePath} {
 		rec := httptest.NewRecorder()
 		s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
 			t.Context(), http.MethodGet, path, nil))
@@ -610,13 +618,13 @@ func TestRoutesAnswer504WhenDatabaseReadIsCutOff(t *testing.T) {
 		path         string
 		wantJSONBody bool
 	}{
-		{path: "/as/64500"},
+		{path: asPagePath},
 		{path: "/prefix/192.0.2.0/24"},
-		{path: "/prefixlength/24"},
+		{path: prefixLengthPagePath},
 		{path: "/prefixlength6/48"},
-		{path: "/status.json", wantJSONBody: true},
+		{path: statusJSONPath, wantJSONBody: true},
 		{path: "/ip/192.0.2.1", wantJSONBody: true},
-		{path: "/api/v1/stats", wantJSONBody: true},
+		{path: statsPath, wantJSONBody: true},
 		{path: "/api/v1/ip/192.0.2.1", wantJSONBody: true},
 		{path: "/api/v1/as/64500", wantJSONBody: true},
 		{path: "/api/v1/prefix/192.0.2.0/24", wantJSONBody: true},
@@ -689,9 +697,9 @@ func TestRoutesAnswer504WhenSecondDatabaseReadIsCutOff(t *testing.T) {
 		path         string
 		wantJSONBody bool
 	}{
-		{path: "/as/64500"},
+		{path: asPagePath},
-		{path: "/status.json", wantJSONBody: true},
+		{path: statusJSONPath, wantJSONBody: true},
-		{path: "/api/v1/stats", wantJSONBody: true},
+		{path: statsPath, wantJSONBody: true},
 	}
 
 	for _, tt := range tests {
@@ -758,6 +766,86 @@ func checkJSONErrorCode(
 	}
 }
 
+// errDiskIO is the database driver's error failingDB answers with.
+var errDiskIO = errors.New("disk I/O error")
+
+// failingDB embeds database.Store (left nil) and fails the reads of the JSON
+// routes and the health check's ping with errDiskIO. Every other Store method
+// is unused and would panic if called.
+type failingDB struct {
+	database.Store
+}
+
+func (failingDB) Ping(_ context.Context) error {
+	return errDiskIO
+}
+
+func (failingDB) GetStatsContext(_ context.Context) (database.Stats, error) {
+	return database.Stats{}, errDiskIO
+}
+
+func (failingDB) GetIPInfoContext(
+	_ context.Context, _ string,
+) (*database.IPInfo, error) {
+	return nil, errDiskIO
+}
+
+func (failingDB) GetASDetailsContext(
+	_ context.Context, _ int,
+) (*database.ASN, []database.LiveRoute, error) {
+	return nil, nil, errDiskIO
+}
+
+func (failingDB) GetPrefixDetailsContext(
+	_ context.Context, _ string,
+) ([]database.LiveRoute, map[int]database.ASN, error) {
+	return nil, nil, errDiskIO
+}
+
+// TestJSONRoutesDoNotSendErrorDetails sends a request to each JSON route and
+// the health check with a store whose reads fail with the driver's error text,
+// and a prefix that does not parse, and checks that no answer carries the
+// driver's or the prefix parser's text. Before the fix the answers sent it to
+// the client (https://git.eeqj.de/sneak/routewatch/issues/108). The IP lookup
+// answers 404, as it does whenever no address could be looked up.
+func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) {
+	t.Parallel()
+
+	s := New(failingDB{}, streamer.New(logger.New(), metrics.New()), logger.New(),
+		&config.Config{})
+
+	tests := []struct {
+		path     string
+		wantCode int
+	}{
+		{"/.well-known/healthcheck.json", http.StatusServiceUnavailable},
+		{statusJSONPath, http.StatusInternalServerError},
+		{statsPath, http.StatusInternalServerError},
+		{"/api/v1/as/64500", http.StatusInternalServerError},
+		{"/api/v1/prefix/192.0.2.0/24", http.StatusInternalServerError},
+		{"/api/v1/ip/192.0.2.1", http.StatusNotFound},
+		{"/api/v1/prefix/foo/24", http.StatusBadRequest},
+	}
+
+	for _, tt := range tests {
+		rec := httptest.NewRecorder()
+		s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
+			t.Context(), http.MethodGet, tt.path, nil))
+
+		if rec.Code != tt.wantCode {
+			t.Errorf("%s: status %d, want %d; body %s",
+				tt.path, rec.Code, tt.wantCode, rec.Body)
+		}
+
+		for _, detail := range []string{errDiskIO.Error(), "netip"} {
+			if strings.Contains(rec.Body.String(), detail) {
+				t.Errorf("%s: body %s carries the error text %q",
+					tt.path, rec.Body, detail)
+			}
+		}
+	}
+}
+
 // settledGoroutineCount lets transient goroutines finish, then reports the
 // current count.
 func settledGoroutineCount() int {

internal/server/middleware.go

@@ -3,11 +3,15 @@ package server
 import (
 	"bytes"
 	"context"
+	"crypto/rand"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
 	"net/http"
+	"net/netip"
+	"slices"
+	"strings"
 	"sync"
 	"time"
 
@@ -118,6 +122,10 @@ func JSONResponseMiddleware(next http.Handler) http.Handler {
 // requestTimeoutMessage is the error message sent with 504 Gateway Timeout.
 const requestTimeoutMessage = "Request timeout"
 
+// internalErrorMessage is the error message sent with 500 Internal Server
+// Error, whatever went wrong; the handler logs the detail.
+const internalErrorMessage = "Internal server error"
+
 // deadlinePassed reports whether ctx ended because its deadline passed.
 //
 // A handler calls it right after each database read or host name lookup,
@@ -194,7 +202,7 @@ func JSONValidationMiddleware(next http.Handler) http.Handler {
 
 		w.WriteHeader(statusCode)
 
-		errorMsg := "Internal server error"
+		errorMsg := internalErrorMessage
 		if statusCode == http.StatusGatewayTimeout {
 			errorMsg = requestTimeoutMessage
 		}
@@ -296,3 +304,105 @@ func DecodedPathMiddleware(next http.Handler) http.Handler {
 		next.ServeHTTP(w, r)
 	})
 }
+
+// The values of the headers SecurityHeadersMiddleware sets.
+const (
+	// strictTransportSecurity tells browsers to reach this host and its
+	// subdomains only over HTTPS for a year. The server listens on plain
+	// HTTP behind a TLS-terminating proxy, which passes the header on.
+	strictTransportSecurity = "max-age=31536000; includeSubDomains"
+
+	// contentSecurityPolicy lets a page load content only from this server,
+	// and run or apply an inline script or style block only when the block
+	// carries the nonce made for that answer, which replaces %[1]s. No other
+	// site may frame the pages.
+	contentSecurityPolicy = "default-src 'self'; " +
+		"script-src 'nonce-%[1]s'; style-src 'nonce-%[1]s'; " +
+		"frame-ancestors 'none'; base-uri 'none'; form-action 'self'"
+
+	frameOptions       = "DENY"
+	contentTypeOptions = "nosniff"
+	referrerPolicy     = "strict-origin-when-cross-origin"
+
+	// permissionsPolicy turns off browser features the pages do not use.
+	permissionsPolicy = "accelerometer=(), camera=(), geolocation=(), " +
+		"gyroscope=(), magnetometer=(), microphone=(), payment=(), usb=()"
+)
+
+// cspNonceKey is the request context key of the nonce SecurityHeadersMiddleware
+// made for the answer.
+type cspNonceKey struct{}
+
+// SecurityHeadersMiddleware sets the security headers REPO_POLICIES.md requires
+// on every answer. Each answer's Content-Security-Policy names a nonce of its
+// own; a page reads it with cspNonce and writes it on its script and style
+// blocks.
+func SecurityHeadersMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		nonce := rand.Text()
+
+		header := w.Header()
+		header.Set("Strict-Transport-Security", strictTransportSecurity)
+		header.Set("Content-Security-Policy",
+			fmt.Sprintf(contentSecurityPolicy, nonce))
+		header.Set("X-Frame-Options", frameOptions)
+		header.Set("X-Content-Type-Options", contentTypeOptions)
+		header.Set("Referrer-Policy", referrerPolicy)
+		header.Set("Permissions-Policy", permissionsPolicy)
+
+		ctx := context.WithValue(r.Context(), cspNonceKey{}, nonce)
+		next.ServeHTTP(w, r.WithContext(ctx))
+	})
+}
+
+// cspNonce returns the nonce SecurityHeadersMiddleware made for the answer to
+// the request with ctx.
+func cspNonce(ctx context.Context) string {
+	nonce, _ := ctx.Value(cspNonceKey{}).(string)
+
+	return nonce
+}
+
+// RealIPMiddleware gives a request from one of trustedProxies the address of
+// the client the proxies forwarded it for as its RemoteAddr: the last
+// X-Forwarded-For entry that is not a trusted proxy. Each proxy appends the
+// address it took the request from, so the entries left of that one are
+// whatever the client sent. A request from any other peer keeps the peer's
+// address, whatever headers it carries.
+func RealIPMiddleware(trustedProxies []netip.Prefix) func(http.Handler) http.Handler {
+	isTrusted := func(addr netip.Addr) bool {
+		return slices.ContainsFunc(trustedProxies, func(proxy netip.Prefix) bool {
+			return proxy.Contains(addr.Unmap())
+		})
+	}
+
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			peer, err := netip.ParseAddrPort(r.RemoteAddr)
+			if err != nil || !isTrusted(peer.Addr()) {
+				next.ServeHTTP(w, r)
+
+				return
+			}
+
+			// A header sent on several lines is one list.
+			entries := strings.Split(
+				strings.Join(r.Header.Values("X-Forwarded-For"), ","), ",")
+
+			for _, entry := range slices.Backward(entries) {
+				addr, err := netip.ParseAddr(strings.TrimSpace(entry))
+				if err != nil {
+					break
+				}
+
+				r.RemoteAddr = addr.String()
+
+				if !isTrusted(addr) {
+					break
+				}
+			}
+
+			next.ServeHTTP(w, r)
+		})
+	}
+}

internal/server/middleware_internal_test.go

@@ -0,0 +1,205 @@
+package server
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/netip"
+	"strings"
+	"testing"
+	"time"
+)
+
+// The addresses the forwarding tests use: the client a proxy names in
+// X-Forwarded-For, a peer inside the trusted range of
+// TestRealIPMiddlewareTrustsOnlyConfiguredProxies, and a peer outside it,
+// which lies in testPrefixV4 so that an IP lookup of it finds a route.
+const (
+	forwardedClient   = "203.0.113.9"
+	trustedPeer       = "10.0.0.2:1234"
+	untrustedPeerAddr = "192.0.2.1"
+	untrustedPeer     = untrustedPeerAddr + ":1234"
+)
+
+// TestResponsesCarrySecurityHeaders checks that a page, a JSON route and a
+// path no route matches each answer with the security headers REPO_POLICIES.md
+// requires, and that the Content-Security-Policy allows no inline content
+// without a nonce (https://git.eeqj.de/sneak/routewatch/issues/108).
+func TestResponsesCarrySecurityHeaders(t *testing.T) {
+	t.Parallel()
+
+	s := newServerWithTestRoutes(t)
+
+	headers := []string{
+		"Strict-Transport-Security",
+		"Content-Security-Policy",
+		"X-Frame-Options",
+		"X-Content-Type-Options",
+		"Referrer-Policy",
+		"Permissions-Policy",
+	}
+
+	for _, path := range []string{"/", statsPath, "/no-such-page"} {
+		rec := httptest.NewRecorder()
+		s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
+			t.Context(), http.MethodGet, path, nil))
+
+		for _, name := range headers {
+			if rec.Header().Get(name) == "" {
+				t.Errorf("%s: no %s header", path, name)
+			}
+		}
+
+		policy := rec.Header().Get("Content-Security-Policy")
+		if strings.Contains(policy, "'unsafe-") {
+			t.Errorf("%s: Content-Security-Policy %q allows unsafe content",
+				path, policy)
+		}
+	}
+}
+
+// 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
+// that no page uses a style attribute, which the policy does not allow.
+func TestPagesCarryTheCSPNonce(t *testing.T) {
+	t.Parallel()
+
+	s := newServerWithTestRoutes(t)
+
+	_, err := s.db.GetOrCreateASN(64500, time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC))
+	if err != nil {
+		t.Fatalf("GetOrCreateASN: %v", err)
+	}
+
+	paths := []string{
+		"/", "/status", asPagePath, "/prefix/" + testPrefixV4,
+		prefixLengthPagePath, "/prefixlength6/32",
+	}
+	seen := make(map[string]bool)
+
+	for _, path := range paths {
+		rec := httptest.NewRecorder()
+		s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
+			t.Context(), http.MethodGet, path, nil))
+
+		if rec.Code != http.StatusOK {
+			t.Errorf("%s: status %d, want %d; body %s",
+				path, rec.Code, http.StatusOK, rec.Body)
+
+			continue
+		}
+
+		policy := rec.Header().Get("Content-Security-Policy")
+
+		_, nonce, found := strings.Cut(policy, "'nonce-")
+		if !found {
+			t.Errorf("%s: Content-Security-Policy %q names no nonce", path, policy)
+
+			continue
+		}
+
+		nonce, _, _ = strings.Cut(nonce, "'")
+		if seen[nonce] {
+			t.Errorf("%s: nonce %q was already used for another answer", path, nonce)
+		}
+
+		seen[nonce] = true
+		page := rec.Body.String()
+
+		for _, tag := range []string{"<script", "<style"} {
+			withNonce := tag + ` nonce="` + nonce + `"`
+			if strings.Count(page, tag) != strings.Count(page, withNonce) {
+				t.Errorf("%s: a %s tag does not carry the nonce %q", path, tag, nonce)
+			}
+		}
+
+		if strings.Contains(page, "style=") {
+			t.Errorf("%s: the page uses a style attribute", path)
+		}
+	}
+}
+
+// TestForwardedForFromUntrustedPeerIsIgnored asks /ip/ for the caller's own
+// address from a peer that is not a trusted proxy, with forwarding headers
+// naming another address, and checks that the lookup is for the TCP peer.
+// Before the fix any client could choose the address it was looked up and
+// logged as (https://git.eeqj.de/sneak/routewatch/issues/108).
+func TestForwardedForFromUntrustedPeerIsIgnored(t *testing.T) {
+	t.Parallel()
+
+	s := newServerWithTestRoutes(t)
+
+	req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/ip/", nil)
+	req.RemoteAddr = untrustedPeer
+	req.Header.Set("X-Forwarded-For", forwardedClient)
+	req.Header.Set("X-Real-IP", forwardedClient)
+	req.Header.Set("True-Client-IP", forwardedClient)
+
+	rec := httptest.NewRecorder()
+	s.router.ServeHTTP(rec, req)
+
+	var body struct {
+		Data IPLookupResponse `json:"data"`
+	}
+
+	err := json.Unmarshal(rec.Body.Bytes(), &body)
+	if err != nil {
+		t.Fatalf("decoding the answer %q: %v", rec.Body, err)
+	}
+
+	if body.Data.Query != untrustedPeerAddr {
+		t.Errorf("looked up %q, want the TCP peer %s; answer %s",
+			body.Data.Query, untrustedPeerAddr, rec.Body)
+	}
+}
+
+// TestRealIPMiddlewareTrustsOnlyConfiguredProxies checks which address
+// RealIPMiddleware gives a request: the TCP peer, unless the peer is a trusted
+// proxy; then the last X-Forwarded-For entry that is not a trusted proxy,
+// since the entries left of it are whatever the client sent.
+func TestRealIPMiddlewareTrustsOnlyConfiguredProxies(t *testing.T) {
+	t.Parallel()
+
+	trusted := []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")}
+
+	tests := []struct {
+		name         string
+		peer         string
+		forwardedFor string
+		want         string
+	}{
+		{"untrusted peer", untrustedPeer, forwardedClient, untrustedPeer},
+		{"trusted peer", trustedPeer, forwardedClient, forwardedClient},
+		{
+			"entries the client wrote", trustedPeer,
+			"198.51.100.7, " + forwardedClient, forwardedClient,
+		},
+		{
+			"two trusted proxies", trustedPeer,
+			forwardedClient + ", 10.0.0.3", forwardedClient,
+		},
+		{"no header", trustedPeer, "", trustedPeer},
+		{"entry not an address", trustedPeer, "client", trustedPeer},
+	}
+
+	for _, tt := range tests {
+		var got string
+
+		handler := RealIPMiddleware(trusted)(http.HandlerFunc(
+			func(_ http.ResponseWriter, r *http.Request) { got = r.RemoteAddr }))
+
+		req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil)
+		req.RemoteAddr = tt.peer
+
+		if tt.forwardedFor != "" {
+			req.Header.Set("X-Forwarded-For", tt.forwardedFor)
+		}
+
+		handler.ServeHTTP(httptest.NewRecorder(), req)
+
+		if got != tt.want {
+			t.Errorf("%s: RemoteAddr %q, want %q", tt.name, got, tt.want)
+		}
+	}
+}

internal/server/routes.go

@@ -13,8 +13,9 @@ func (s *Server) setupRoutes() {
 
 	// Middleware
 	r.Use(DecodedPathMiddleware)
+	r.Use(SecurityHeadersMiddleware)
 	r.Use(middleware.RequestID)
-	r.Use(middleware.RealIP)
+	r.Use(RealIPMiddleware(s.trustedProxies))
 	r.Use(RequestLoggerMiddleware(s.logger.Logger)) // Structured request logging
 	r.Use(middleware.Recoverer)
 	r.Use(JSONResponseMiddleware)

internal/server/server.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"net/netip"
 	"strconv"
 	"time"
 
@@ -40,6 +41,10 @@ type Server struct {
 	port       int
 	srv        *http.Server
 	asnFetcher ASNFetcher
+
+	// trustedProxies are the reverse proxies whose X-Forwarded-For header
+	// RealIPMiddleware believes
+	trustedProxies []netip.Prefix
 }
 
 // New creates a new HTTP server
@@ -50,10 +55,11 @@ func New(
 	cfg *config.Config,
 ) *Server {
 	s := &Server{
-		db:       db,
+		db:             db,
-		streamer: streamer,
+		streamer:       streamer,
-		logger:   logger,
+		logger:         logger,
-		port:     cfg.Port,
+		port:           cfg.Port,
+		trustedProxies: cfg.TrustedProxies,
 	}
 
 	s.setupRoutes()
@@ -103,7 +109,9 @@ func (s *Server) Start(ctx context.Context) error {
 	return nil
 }
 
-// Stop gracefully stops the HTTP server
+// Stop stops the HTTP server. It waits, until ctx ends, for the requests in
+// progress to finish, and then closes the connections of those still running
+// and returns ctx's error.
 func (s *Server) Stop(ctx context.Context) error {
 	if s.srv == nil {
 		return nil
@@ -111,7 +119,12 @@ func (s *Server) Stop(ctx context.Context) error {
 
 	s.logger.Info("Stopping HTTP server")
 
-	return s.srv.Shutdown(ctx)
+	err := s.srv.Shutdown(ctx)
+	if err != nil {
+		return errors.Join(err, s.srv.Close())
+	}
+
+	return nil
 }
 
 // SetASNFetcher sets the ASN WHOIS fetcher for on-demand lookups.

internal/server/server_internal_test.go

@@ -0,0 +1,92 @@
+package server
+
+import (
+	"context"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/logger"
+)
+
+// TestStopClosesTheConnectionOfARequestStillRunning serves a request whose
+// handler does not return until the test ends, and stops the server with a
+// deadline that passes while the request runs. The client must then see its
+// connection closed. Before the fix Stop returned at the deadline and left the
+// connection open until the handler returned.
+func TestStopClosesTheConnectionOfARequestStillRunning(t *testing.T) {
+	t.Parallel()
+
+	const (
+		stopDeadline = 100 * time.Millisecond
+		// waitTimeout is how long the test waits for the request to reach
+		// the handler, and then for the client to see its connection closed.
+		waitTimeout = 2 * time.Second
+		// readHeaderTimeout bounds the test server's read of the request
+		// header, as every http.Server must.
+		readHeaderTimeout = time.Second
+	)
+
+	release := make(chan struct{})
+	defer close(release)
+
+	started := make(chan struct{})
+
+	var listenConfig net.ListenConfig
+
+	listener, err := listenConfig.Listen(t.Context(), "tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("failed to open the test listener: %v", err)
+	}
+
+	s := &Server{logger: logger.New(), srv: &http.Server{
+		Handler: http.HandlerFunc(func(http.ResponseWriter, *http.Request) {
+			close(started)
+			<-release
+		}),
+		ReadHeaderTimeout: readHeaderTimeout,
+	}}
+
+	go func() { _ = s.srv.Serve(listener) }()
+
+	answered := make(chan error, 1)
+
+	go func() {
+		req, err := http.NewRequestWithContext(t.Context(), http.MethodGet,
+			"http://"+listener.Addr().String()+"/", nil)
+		if err != nil {
+			answered <- err
+
+			return
+		}
+
+		resp, err := http.DefaultClient.Do(req)
+		if err == nil {
+			_ = resp.Body.Close()
+		}
+
+		answered <- err
+	}()
+
+	select {
+	case <-started:
+	case <-time.After(waitTimeout):
+		t.Fatal("the request did not reach the handler")
+	}
+
+	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
+	defer cancel()
+
+	_ = s.Stop(ctx)
+
+	select {
+	case err := <-answered:
+		if err == nil {
+			t.Error("the request still running at the deadline was answered")
+		}
+	case <-time.After(waitTimeout):
+		t.Fatal("the connection of the request still running at the " +
+			"deadline was open after Stop returned")
+	}
+}

internal/streamer/streamer.go

@@ -253,9 +253,10 @@ func (s *Streamer) Start(ctx context.Context) error {
 
 // Stop ends the stream and stops the handler workers. It returns once each
 // worker has handed its handler every message left on its queue and then the
-// STATE messages held for it (see queueMessage). This method is safe to call
+// STATE messages held for it (see queueMessage), or once ctx ends, and then
+// logs how many messages were not handed over. This method is safe to call
 // multiple times.
-func (s *Streamer) Stop() {
+func (s *Streamer) Stop(ctx context.Context) {
 	s.mu.Lock()
 	if s.cancel == nil {
 		// Not started, or already stopped: closing the queues again would panic.
@@ -264,28 +265,56 @@ func (s *Streamer) Stop() {
 		return
 	}
 
-	// Once ctx is cancelled the read loop queues and holds nothing more (see
+	// Once the stream's context is cancelled the read loop queues and holds
-	// dispatchMessage).
+	// nothing more (see dispatchMessage).
 	s.cancel()
 	s.cancel = nil
 
+	notHandedOver := 0
+
 	for _, info := range s.handlers {
 		// The held messages came after everything on the queue. Each waits
-		// for the worker, which does not take s.mu, to make room.
+		// for the worker, which does not take s.mu, to make room, until ctx
-		for _, msg := range info.heldState {
+		// ends.
-			info.queue <- msg
+		for len(info.heldState) > 0 && ctx.Err() == nil {
+			select {
+			case info.queue <- info.heldState[0]:
+				info.heldState = info.heldState[1:]
+			case <-ctx.Done():
+			}
 		}
 
+		notHandedOver += len(info.heldState)
 		info.heldState = nil
 
 		// The worker returns once it has emptied the closed queue.
 		close(info.queue)
 	}
 
+	handlers := s.handlers
 	s.running = false
 	s.mu.Unlock()
 	s.metrics.SetConnected(false)
-	s.workers.Wait()
+
+	workersDone := make(chan struct{})
+
+	go func() {
+		s.workers.Wait()
+		close(workersDone)
+	}()
+
+	select {
+	case <-workersDone:
+	case <-ctx.Done():
+		for _, info := range handlers {
+			notHandedOver += len(info.queue)
+		}
+	}
+
+	if notHandedOver > 0 {
+		s.logger.Error("Stop deadline passed before the handlers took every "+
+			"message", "messages", notHandedOver)
+	}
 }
 
 // IsRunning reports whether the streamer is currently connected and processing

internal/streamer/streamer_internal_test.go

@@ -377,7 +377,7 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
 	stopped := make(chan struct{})
 
 	go func() {
-		s.Stop()
+		s.Stop(t.Context())
 		close(stopped)
 	}()
 
@@ -412,6 +412,48 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
 	}
 }
 
+// TestStopReturnsByItsDeadline stops a streamer whose handler has a full queue
+// and a STATE message held for it, and a worker that stays busy until the test
+// ends, as one does whose handler waits for a database write. Stop must return
+// within its deadline plus a small margin instead of waiting for room on the
+// queue and for the worker.
+func TestStopReturnsByItsDeadline(t *testing.T) {
+	t.Parallel()
+
+	const (
+		deadline = 100 * time.Millisecond
+		margin   = 2 * time.Second
+	)
+
+	s := newStreamerWithFullStateQueue("")
+	s.handlers[0].heldState = []*ristypes.RISMessage{
+		{Type: ristypes.MessageTypeState},
+	}
+	// Stop cancels the stream Start would have started; there is none here.
+	s.cancel = func() {}
+
+	release := make(chan struct{})
+	defer close(release)
+
+	s.workers.Go(func() { <-release })
+
+	ctx, cancel := context.WithTimeout(t.Context(), deadline)
+	defer cancel()
+
+	stopped := make(chan struct{})
+
+	go func() {
+		s.Stop(ctx)
+		close(stopped)
+	}()
+
+	select {
+	case <-stopped:
+	case <-time.After(deadline + margin):
+		t.Fatalf("Stop did not return within %s of its deadline", margin)
+	}
+}
+
 // 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
@@ -429,7 +471,7 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
 	s := New(logger.New(), metrics.New())
 	s.url = srv.URL
 	s.RegisterHandler(updateHandler{})
-	s.RegisterRawHandler(func(string) { s.Stop() })
+	s.RegisterRawHandler(func(string) { s.Stop(t.Context()) })
 
 	// 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
@@ -443,7 +485,7 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
 	}
 
 	// A second Stop must not close the queues again.
-	s.Stop()
+	s.Stop(t.Context())
 }
 
 // TestReconnectWhenStreamEnds has the server end the first response
@@ -676,7 +718,7 @@ func checkReconnect(
 	if err != nil {
 		t.Fatalf("Start returned error: %v", err)
 	}
-	defer s.Stop()
+	defer s.Stop(t.Context())
 
 	select {
 	case <-handler.handled:

internal/templates/as_detail.html

@@ -4,7 +4,7 @@
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
     <title>AS{{.ASN.ASN}} - {{.ASN.Handle}} - RouteWatch</title>
-    <style>
+    <style nonce="{{.CSPNonce}}">
         * {
             box-sizing: border-box;
         }

internal/templates/index.html

@@ -4,7 +4,7 @@
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
     <title>RouteWatch - BGP Route Monitor</title>
-    <style>
+    <style nonce="{{.CSPNonce}}">
         * {
             box-sizing: border-box;
         }
@@ -167,6 +167,13 @@
             color: #95a5a6;
             margin-top: 8px;
         }
+        #asname-results p {
+            color: #7f8c8d;
+            margin-top: 12px;
+        }
+        #asname-results .search-message {
+            font-size: 13px;
+        }
 
         /* IP Lookup result */
         .ip-result {
@@ -329,7 +336,7 @@
         <span><a href="{{appGitCommitURL}}">{{appGitRevision}}</a></span>
     </footer>
 
-    <script>
+    <script nonce="{{.CSPNonce}}">
         function formatNumber(num) {
             if (num >= 1000000) {
                 return (num / 1000000).toFixed(1) + 'M';
@@ -387,7 +394,7 @@
             if (!query) return;
 
             const resultsDiv = document.getElementById('asname-results');
-            resultsDiv.innerHTML = '<p class="loading" style="color: #7f8c8d; margin-top: 12px;">Searching...</p>';
+            resultsDiv.innerHTML = '<p class="loading">Searching...</p>';
 
             // Use a simple client-side search against the asinfo data
             // For now, redirect to AS page if it looks like an ASN
@@ -397,7 +404,7 @@
             }
 
             // Show a message that server-side search is coming
-            resultsDiv.innerHTML = '<p style="color: #7f8c8d; margin-top: 12px; font-size: 13px;">AS name search coming soon. For now, try an AS number.</p>';
+            resultsDiv.innerHTML = '<p class="search-message">AS name search coming soon. For now, try an AS number.</p>';
         });
 
         // IP lookup

internal/templates/prefix_detail.html

@@ -4,7 +4,7 @@
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
     <title>{{.Prefix}} - RouteWatch</title>
-    <style>
+    <style nonce="{{.CSPNonce}}">
         * {
             box-sizing: border-box;
         }
@@ -221,6 +221,10 @@
             border-radius: 4px;
             border: 1px solid #e0e0e0;
         }
+        .origin-item .peer-count {
+            color: #7f8c8d;
+            margin-left: 10px;
+        }
         .empty-state {
             text-align: center;
             padding: 40px;
@@ -288,7 +292,7 @@
                     {{else}}
                     No known origin
                     {{end}}
-                    <span style="color: #7f8c8d; margin-left: 10px;">{{.PeerCount}} peer{{if ne .PeerCount 1}}s{{end}}</span>
+                    <span class="peer-count">{{.PeerCount}} peer{{if ne .PeerCount 1}}s{{end}}</span>
                 </div>
                 {{end}}
             </div>

internal/templates/prefix_length.html

@@ -4,7 +4,7 @@
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
     <title>Prefixes with /{{ .MaskLength }} - RouteWatch</title>
-    <style>
+    <style nonce="{{.CSPNonce}}">
         * {
             box-sizing: border-box;
         }

internal/templates/status.html

@@ -4,7 +4,7 @@
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
     <title>RouteWatch Status</title>
-    <style>
+    <style nonce="{{.CSPNonce}}">
         * {
             box-sizing: border-box;
         }
@@ -130,6 +130,7 @@
             color: #ef4444;
         }
         .error {
+            display: none;
             background: #fee;
             color: #c00;
             padding: 10px;
@@ -173,7 +174,7 @@
     </nav>
 
     <main class="main-content">
-    <div id="error" class="error" style="display: none;"></div>
+    <div id="error" class="error"></div>
     <div class="status-grid">
         <div class="status-card">
             <h2>RouteWatch</h2>
@@ -383,8 +384,8 @@
     <div id="handler-stats-container" class="status-grid">
         <!-- Handler stats will be dynamically added here -->
     </div>
-    
+
-    <script>
+    <script nonce="{{.CSPNonce}}">
         function formatBytes(bytes) {
             if (bytes === 0) return '0 B';
             const k = 1024;