daemon: forget peers nobody uses, so agents stop growing until they run out of memory

CHANGELOG.md

@@ -341,6 +341,56 @@ Detailed per-release notes are on the
 
   A `--data` over 65535 bytes is now refused before the daemon is
   contacted.
+- **Peers nobody is using no longer cost anything.** The daemon kept every
+  peer it had ever exchanged keys with warm forever: a NAT keepalive every
+  25 s, path-watchdog probes and resets when the peer went quiet, and for
+  each relayed peer a direct-upgrade attempt (registry lookup, beacon punch,
+  five probes) every 15 s. A public service agent that had answered a few
+  thousand clients sent about 1,600 packets a second on this while serving
+  a few requests a minute. A peer with no application traffic for 2 minutes
+  and no open connection now gets none of it. Sending to it works as
+  before, re-establishing the path through the usual fallbacks if a NAT
+  mapping expired meanwhile. Five minutes after the last frame either way,
+  the stale-peer reaper now drops the peer, as it was meant to: the
+  keepalives counted as contact, so it never fired, and busy nodes held
+  every peer they had ever seen (service agents: 3,500 to 10,900). The next
+  contact runs a fresh key exchange. The reaper also counts relayed inbound
+  frames as contact now, so a peer still sending to us is kept.
+- **A reaped peer leaves nothing behind, and frames for a peer that never
+  answers no longer wait forever.** Frames queued for a key exchange were
+  dropped only when the exchange completed; for a peer that never answered
+  they stayed until the daemon exited, reaper or not. Once 256 such peers
+  had piled up, every first send to a new peer failed with "too many
+  pending key exchanges". A queue now goes after 2 minutes (the exchange
+  itself gives up after 20 s), and with its peer when the reaper drops it.
+- **A rekey that gives up no longer resets a peer nobody is using.** The
+  reset starts a new key exchange; for a client that had gone away it gave
+  up again, and the next reset followed, for as long as the daemon ran.
+  With many such peers waiting, the cap of 64 key-exchange retransmits per
+  4 s stretched each round past the 30 s reset cooldown, so nothing broke
+  the cycle: on 2026-10-08 one service agent logged 25,147 of these resets
+  in 90 minutes, each a registry lookup and a new key exchange, and the
+  key-exchange frames counted as contact, so the reaper never dropped those
+  peers. The reset now runs only for a peer with an open connection or
+  application traffic in the last 2 minutes. A path reset also keeps the
+  record of when the peer was last used.
+- **State left behind for a node with no tunnel is reaped too.** A path
+  reset that could not re-resolve its peer left the peer's relay flag (and
+  whatever else it had) with no tunnel entry, which the stale-peer reaper,
+  walking tunnel peers, never visited: the flag kept the node in the 15 s
+  direct-upgrade loop and counted against the 4,096 relay peers a daemon
+  admits, for as long as the daemon ran. Such state now goes 5 minutes
+  after the last frame with the node, and the upgrade loop skips nodes with
+  no tunnel.
+- **A slow registry no longer piles up goroutines.** Every 15 s the daemon
+  tries to move each relayed peer to a direct path, and an attempt that
+  must resolve the peer waits for the registry. It gave up waiting after
+  8 s but left the call running, and the next round started another, so
+  while the registry was slow the waiting calls grew by one per relayed
+  peer every 15 s. A local soak test reached 21,000 goroutines, and since
+  Go keeps a goroutine's descriptor for the life of the process, the heap
+  stayed ~30 MB higher afterwards. A peer now has at most one attempt
+  running.
 - **The tunnel socket asks the kernel for 4 MB buffers** in each direction
   instead of the default (about 200 KB on Linux). Every tunnel shares the one
   socket, and several streams sending at once overflowed it. The kernel caps

cmd/daemon/main.go

@@ -116,6 +116,7 @@ func main() {
 	compatBeacon := flag.String("compat-beacon", defaultCompatBeacon, "beacon WSS URL for -transport=compat")
 	tlsTrust := flag.String("tls-trust", "system", "TLS trust store for -transport=compat: 'system' (OS trust store; current default while compat mode uses Let's Encrypt certs on beacon.pilotprotocol.network — on a host without a CA bundle set SSL_CERT_FILE or SSL_CERT_DIR) or 'pinned' (Pilot CA root embedded in the daemon binary; will become the default in a future release once production root ships)")
 	showVersion := flag.Bool("version", false, "print version and exit")
+	pprofAddr := flag.String("pprof", "", "serve Go runtime profiles (heap, goroutine, CPU, trace) over HTTP at this loopback address, e.g. 127.0.0.1:6060; empty (the default) disables. Loopback only: any local user can connect, so leave it off on shared hosts")
 	logLevel := flag.String("log-level", "info", "log level (debug, info, warn, error)")
 	logFormat := flag.String("log-format", "text", "log format (text, json)")
 	logMaxSize := flag.Int("log-max-size", 50, "rotate the daemon log once it exceeds this many MB (copy-truncate into gzipped <log>.pilot.N.gz backups). By default only a log Pilot set up is rotated: one inside ~/.pilot (install.sh's launchd daemon.log, pilotctl's pilot-<pid>.log) or the Homebrew service's <brew prefix>/var/log/pilot-daemon.log; a log elsewhere only when this is set explicitly, here or in config.json. 0 disables")
@@ -178,6 +179,13 @@ func main() {
 	}
 
 	logging.Setup(*logLevel, *logFormat)
+	if *pprofAddr != "" {
+		bound, err := startPprof(*pprofAddr)
+		if err != nil {
+			log.Fatalf("pprof: %v", err)
+		}
+		slog.Info("pprof listening", "addr", bound.String(), "path", "/debug/pprof/")
+	}
 	// launchd never rotates StandardOutPath/StandardErrorPath (daemon.log
 	// reached 22 MB on one laptop), so cap it from the inside. No-op when
 	// stderr isn't a regular file (journald, a terminal, a pipe), and by

cmd/daemon/pprof.go

@@ -0,0 +1,75 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+	"errors"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/pprof"
+	"strings"
+	"time"
+)
+
+// errPprofNotLoopback rejects a -pprof address that is not on a loopback
+// interface. Profiles show where the daemon spends memory and CPU, and its
+// goroutine stacks; they are for the operator of this host, so the listener
+// never binds a routable address.
+var errPprofNotLoopback = errors.New("-pprof must be a loopback address (127.0.0.1:<port>, [::1]:<port> or localhost:<port>)")
+
+// pprofListenAddr checks a -pprof value and returns the address to bind:
+// "localhost" becomes 127.0.0.1 so a resolver cannot send it elsewhere.
+func pprofListenAddr(addr string) (string, error) {
+	host, port, err := net.SplitHostPort(addr)
+	if err != nil {
+		return "", fmt.Errorf("-pprof %q: %w", addr, err)
+	}
+	if strings.EqualFold(host, "localhost") {
+		host = "127.0.0.1"
+	}
+	ip := net.ParseIP(host)
+	if ip == nil || !ip.IsLoopback() {
+		return "", errPprofNotLoopback
+	}
+	return net.JoinHostPort(ip.String(), port), nil
+}
+
+// pprofMux serves Go's runtime profiles under /debug/pprof/: heap, allocs,
+// goroutine, threadcreate, block, mutex (through the index), plus a CPU
+// profile, an execution trace and symbol lookup. /debug/pprof/cmdline is
+// left out: the command line can carry -admin-token.
+func pprofMux() *http.ServeMux {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/debug/pprof/", pprof.Index)
+	mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
+	mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
+	mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
+	return mux
+}
+
+// startPprof serves runtime profiles on addr (see pprofListenAddr) for the
+// life of the process and returns the bound address. Off unless -pprof is
+// set. Any local user can connect to a loopback TCP port, unlike the IPC
+// socket, so leave it off where other users share the host.
+func startPprof(addr string) (net.Addr, error) {
+	bind, err := pprofListenAddr(addr)
+	if err != nil {
+		return nil, err
+	}
+	ln, err := net.Listen("tcp", bind)
+	if err != nil {
+		return nil, fmt.Errorf("-pprof listen %s: %w", bind, err)
+	}
+	srv := &http.Server{
+		Handler:           pprofMux(),
+		ReadHeaderTimeout: 10 * time.Second,
+	}
+	go func() {
+		if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			slog.Warn("pprof listener stopped", "addr", ln.Addr().String(), "error", err)
+		}
+	}()
+	return ln.Addr(), nil
+}

cmd/daemon/zz_pprof_test.go

@@ -0,0 +1,74 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestPprofListenAddrLoopbackOnly(t *testing.T) {
+	ok := map[string]string{
+		"127.0.0.1:6060": "127.0.0.1:6060",
+		"127.0.0.1:0":    "127.0.0.1:0",
+		"[::1]:6060":     "[::1]:6060",
+		"localhost:6060": "127.0.0.1:6060",
+		"LOCALHOST:7":    "127.0.0.1:7",
+		"127.1.2.3:6060": "127.1.2.3:6060",
+	}
+	for in, want := range ok {
+		got, err := pprofListenAddr(in)
+		if err != nil || got != want {
+			t.Errorf("pprofListenAddr(%q) = %q, %v; want %q", in, got, err, want)
+		}
+	}
+	for _, in := range []string{":6060", "0.0.0.0:6060", "[::]:6060", "10.0.0.1:6060", "34.71.57.205:6060", "example.com:6060"} {
+		if _, err := pprofListenAddr(in); !errors.Is(err, errPprofNotLoopback) {
+			t.Errorf("pprofListenAddr(%q) err = %v; want errPprofNotLoopback", in, err)
+		}
+	}
+	for _, in := range []string{"", "6060", "127.0.0.1"} {
+		if _, err := pprofListenAddr(in); err == nil {
+			t.Errorf("pprofListenAddr(%q) accepted a value with no port", in)
+		}
+	}
+}
+
+func TestPprofServesProfilesButNotCmdline(t *testing.T) {
+	addr, err := startPprof("127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	client := &http.Client{Timeout: 10 * time.Second}
+	get := func(path string) (int, string) {
+		t.Helper()
+		resp, err := client.Get("http://" + addr.String() + path)
+		if err != nil {
+			t.Fatalf("GET %s: %v", path, err)
+		}
+		defer resp.Body.Close()
+		body, _ := io.ReadAll(resp.Body)
+		return resp.StatusCode, string(body)
+	}
+	if code, body := get("/debug/pprof/heap?debug=1"); code != http.StatusOK || !strings.Contains(body, "heap profile") {
+		t.Fatalf("heap profile: status %d, body starts %.80q", code, body)
+	}
+	if code, body := get("/debug/pprof/goroutine?debug=1"); code != http.StatusOK || !strings.Contains(body, "goroutine profile") {
+		t.Fatalf("goroutine profile: status %d, body starts %.80q", code, body)
+	}
+	// The index would otherwise serve cmdline through its generic handler;
+	// it must not, because the command line can carry -admin-token.
+	if code, _ := get("/debug/pprof/cmdline"); code == http.StatusOK {
+		t.Fatalf("/debug/pprof/cmdline served (status %d); it must not be", code)
+	}
+}
+
+func TestStartPprofRejectsRoutableAddress(t *testing.T) {
+	if _, err := startPprof("0.0.0.0:0"); !errors.Is(err, errPprofNotLoopback) {
+		t.Fatalf("startPprof(0.0.0.0:0) err = %v; want errPprofNotLoopback", err)
+	}
+}