From cad18f4e7a4ca129b91846f01b8d08623edcaab8 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 14:19:07 +0300 Subject: [PATCH 01/10] daemon: opt-in -pprof listener for heap and goroutine profiles The service-agent fleet's daemons grow from ~26 MB to ~330 MB RSS over two days, and nothing in the binary could show where: v1.15.0 exposes no runtime metrics and no profiles, so the only evidence was RSS from /proc. -pprof 127.0.0.1: serves Go's runtime profiles (heap, allocs, goroutine, threadcreate, block, mutex, CPU profile, execution trace, symbol) on a private mux. It is off by default, and refuses any address that is not loopback ("localhost" binds 127.0.0.1). /debug/pprof/cmdline is not served, since the command line can carry -admin-token. Any local user can reach a loopback TCP port, unlike the IPC socket, so the flag's help says to leave it off on shared hosts. Co-Authored-By: Claude Opus 5.5 --- cmd/daemon/main.go | 8 ++++ cmd/daemon/pprof.go | 75 +++++++++++++++++++++++++++++++++++++ cmd/daemon/zz_pprof_test.go | 74 ++++++++++++++++++++++++++++++++++++ 3 files changed, 157 insertions(+) create mode 100644 cmd/daemon/pprof.go create mode 100644 cmd/daemon/zz_pprof_test.go diff --git a/cmd/daemon/main.go b/cmd/daemon/main.go index 2e50df24..766c219c 100644 --- a/cmd/daemon/main.go +++ b/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 .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-.log) or the Homebrew service's /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 diff --git a/cmd/daemon/pprof.go b/cmd/daemon/pprof.go new file mode 100644 index 00000000..512b8817 --- /dev/null +++ b/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:, [::1]: or localhost:)") + +// 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 +} diff --git a/cmd/daemon/zz_pprof_test.go b/cmd/daemon/zz_pprof_test.go new file mode 100644 index 00000000..9b54b1ca --- /dev/null +++ b/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) + } +} From a18c33cd4252c57a63254183241628d10aec7342 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Tue, 6 Oct 2026 02:10:30 +0300 Subject: [PATCH 02/10] daemon: stop maintaining paths to peers nobody is using Every peer the daemon had exchanged keys with got a NAT keepalive every 25 s, path-watchdog probes and resets when quiet, and, when relayed, a direct-upgrade attempt every 15 s (beacon punch, five probes, a registry resolve each minute), forever. The keepalives also counted as contact, so the stale-peer reaper never fired. A service agent with ~4,000 peers and no open connection sent ~120 KB/s of this. Track application traffic per peer. A peer with none for 2 minutes and no open connection gets no keepalives, path probes or upgrade attempts; the reaper then drops it five minutes after its last frame. The reaper now counts relayed inbound frames as contact, so a peer still sending to us is kept. Co-Authored-By: Claude Opus 5.5 (cherry picked from commit 0b1135d0dfe964dd99217ceb0b602e1f80d81975) --- CHANGELOG.md | 15 +++ pkg/daemon/daemon.go | 15 ++- pkg/daemon/pathwatch.go | 7 ++ pkg/daemon/peeractivity.go | 114 +++++++++++++++++ pkg/daemon/tunnel.go | 24 ++++ pkg/daemon/zz_idle_peer_upkeep_test.go | 167 +++++++++++++++++++++++++ 6 files changed, 341 insertions(+), 1 deletion(-) create mode 100644 pkg/daemon/peeractivity.go create mode 100644 pkg/daemon/zz_idle_peer_upkeep_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 8468314a..e60b687d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -341,6 +341,21 @@ 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. - **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 diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index d0829569..b210f658 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -701,6 +701,7 @@ func New(cfg Config) *Daemon { // peer trusted only through the registry now gets the redacted event, // which is the conservative side. d.tunnels.SetPeerTrustFn(d.handshakeTrusts) + d.tunnels.SetOpenConnPeers(d.ports.ActiveNodeIDs) d.ipc = NewIPCServer(cfg.SocketPath, d) // HandshakeService is wired post-construction by the composition // root via RegisterHandshakeService (T3.3 — handshake plugin moved @@ -6470,11 +6471,17 @@ func (d *Daemon) reapStalePeers() { lastDirect := d.tunnels.LastDirectRecv(p.NodeID) lastOutbound, ok := d.tunnels.LastOutboundSend(p.NodeID) - // Find the latest known contact. + // Find the latest known contact. Any authenticated inbound frame + // counts, relayed ones too (LastDirectRecv skips those): a peer + // still talking to us is not stale, and dropping it would only + // make its next frame trigger a rekey. latest := lastDirect if ok && lastOutbound.After(latest) { latest = lastOutbound } + if lastIn, ok := d.tunnels.LastInboundDecrypt(p.NodeID); ok && lastIn.After(latest) { + latest = lastIn + } if latest.IsZero() { // Never contacted — fresh peer entry, don't reap yet. @@ -6651,7 +6658,13 @@ func (d *Daemon) relayProbeLoop() { case <-d.stopCh: return case <-ticker.C: + idle := d.tunnels.idleFilter(time.Now()) for _, nodeID := range d.tunnels.RelayPeerIDs() { + if idle(nodeID) { + // An upgrade costs a registry lookup, a beacon punch + // and five probes; nobody is using this path. + continue + } go d.tryDirectUpgrade(nodeID) } } diff --git a/pkg/daemon/pathwatch.go b/pkg/daemon/pathwatch.go index 8aabe524..2d0920e1 100644 --- a/pkg/daemon/pathwatch.go +++ b/pkg/daemon/pathwatch.go @@ -124,8 +124,15 @@ func (d *Daemon) pathWatchTick(states map[uint32]*pathPeerState, now time.Time) defer recoverLayer("L4", "pathWatchTick", d.bus, nil) ready := d.tunnels.ReadyPeerIDs() + idle := d.tunnels.idleFilter(now) live := make(map[uint32]bool, len(ready)) for _, nodeID := range ready { + if idle(nodeID) { + // Nobody is using this peer: silence from it is expected, + // and probing or resetting it would only cost both sides + // packets and a key exchange. Its state is dropped below. + continue + } live[nodeID] = true st := states[nodeID] if st == nil { diff --git a/pkg/daemon/peeractivity.go b/pkg/daemon/peeractivity.go new file mode 100644 index 00000000..718c6ea3 --- /dev/null +++ b/pkg/daemon/peeractivity.go @@ -0,0 +1,114 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "sync" + "sync/atomic" + "time" +) + +// PeerIdleAfter is how long a peer may go without application traffic +// before the daemon stops maintaining its path in the background. +// +// The daemon keeps a session warm with three periodic jobs: NAT keepalives +// every TunnelKeepaliveInterval, path-watchdog probes when the peer goes +// quiet, and a direct-path upgrade attempt (registry lookup, beacon punch, +// five probes) every RelayProbeInterval for each relayed peer. All three +// ran for every peer the daemon had ever exchanged keys with, forever: a +// public service agent that had answered a few thousand clients spent +// about 1,600 packets a second on them while serving a few requests a +// minute. +// +// Sending to an idle peer works as before: the first packet re-establishes +// the path through the usual fallbacks (address learning, blackhole +// detection, relay) if a NAT mapping expired in the meantime. Five minutes +// after the last frame either way, the stale-peer reaper drops the peer +// (reapStalePeers), which the keepalives used to prevent; the next contact +// then runs a fresh key exchange, as first contact does. A peer with an open +// connection is never idle. +var PeerIdleAfter = 2 * time.Minute + +// peerActivity records, per peer, when application traffic last went to or +// came from it. Path upkeep (keepalives, path probes, upgrade probes, key +// exchange) is not application traffic and does not count. +type peerActivity struct { + mu sync.RWMutex + m map[uint32]*atomic.Int64 // unix nanoseconds +} + +func newPeerActivity() *peerActivity { + return &peerActivity{m: make(map[uint32]*atomic.Int64)} +} + +func (a *peerActivity) note(nodeID uint32, now time.Time) { + a.mu.RLock() + v := a.m[nodeID] + a.mu.RUnlock() + if v == nil { + a.mu.Lock() + if v = a.m[nodeID]; v == nil { + v = new(atomic.Int64) + a.m[nodeID] = v + } + a.mu.Unlock() + } + v.Store(now.UnixNano()) +} + +// last reports the last activity, and false when none was ever recorded. +func (a *peerActivity) last(nodeID uint32) (time.Time, bool) { + a.mu.RLock() + v := a.m[nodeID] + a.mu.RUnlock() + if v == nil { + return time.Time{}, false + } + return time.Unix(0, v.Load()), true +} + +func (a *peerActivity) forget(nodeID uint32) { + a.mu.Lock() + delete(a.m, nodeID) + a.mu.Unlock() +} + +// noteAppActivity marks application traffic to or from a peer. +func (tm *TunnelManager) noteAppActivity(nodeID uint32) { + if tm.activity != nil { + tm.activity.note(nodeID, time.Now()) + } +} + +// SetOpenConnPeers installs the daemon's view of which peers have an open +// connection; such a peer is never idle however quiet the connection is. +func (tm *TunnelManager) SetOpenConnPeers(fn func() map[uint32]bool) { + tm.openConnPeers = fn +} + +// idleFilter returns a predicate reporting whether a peer is idle at now. +// It reads the open-connection set once, so callers sweeping many peers +// pay for it once per sweep. A peer with no recorded activity counts as +// active: every session records activity when it is first established, so +// only state that predates tracking falls in that case. +func (tm *TunnelManager) idleFilter(now time.Time) func(nodeID uint32) bool { + var open map[uint32]bool + if tm.openConnPeers != nil { + open = tm.openConnPeers() + } + return func(nodeID uint32) bool { + if PeerIdleAfter <= 0 || tm.activity == nil || open[nodeID] { + return false + } + last, ok := tm.activity.last(nodeID) + if !ok { + return false + } + return now.Sub(last) > PeerIdleAfter + } +} + +// PeerIdle reports whether one peer is idle now. +func (tm *TunnelManager) PeerIdle(nodeID uint32) bool { + return tm.idleFilter(time.Now())(nodeID) +} diff --git a/pkg/daemon/tunnel.go b/pkg/daemon/tunnel.go index 01375335..7a9c5b3e 100644 --- a/pkg/daemon/tunnel.go +++ b/pkg/daemon/tunnel.go @@ -85,6 +85,10 @@ type TunnelManager struct { sock transport.Transport // peers maps node_id → real UDP endpoint. Owned by L4 (routing). peers map[uint32]*net.UDPAddr + // activity and openConnPeers decide which peers are idle; see + // peeractivity.go. + activity *peerActivity + openConnPeers func() map[uint32]bool // envelope is the L5-owned per-peer crypto Store (named "envelope" // for historical reasons — pre-T5.x-followup the Store lived in // pkg/daemon/envelope). Accessed by: @@ -413,6 +417,7 @@ func NewTunnelManager() *TunnelManager { routing: routing.New(), kxRateLim: make(map[string]*srcKxBucket), relayKxLim: make(map[uint32]*srcKxBucket), + activity: newPeerActivity(), } tm.routing.SetLocalNodeIDFn(tm.loadNodeID) tm.kx = keyexchange.New(store) @@ -988,6 +993,7 @@ func (tm *TunnelManager) keepaliveSweep(now time.Time) int { addr *net.UDPAddr pc *peerCrypto } + idle := tm.idleFilter(now) tm.mu.RLock() stale := make([]peerInfo, 0, len(tm.peers)) for nodeID, addr := range tm.peers { @@ -995,6 +1001,9 @@ func (tm *TunnelManager) keepaliveSweep(now time.Time) int { if ok && now.Sub(last) < TunnelKeepaliveInterval { continue } + if idle(nodeID) { + continue + } pc := tm.envelope.Get(nodeID) if pc == nil || !pc.Ready { continue @@ -1585,6 +1594,12 @@ func (tm *TunnelManager) onKeyInstalled(ev keyexchange.PostInstallEvent) { // and the per-peer 1s reply cooldown holds the ping-pong back but // never lets the staleness flag clear. tm.recordInboundDecrypt(peerNodeID) + if !ev.HadCrypto { + // A new session counts as activity, so it starts out maintained + // and ages into idleness like any other. A rekey of an existing + // session does not: it is upkeep, not use. + tm.noteAppActivity(peerNodeID) + } if !ev.HadCrypto || ev.KeyChanged { tm.keyViaRelay.Store(peerNodeID, fromRelay) } @@ -1866,6 +1881,9 @@ func (tm *TunnelManager) handleEncrypted(data []byte, from *net.UDPAddr) { if isTunnelKeepalive(pkt) { return } + if !(pkt.Protocol == protocol.ProtoControl && pkt.DstPort == protocol.PortPing) { + tm.noteAppActivity(peerNodeID) + } select { case tm.recvCh <- &IncomingPacket{Packet: pkt, From: from}: @@ -2192,6 +2210,9 @@ func (tm *TunnelManager) SetRelayPeerPinned(nodeID uint32, relay bool) { // SendTo sends a packet to a specific UDP address (relay-aware). func (tm *TunnelManager) SendTo(addr *net.UDPAddr, nodeID uint32, pkt *protocol.Packet) error { + // Path upkeep (keepalives, probes, key exchange) bypasses SendTo, so + // everything that arrives here is traffic someone asked for. + tm.noteAppActivity(nodeID) data, err := pkt.Marshal() if err != nil { return fmt.Errorf("marshal: %w", err) @@ -2329,6 +2350,9 @@ func (tm *TunnelManager) RemovePeer(nodeID uint32) { // L5-owned per-peer state (peerPubKeys, pendingRekey, lastInboundDecrypt). tm.kx.RemovePeer(nodeID) tm.keyViaRelay.Delete(nodeID) + if tm.activity != nil { + tm.activity.forget(nodeID) + } } // KeyArrivedViaRelayOnly reports whether the peer's session key was diff --git a/pkg/daemon/zz_idle_peer_upkeep_test.go b/pkg/daemon/zz_idle_peer_upkeep_test.go new file mode 100644 index 00000000..1c33a6fd --- /dev/null +++ b/pkg/daemon/zz_idle_peer_upkeep_test.go @@ -0,0 +1,167 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "crypto/ecdh" + "crypto/rand" + "net" + "testing" + "time" + + "github.com/pilot-protocol/common/protocol" +) + +// readyPeer installs a peer with a ready session whose last outbound send +// was long ago, so keepaliveSweep would normally refresh it. +func readyPeer(t *testing.T, tm *TunnelManager, nodeID uint32) { + t.Helper() + sock, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0}) + if err != nil { + t.Fatalf("peer listen: %v", err) + } + t.Cleanup(func() { sock.Close() }) + priv, err := ecdh.X25519().GenerateKey(rand.Reader) + if err != nil { + t.Fatalf("keygen: %v", err) + } + pc, err := tm.deriveSecret(priv.PublicKey().Bytes()) + if err != nil { + t.Fatalf("deriveSecret: %v", err) + } + pc.Ready = true + tm.mu.Lock() + tm.peers[nodeID] = sock.LocalAddr().(*net.UDPAddr) + tm.envelope.Install(nodeID, pc) + tm.routing.RecordOutboundSend(nodeID, time.Now().Add(-time.Minute)) + tm.mu.Unlock() +} + +func newEncryptedTunnel(t *testing.T) *TunnelManager { + t.Helper() + tm := NewTunnelManager() + t.Cleanup(func() { tm.Close() }) + if err := tm.Listen("127.0.0.1:0"); err != nil { + t.Fatalf("Listen: %v", err) + } + if err := tm.EnableEncryption(); err != nil { + t.Fatalf("EnableEncryption: %v", err) + } + tm.SetNodeID(0xAA000002) + return tm +} + +// Keepalives go to peers in use and to peers with an open connection, and +// not to peers nobody has used for longer than PeerIdleAfter. +func TestKeepaliveSweepSkipsIdlePeers(t *testing.T) { + t.Parallel() + tm := newEncryptedTunnel(t) + + const active, idle, idleOpenConn, untracked = 0xB1, 0xB2, 0xB3, 0xB4 + for _, id := range []uint32{active, idle, idleOpenConn, untracked} { + readyPeer(t, tm, id) + } + now := time.Now() + tm.activity.note(active, now.Add(-10*time.Second)) + tm.activity.note(idle, now.Add(-PeerIdleAfter-time.Minute)) + tm.activity.note(idleOpenConn, now.Add(-PeerIdleAfter-time.Minute)) + tm.SetOpenConnPeers(func() map[uint32]bool { return map[uint32]bool{idleOpenConn: true} }) + + if sent := tm.keepaliveSweep(now); sent != 3 { + t.Fatalf("keepalives sent = %d, want 3 (active, open connection, untracked; not the idle peer)", sent) + } + if last, _ := tm.routing.LastOutboundSend(idle); now.Sub(last) < 30*time.Second { + t.Fatal("the idle peer was sent a keepalive") + } +} + +// Application traffic in either direction makes a peer active; path +// upkeep does not. +func TestOnlyApplicationTrafficCountsAsActivity(t *testing.T) { + t.Parallel() + tm := newEncryptedTunnel(t) + const peer = 0xC1 + readyPeer(t, tm, peer) + old := time.Now().Add(-PeerIdleAfter - time.Minute) + tm.activity.note(peer, old) + + // Upkeep: a keepalive and a path probe must not refresh activity. + tm.keepaliveSweep(time.Now()) + if err := tm.SendPathProbe(peer); err != nil { + t.Fatalf("SendPathProbe: %v", err) + } + if last, _ := tm.activity.last(peer); !last.Equal(time.Unix(0, old.UnixNano())) { + t.Fatal("keepalive or path probe counted as application activity") + } + if !tm.PeerIdle(peer) { + t.Fatal("peer should be idle") + } + + // Application traffic: any packet sent through Send. + _ = tm.Send(peer, &protocol.Packet{ + Version: protocol.Version, + Protocol: protocol.ProtoStream, + Src: protocol.Addr{Node: 0xAA000002}, + Dst: protocol.Addr{Node: peer}, + DstPort: protocol.PortDataExchange, + }) + if tm.PeerIdle(peer) { + t.Fatal("sending application data did not make the peer active") + } +} + +// The path watchdog leaves idle peers alone: no probes, no resets. +func TestPathWatchSkipsIdlePeers(t *testing.T) { + resets := swapPathResetForTest(t) + const peer = 61 + d := newPathWatchTestDaemon(t, peer) + // Long silent, and long unused. + d.tunnels.kx.RecordInboundDecrypt(peer) + states := map[uint32]*pathPeerState{} + silentNow := time.Now().Add(2 * pathSilenceThreshold) + d.tunnels.activity.note(peer, silentNow.Add(-PeerIdleAfter-time.Minute)) + + for i := 0; i < pathProbeMax+3; i++ { + d.pathWatchTick(states, silentNow.Add(time.Duration(i)*pathWatchTickInterval)) + } + if len(*resets) != 0 { + t.Fatalf("idle peer was reset %d times", len(*resets)) + } + if st := states[peer]; st != nil && st.probesSent != 0 { + t.Fatalf("idle peer was probed %d times", st.probesSent) + } + + // The same peer in use is still watched. + d.tunnels.activity.note(peer, silentNow) + for i := 0; i < pathProbeMax+2; i++ { + d.pathWatchTick(states, silentNow.Add(time.Duration(i)*pathWatchTickInterval)) + } + if len(*resets) == 0 { + t.Fatal("an active silent peer was not reset; the watchdog no longer works") + } +} + +// Once an idle peer gets no keepalives, the stale-peer reaper can drop it +// (keepalives used to count as contact and kept every peer forever). A +// peer still sending to us is kept, even when its frames come through the +// relay, which LastDirectRecv does not count. +func TestReaperDropsIdlePeersButKeepsTalkingOnes(t *testing.T) { + const silent, talking = 71, 72 + d := newPathWatchTestDaemon(t, silent) + d.ports = NewPortManager() + d.tunnels.AddPeer(talking, &net.UDPAddr{IP: net.IPv4(203, 0, 113, 8), Port: 4000}) + long := time.Now().Add(-10 * time.Minute) + for _, id := range []uint32{silent, talking} { + d.tunnels.routing.RecordOutboundSend(id, long) + d.tunnels.routing.RecordDirectRecv(id, long) + } + d.tunnels.kx.RecordInboundDecrypt(talking) // a relayed frame just now + + d.reapStalePeers() + if d.tunnels.HasPeer(silent) { + t.Fatal("a peer silent for 10 minutes was not reaped") + } + if !d.tunnels.HasPeer(talking) { + t.Fatal("a peer that just sent us a frame was reaped") + } +} From 13d9108eaa61c80a91684d98ee6554a3a7b75304 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Tue, 6 Oct 2026 03:07:02 +0300 Subject: [PATCH 03/10] tests: idle peers behind a NAT, before and after the reap Two Docker scenarios on a cone NAT with a 10 s UDP conntrack timeout: idle 160 s (keepalives stopped, mapping expired) and idle 8 minutes (both sides reap each other), then an echo in each direction. Co-Authored-By: Claude Opus 5.5 (cherry picked from commit a6da8d1f4c9bdea164cc9bf71fd473f940905dae) --- .../local/test_nat_idle_peer_reaped.sh | 87 +++++++++++++++++++ .../local/test_nat_idle_peer_resume.sh | 76 ++++++++++++++++ 2 files changed, 163 insertions(+) create mode 100755 tests/integration/local/test_nat_idle_peer_reaped.sh create mode 100755 tests/integration/local/test_nat_idle_peer_resume.sh diff --git a/tests/integration/local/test_nat_idle_peer_reaped.sh b/tests/integration/local/test_nat_idle_peer_reaped.sh new file mode 100755 index 00000000..19eac250 --- /dev/null +++ b/tests/integration/local/test_nat_idle_peer_reaped.sh @@ -0,0 +1,87 @@ +#!/bin/bash +# A peer left unused past PeerIdleAfter (2 min) gets no keepalives, so five +# minutes later the stale-peer reaper forgets it (reapStalePeers: no open +# connection, no contact for 5 min). Before idle peers stopped getting +# keepalives the reaper never fired, as keepalives counted as contact. +# After the reap both sides have to build the path again, from scratch, +# and traffic must get through in both directions. +# +# Cone NAT with a 10 s UDP conntrack timeout, idle for 8 minutes. +source "$(dirname "$0")/nat_test_common.sh" +export NAT_MODE="conntrack_short" +cd "$(dirname "$0")" || exit 1 +trap cleanup_nat EXIT + +boot_nat_stack + +log_test "agents register" +REG=$(wait_registered 2 60 || echo "0") +if [ "${REG:-0}" -ge 2 ]; then + log_pass "$REG nodes" +else + log_fail "only $REG registered" + exit 1 +fi + +log_test "a->b initial echo, then mutual trust" +OUT=$(echo_rt agent-a agent-b "reap-probe-1") +if echo "$OUT" | grep -q "reap-probe-1"; then + log_pass "initial echo ok" +else + log_fail "initial echo failed" +fi +establish_trust agent-b agent-a >/dev/null + +NID_A=$(agent_node_id agent-a) +AGENT_A_ADDR=$(pilot_addr "$NID_A") + +log_test "b->a echo while in use" +OUT=$(echo_rt agent-b "$AGENT_A_ADDR" "reap-probe-2") +if echo "$OUT" | grep -q "reap-probe-2"; then + log_pass "echo ok" +else + log_fail "echo failed before idling" +fi + +log_test "idle 480 s (2 min to idle, 5 more to the reap)" +sleep 480 +log_pass "idle period elapsed" + +NID_B=$(agent_node_id agent-b) +has_peer() { $DC exec -T "$1" pilotctl --json peers 2>/dev/null | jq -e --argjson n "$2" '[.data.peers[]?.node_id] | index($n) != null' >/dev/null; } +log_test "the idle peers were forgotten" +GONE="" +has_peer agent-a "$NID_B" || GONE="$GONE a-forgot-b" +has_peer agent-b "$NID_A" || GONE="$GONE b-forgot-a" +echo " reaped:${GONE:- none}" +log_pass "peer tables after idle:${GONE:- both still listed}" + +log_test "b->a echo after idle: inbound to the NATed peer" +T0=$(date +%s) +OUT=$(echo_rt agent-b "$AGENT_A_ADDR" "reap-probe-3" 30s) +T1=$(date +%s) +if echo "$OUT" | grep -q "reap-probe-3"; then + log_pass "echo ok after idle in $((T1 - T0)) s" +else + log_fail "echo failed after idle ($((T1 - T0)) s): $(echo "$OUT" | head -c 200)" +fi + +log_test "a->b echo after idle: outbound from the NATed peer" +T0=$(date +%s) +OUT=$(echo_rt agent-a agent-b "reap-probe-4" 30s) +T1=$(date +%s) +if echo "$OUT" | grep -q "reap-probe-4"; then + log_pass "echo ok after idle in $((T1 - T0)) s" +else + log_fail "echo failed after idle ($((T1 - T0)) s): $(echo "$OUT" | head -c 200)" +fi + +log_test "no panics" +BAD=$($DC logs rendezvous agent-a agent-b nat-gw 2>&1 | grep -iE "panic|fatal|race detected" | head -3) +if [ -z "$BAD" ]; then + log_pass "clean logs" +else + log_fail "found: $BAD" +fi + +print_summary_and_exit diff --git a/tests/integration/local/test_nat_idle_peer_resume.sh b/tests/integration/local/test_nat_idle_peer_resume.sh new file mode 100755 index 00000000..03621764 --- /dev/null +++ b/tests/integration/local/test_nat_idle_peer_resume.sh @@ -0,0 +1,76 @@ +#!/bin/bash +# A peer left unused past PeerIdleAfter (2 min) gets no keepalives or path +# probes, so a NAT mapping toward it can expire. Traffic must still get +# through afterwards, in both directions, by the usual fallbacks. +# +# Cone NAT with a 10 s UDP conntrack timeout (as test_nat_conntrack_timeout), +# but idle for 160 s: past PeerIdleAfter, and far past the mapping's life. +source "$(dirname "$0")/nat_test_common.sh" +export NAT_MODE="conntrack_short" +cd "$(dirname "$0")" || exit 1 +trap cleanup_nat EXIT + +boot_nat_stack + +log_test "agents register" +REG=$(wait_registered 2 60 || echo "0") +if [ "${REG:-0}" -ge 2 ]; then + log_pass "$REG nodes" +else + log_fail "only $REG registered" + exit 1 +fi + +log_test "a->b initial echo, then mutual trust" +OUT=$(echo_rt agent-a agent-b "idle-probe-1") +if echo "$OUT" | grep -q "idle-probe-1"; then + log_pass "initial echo ok" +else + log_fail "initial echo failed" +fi +establish_trust agent-b agent-a >/dev/null + +NID_A=$(agent_node_id agent-a) +AGENT_A_ADDR=$(pilot_addr "$NID_A") + +log_test "b->a echo while in use" +OUT=$(echo_rt agent-b "$AGENT_A_ADDR" "idle-probe-2") +if echo "$OUT" | grep -q "idle-probe-2"; then + log_pass "echo ok" +else + log_fail "echo failed before idling" +fi + +log_test "idle 160 s (past the 2 min idle threshold and the 10 s conntrack timeout)" +sleep 160 +log_pass "idle period elapsed" + +log_test "b->a echo after idle: inbound to the NATed peer" +T0=$(date +%s) +OUT=$(echo_rt agent-b "$AGENT_A_ADDR" "idle-probe-3" 30s) +T1=$(date +%s) +if echo "$OUT" | grep -q "idle-probe-3"; then + log_pass "echo ok after idle in $((T1 - T0)) s" +else + log_fail "echo failed after idle ($((T1 - T0)) s): $(echo "$OUT" | head -c 200)" +fi + +log_test "a->b echo after idle: outbound from the NATed peer" +T0=$(date +%s) +OUT=$(echo_rt agent-a agent-b "idle-probe-4" 30s) +T1=$(date +%s) +if echo "$OUT" | grep -q "idle-probe-4"; then + log_pass "echo ok after idle in $((T1 - T0)) s" +else + log_fail "echo failed after idle ($((T1 - T0)) s): $(echo "$OUT" | head -c 200)" +fi + +log_test "no panics" +BAD=$($DC logs rendezvous agent-a agent-b nat-gw 2>&1 | grep -iE "panic|fatal|race detected" | head -3) +if [ -z "$BAD" ]; then + log_pass "clean logs" +else + log_fail "found: $BAD" +fi + +print_summary_and_exit From 7385aac6e24be330dbc53d21c4f929a6e18cc069 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 11:51:56 +0300 Subject: [PATCH 04/10] daemon: answering a peer's path probe is not activity Review fix. SendTo counted every packet it sent as application traffic, including the pong to a peer's path probe. Any idle peer with a path watchdog (v1.13.10 and later) probes after 55 s of quiet, so the pong made it active again for PeerIdleAfter and the upkeep this change stops came back for those peers. Outbound pings and pongs are now treated as they already were inbound. Co-Authored-By: Claude Opus 5.5 (cherry picked from commit ec10ffd9785fbde867348074801bfe63f76d8aed) --- pkg/daemon/tunnel.go | 11 +++++++++-- pkg/daemon/zz_idle_peer_upkeep_test.go | 16 ++++++++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/pkg/daemon/tunnel.go b/pkg/daemon/tunnel.go index 7a9c5b3e..c4fbb787 100644 --- a/pkg/daemon/tunnel.go +++ b/pkg/daemon/tunnel.go @@ -2211,8 +2211,15 @@ func (tm *TunnelManager) SetRelayPeerPinned(nodeID uint32, relay bool) { // SendTo sends a packet to a specific UDP address (relay-aware). func (tm *TunnelManager) SendTo(addr *net.UDPAddr, nodeID uint32, pkt *protocol.Packet) error { // Path upkeep (keepalives, probes, key exchange) bypasses SendTo, so - // everything that arrives here is traffic someone asked for. - tm.noteAppActivity(nodeID) + // everything that arrives here is traffic someone asked for — except + // the pong to a peer's path probe, which is upkeep on that peer's + // behalf. Counting it kept every idle peer with a path watchdog + // (v1.13.10 and later) maintained: it probes after 55 s of quiet, the + // pong made it active again for PeerIdleAfter, and so on. Same rule as + // for packets received (handleEncrypted). + if !(pkt.Protocol == protocol.ProtoControl && pkt.DstPort == protocol.PortPing) { + tm.noteAppActivity(nodeID) + } data, err := pkt.Marshal() if err != nil { return fmt.Errorf("marshal: %w", err) diff --git a/pkg/daemon/zz_idle_peer_upkeep_test.go b/pkg/daemon/zz_idle_peer_upkeep_test.go index 1c33a6fd..e77d9c79 100644 --- a/pkg/daemon/zz_idle_peer_upkeep_test.go +++ b/pkg/daemon/zz_idle_peer_upkeep_test.go @@ -97,6 +97,22 @@ func TestOnlyApplicationTrafficCountsAsActivity(t *testing.T) { t.Fatal("peer should be idle") } + // Our pong to the peer's path probe is upkeep too: counting it made + // every peer with a path watchdog active again 55 s after it went quiet. + _ = tm.Send(peer, &protocol.Packet{ + Version: protocol.Version, + Flags: protocol.FlagACK, + Protocol: protocol.ProtoControl, + Src: protocol.Addr{Node: 0xAA000002}, + Dst: protocol.Addr{Node: peer}, + SrcPort: protocol.PortPing, + DstPort: protocol.PortPing, + Payload: []byte("probe"), + }) + if !tm.PeerIdle(peer) { + t.Fatal("answering the peer's path probe counted as application activity") + } + // Application traffic: any packet sent through Send. _ = tm.Send(peer, &protocol.Packet{ Version: protocol.Version, From d1af6cf36afba3a56ad2f760afc197cf9f5cf2f2 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 14:31:58 +0300 Subject: [PATCH 05/10] daemon: a reaped peer leaves nothing behind; stale pending queues age out The stale-peer reaper (now able to fire, see the idle-peer change before this) removes a peer with RemovePeer, which keeps the frames queued for the peer's key exchange: a path reset calls it too, and re-keys at once. For a peer that is being forgotten those frames are never sent. They were dropped only when a key exchange completed, so for a peer that never answered they stayed for the daemon's lifetime; once maxPendingPeers (256) of them had piled up, every first send to a new peer without a session failed with "too many pending key exchanges". - reapStalePeers drops a peer through forgetPeer, which also discards its pending queue. - The idle sweep drops any pending queue older than pendingMaxAge (2 min; a key exchange is given up after 20 s), whoever still holds it. - routing.Manager and keyexchange.Manager report how many per-peer entries they hold (PeerStateEntries), and TunnelManager.peerStateEntries sums every per-peer table, for tests and diagnostics. TestPeerChurnStateStaysBounded runs the service-agent workload in-process: 12 rounds of 400 new clients that each make one request and go quiet, with the daemon's keepalive sweep and reaper run between rounds. The tables hold one round (400 peers, 2,800 entries) throughout and the live heap stays at 1.2-1.5 MB. Its control subtest turns idleness tracking off, the keepalive behaviour before the idle-peer change, and shows the leak: 400 -> 4,800 peers and 1.2 -> 8.0 MB, one round's worth per round. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 + pkg/daemon/daemon.go | 16 +- pkg/daemon/keyexchange/keyexchange.go | 15 ++ pkg/daemon/routing/relay.go | 11 + pkg/daemon/tunnel.go | 81 ++++++++ pkg/daemon/zz_peer_churn_bounded_test.go | 246 +++++++++++++++++++++++ 6 files changed, 375 insertions(+), 1 deletion(-) create mode 100644 pkg/daemon/zz_peer_churn_bounded_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index e60b687d..9a36663c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -356,6 +356,13 @@ Detailed per-release notes are on the 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. - **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 diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index b210f658..dff3f99c 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -6492,11 +6492,21 @@ func (d *Daemon) reapStalePeers() { slog.Debug("reaping stale peer", "node_id", p.NodeID, "last_contact", latest.Round(time.Second), "has_encryption", p.Encrypted) - d.tunnels.RemovePeer(p.NodeID) + d.forgetPeer(p.NodeID) } } } +// forgetPeer drops everything the daemon keeps for a peer it no longer +// talks to: the tunnel entry and its routing and key state (RemovePeer), +// and the frames still queued for a key exchange the peer never answered, +// which RemovePeer keeps for a path reset's re-key. The next contact +// starts from scratch, as first contact does. +func (d *Daemon) forgetPeer(nodeID uint32) { + d.tunnels.RemovePeer(nodeID) + d.tunnels.dropPending(nodeID) +} + // hostnameReannounceLoop periodically re-sets the daemon's hostname // with the registry, so hostname resolution self-heals after a registry // restart/roll that wipes the in-memory hostname store. @@ -6563,6 +6573,10 @@ func (d *Daemon) idleSweepLoop() { // unique-peer-ever-throttled. d.tunnels.reapPendDropLog() + // Drop frames that have waited too long for a key exchange + // that is not coming, freeing their pending-peer slot. + d.tunnels.reapStalePending(time.Now()) + // Evict expired entries from the three peer-resolution caches. // epCache: stale-but-usable after EndpointCacheTTL; evict after 1 hour. // resolveCache and hostnameCache: strictly TTL-gated on read; evict after 2× TTL. diff --git a/pkg/daemon/keyexchange/keyexchange.go b/pkg/daemon/keyexchange/keyexchange.go index 345fa0f8..62192d3f 100644 --- a/pkg/daemon/keyexchange/keyexchange.go +++ b/pkg/daemon/keyexchange/keyexchange.go @@ -738,6 +738,21 @@ func (m *Manager) SetReplyWhitelistMatchAll(on bool) { m.replyWhitelistAll.Store(on) } +// PeerStateEntries counts the entries in every per-peer map this manager +// keeps. RemovePeer clears most of them; rekeyGaveUp expires on its own +// cooldown. A count that only grows is per-peer state that outlives its +// peer. +func (m *Manager) PeerStateEntries() int { + m.pubKeysMu.RLock() + n := len(m.peerPubKeys) + len(m.negPubKeys) + m.pubKeysMu.RUnlock() + m.rkPendingMu.Lock() + n += len(m.pendingRekey) + len(m.lastInboundDecrypt) + len(m.rekeyGaveUp) + + len(m.lastKeyExchangeReply) + len(m.keyRequestDue) + m.rkPendingMu.Unlock() + return n +} + // RemovePeer wipes per-peer L5 state (called from TunnelManager.RemovePeer). func (m *Manager) RemovePeer(nodeID uint32) { m.pubKeysMu.Lock() diff --git a/pkg/daemon/routing/relay.go b/pkg/daemon/routing/relay.go index 25cd15f7..425a03af 100644 --- a/pkg/daemon/routing/relay.go +++ b/pkg/daemon/routing/relay.go @@ -168,6 +168,17 @@ func (m *Manager) ClearRelayOnDirect(peerNodeID uint32, from *net.UDPAddr) bool return false } +// PeerStateEntries counts the entries in every per-peer map this manager +// keeps. It drops back as peers are removed (RemovePeer): a count that +// only grows is per-peer state that outlives its peer. +func (m *Manager) PeerStateEntries() int { + m.mu.RLock() + defer m.mu.RUnlock() + return len(m.relayPeers) + len(m.relayPinned) + len(m.lastOutboundSend) + + len(m.firstOutboundSend) + len(m.sendErrCount) + len(m.lastDirectRecv) + + len(m.blackholeMissCount) + len(m.directClearCount) +} + // RemovePeer wipes per-peer L4 state. func (m *Manager) RemovePeer(nodeID uint32) { m.mu.Lock() diff --git a/pkg/daemon/tunnel.go b/pkg/daemon/tunnel.go index c4fbb787..3fae6b22 100644 --- a/pkg/daemon/tunnel.go +++ b/pkg/daemon/tunnel.go @@ -136,6 +136,9 @@ type TunnelManager struct { // Pending sends waiting for key exchange to complete pendMu sync.Mutex pending map[uint32][][]byte // node_id → queued frames + // pendingSince is when each queue in pending got its first frame; + // reapStalePending drops queues older than pendingMaxAge. + pendingSince map[uint32]time.Time // lastPendDropLog records when we last logged a "tunnel pending queue // full; dropped newest" warning for a peer. The drop counter // (PendingDrops) still increments on every drop so metrics are exact; @@ -410,6 +413,7 @@ func NewTunnelManager() *TunnelManager { peers: make(map[uint32]*net.UDPAddr), envelope: store, pending: make(map[uint32][][]byte), + pendingSince: make(map[uint32]time.Time), lastPendDropLog: make(map[uint32]time.Time), lastRekeyReq: make(map[uint32]time.Time), recvCh: make(chan *IncomingPacket, RecvChSize), @@ -521,6 +525,79 @@ const rekeyRequestInterval = 3 * time.Second // metric accuracy is unaffected. const pendingDropLogInterval = 5 * time.Second +// pendingMaxAge bounds how long frames wait in the pending queue for a key +// exchange. A key exchange is retransmitted for 20 s (MaxRekeyAttempts × +// RekeyRetransmitInterval) and then given up, so a queue still waiting +// after this long belongs to a peer that never answered: its frames would +// only ever be sent stale (a SYN for a dial abandoned minutes ago), and +// the queue holds one of the maxPendingPeers slots that every first +// contact with a new peer needs. +const pendingMaxAge = 2 * time.Minute + +// reapStalePending drops pending queues older than pendingMaxAge. Before +// it, a queue was removed only when its peer's key exchange completed +// (flushPending): queues for peers that never answered stayed for the +// daemon's lifetime, and once maxPendingPeers of them had piled up, every +// send to a peer without a session failed with "too many pending key +// exchanges". Called from the idleSweepLoop tick. +func (tm *TunnelManager) reapStalePending(now time.Time) int { + dropped := 0 + tm.pendMu.Lock() + for nodeID, since := range tm.pendingSince { + if now.Sub(since) > pendingMaxAge { + delete(tm.pending, nodeID) + delete(tm.pendingSince, nodeID) + dropped++ + } + } + // A queue with no start time cannot age out; give it one. + for nodeID := range tm.pending { + if _, ok := tm.pendingSince[nodeID]; !ok { + tm.pendingSince[nodeID] = now + } + } + tm.pendMu.Unlock() + return dropped +} + +// dropPending discards a peer's pending queue and its log throttle, for a +// peer the daemon is forgetting (see Daemon.forgetPeer). RemovePeer keeps +// the queue on purpose: a path reset removes the peer and re-keys at once, +// and the queued frames go out when the new key is in. +func (tm *TunnelManager) dropPending(nodeID uint32) { + tm.pendMu.Lock() + delete(tm.pending, nodeID) + delete(tm.pendingSince, nodeID) + delete(tm.lastPendDropLog, nodeID) + tm.pendMu.Unlock() +} + +// peerStateEntries counts the entries in every per-peer table the tunnel +// and its routing and key-exchange managers keep, by table. Tests use it to +// check that per-peer state goes when the peer does. +func (tm *TunnelManager) peerStateEntries() map[string]int { + out := map[string]int{} + tm.mu.RLock() + out["peers"] = len(tm.peers) + tm.mu.RUnlock() + out["crypto"] = tm.envelope.Len() + tm.pendMu.Lock() + out["pending"] = len(tm.pending) + len(tm.pendingSince) + out["pending_drop_log"] = len(tm.lastPendDropLog) + tm.pendMu.Unlock() + keyViaRelay := 0 + tm.keyViaRelay.Range(func(_, _ any) bool { keyViaRelay++; return true }) + out["key_via_relay"] = keyViaRelay + if tm.activity != nil { + tm.activity.mu.RLock() + out["activity"] = len(tm.activity.m) + tm.activity.mu.RUnlock() + } + out["routing"] = tm.routing.PeerStateEntries() + out["keyexchange"] = tm.kx.PeerStateEntries() + return out +} + // reapPendDropLog evicts lastPendDropLog entries older than // pendingDropLogInterval. Without periodic reaping, every peer that // ever triggered a queue-full warn keeps a permanent entry — a slow @@ -2032,6 +2109,7 @@ func (tm *TunnelManager) flushPending(nodeID uint32) { tm.pendMu.Lock() frames := tm.pending[nodeID] delete(tm.pending, nodeID) + delete(tm.pendingSince, nodeID) tm.pendMu.Unlock() if len(frames) == 0 { @@ -2273,6 +2351,9 @@ func (tm *TunnelManager) SendTo(addr *net.UDPAddr, nodeID uint32, pkt *protocol. dropped = true atomic.AddUint64(&tm.PendingDrops, 1) } else { + if len(q) == 0 { + tm.pendingSince[nodeID] = time.Now() + } tm.pending[nodeID] = append(q, data) } qlen := len(tm.pending[nodeID]) diff --git a/pkg/daemon/zz_peer_churn_bounded_test.go b/pkg/daemon/zz_peer_churn_bounded_test.go new file mode 100644 index 00000000..3902bee1 --- /dev/null +++ b/pkg/daemon/zz_peer_churn_bounded_test.go @@ -0,0 +1,246 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "crypto/ecdh" + "crypto/rand" + "net" + "runtime" + "testing" + "time" + + "github.com/pilot-protocol/common/protocol" +) + +// churnDaemon is the part of a daemon that holds per-peer state: an +// encrypted tunnel on a real socket, and a port table for the reaper's +// open-connection check. +func churnDaemon(t *testing.T) *Daemon { + t.Helper() + d := &Daemon{ + tunnels: newEncryptedTunnel(t), + ports: NewPortManager(), + startTime: time.Now(), + stopCh: make(chan struct{}), + } + d.tunnels.SetOpenConnPeers(d.ports.ActiveNodeIDs) + return d +} + +// addChurnPeer gives the daemon a peer the way a client's first request +// does: a key exchange (a real AES-GCM session) and some application +// traffic. Its frames go to sink, which nobody reads. +func addChurnPeer(t *testing.T, d *Daemon, nodeID uint32, sink *net.UDPAddr) { + t.Helper() + priv, err := ecdh.X25519().GenerateKey(rand.Reader) + if err != nil { + t.Fatal(err) + } + pc, err := d.tunnels.deriveSecret(priv.PublicKey().Bytes()) + if err != nil { + t.Fatal(err) + } + pc.Ready = true + pc.Authenticated = true + d.tunnels.AddPeer(nodeID, sink) + d.tunnels.envelope.Install(nodeID, pc) + d.tunnels.noteAppActivity(nodeID) // as onKeyInstalled does for a new session + d.tunnels.kx.RecordInboundDecrypt(nodeID) + _ = d.tunnels.Send(nodeID, &protocol.Packet{ // the reply + Version: protocol.Version, + Protocol: protocol.ProtoStream, + Src: protocol.Addr{Node: 0xAA000002}, + Dst: protocol.Addr{Node: nodeID}, + DstPort: protocol.PortDataExchange, + Payload: []byte("reply"), + }) +} + +// ageChurnPeer moves a peer's clocks back as if it had used the daemon long +// ago and nothing had been sent either way since: the daemon's own +// periodic jobs decide what happens next. +func ageChurnPeer(d *Daemon, nodeID uint32, ago time.Time) { + d.tunnels.activity.note(nodeID, ago) + d.tunnels.routing.RecordOutboundSend(nodeID, ago) + d.tunnels.routing.RecordDirectRecv(nodeID, ago) + d.tunnels.kx.SetLastInboundDecryptForTest(nodeID, ago) +} + +func totalEntries(m map[string]int) int { + n := 0 + for _, v := range m { + n += v + } + return n +} + +func liveHeap() uint64 { + runtime.GC() + runtime.GC() + var ms runtime.MemStats + runtime.ReadMemStats(&ms) + return ms.HeapAlloc +} + +// churnRounds runs the service-agent workload that grew the fleet's +// daemons from ~26 MB to ~330 MB: every round perRound new clients make one +// request each and go quiet. Between rounds, time passes — every earlier +// client has been quiet for longer than PeerIdleAfter plus the reaper's +// five minutes — and the daemon runs its periodic jobs as idleSweepLoop +// and keepaliveLoop do: the NAT keepalive sweep, then the stale-peer +// reaper, then the pending-queue sweep. It returns the per-peer table +// sizes and the live heap after each round. +func churnRounds(t *testing.T, rounds, perRound int) (peers []int, entries []int, heap []uint64) { + t.Helper() + d := churnDaemon(t) + sink, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { sink.Close() }) + sinkAddr := sink.LocalAddr().(*net.UDPAddr) + + var ids []uint32 + next := uint32(0x00100000) + for r := 0; r < rounds; r++ { + longAgo := time.Now().Add(-(PeerIdleAfter + 6*time.Minute)) + for _, id := range ids { + if d.tunnels.HasPeer(id) { + ageChurnPeer(d, id, longAgo) + } + } + for i := 0; i < perRound; i++ { + next++ + addChurnPeer(t, d, next, sinkAddr) + ids = append(ids, next) + } + now := time.Now() + d.tunnels.keepaliveSweep(now) + d.reapStalePeers() + d.tunnels.reapStalePending(now) + + sizes := d.tunnels.peerStateEntries() + peers = append(peers, sizes["peers"]) + entries = append(entries, totalEntries(sizes)) + heap = append(heap, liveHeap()) + } + return peers, entries, heap +} + +// Per-peer state stays at about one round of clients however many rounds +// run. Before the idle-peer change the NAT keepalive refreshed every +// peer's last contact, the stale-peer reaper never fired, and the tables +// (and the heap) grew by a round's worth every round; the control below +// runs that behaviour and must show the growth, so this test fails if +// the reaper stops firing again. +func TestPeerChurnStateStaysBounded(t *testing.T) { + // Not parallel: PeerIdleAfter is package state. + const rounds, perRound = 12, 400 + + t.Run("idle peers are reaped", func(t *testing.T) { + peers, entries, heap := churnRounds(t, rounds, perRound) + t.Logf("peers per round: %v", peers) + t.Logf("per-peer table entries per round: %v", entries) + t.Logf("live heap per round (KB): %v", heapKB(heap)) + for r, n := range peers { + if n > perRound { + t.Fatalf("round %d: %d peers held, want at most the %d of the latest round", r, n, perRound) + } + } + // Every table drops back with its peers: the last round holds no + // more entries than the first did. + if last, first := entries[len(entries)-1], entries[0]; last > first+first/10 { + t.Fatalf("per-peer tables grew from %d to %d entries across %d rounds of churn", first, last, rounds) + } + // The heap stays flat too: well under what the peers reaped since + // round 2 would cost if they were still held (a few KB each). + if grew := int64(heap[len(heap)-1]) - int64(heap[1]); grew > int64(perRound)*1024 { + t.Fatalf("live heap grew by %d KB over %d rounds; the peers are reaped, so it should be flat", grew/1024, rounds-2) + } + }) + + t.Run("control: upkeep forever keeps every peer", func(t *testing.T) { + prev := PeerIdleAfter + PeerIdleAfter = 0 // no peer is ever idle: the keepalive behaviour before the change + t.Cleanup(func() { PeerIdleAfter = prev }) + peers, _, heap := churnRounds(t, rounds, perRound) + t.Logf("peers per round: %v", peers) + t.Logf("live heap per round (KB): %v", heapKB(heap)) + if got, want := peers[len(peers)-1], rounds*perRound; got != want { + t.Fatalf("with keepalives to every peer, %d peers held after %d rounds; want all %d (the reaper must not fire)", got, rounds, want) + } + }) +} + +func heapKB(h []uint64) []uint64 { + out := make([]uint64, len(h)) + for i, v := range h { + out[i] = v / 1024 + } + return out +} + +// A queue of frames waiting for a key exchange the peer never answers is +// dropped after pendingMaxAge, and when the reaper forgets the peer. +// Before, it stayed until the daemon exited: dead peers filled all +// maxPendingPeers slots and every first send to a new peer then failed +// with "too many pending key exchanges". +func TestPendingQueueForSilentPeerDoesNotOutliveIt(t *testing.T) { + t.Parallel() + tm := newEncryptedTunnel(t) + sink, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { sink.Close() }) + addr := sink.LocalAddr().(*net.UDPAddr) + + syn := func(nodeID uint32) error { + return tm.Send(nodeID, &protocol.Packet{ + Version: protocol.Version, + Flags: protocol.FlagSYN, + Protocol: protocol.ProtoStream, + Src: protocol.Addr{Node: 0xAA000002}, + Dst: protocol.Addr{Node: nodeID}, + DstPort: protocol.PortDataExchange, + }) + } + // Fill every pending slot with peers that will never answer. + for i := uint32(0); i < maxPendingPeers; i++ { + tm.AddPeer(0x00200000+i, addr) + if err := syn(0x00200000 + i); err != nil { + t.Fatalf("queueing for dead peer %d: %v", i, err) + } + } + const fresh = 0x00300000 + tm.AddPeer(fresh, addr) + if err := syn(fresh); err == nil { + t.Fatal("expected the pending table to be full") + } + + // Within pendingMaxAge nothing is dropped. + if n := tm.reapStalePending(time.Now()); n != 0 { + t.Fatalf("dropped %d queues before they were stale", n) + } + // Past it, the dead peers' queues go and a new peer can be reached. + if n := tm.reapStalePending(time.Now().Add(pendingMaxAge + time.Second)); n != maxPendingPeers { + t.Fatalf("dropped %d stale queues, want %d", n, maxPendingPeers) + } + if err := syn(fresh); err != nil { + t.Fatalf("send to a new peer after the sweep: %v", err) + } + if got := tm.peerStateEntries()["pending"]; got != 2 { // one queue: pending + pendingSince + t.Fatalf("pending entries = %d, want only the new peer's", got) + } + + // Forgetting a peer drops its queue at once. + d := &Daemon{tunnels: tm} + d.forgetPeer(fresh) + if got := tm.peerStateEntries()["pending"]; got != 0 { + t.Fatalf("pending entries after forgetPeer = %d, want 0", got) + } + if tm.HasPeer(fresh) { + t.Fatal("forgetPeer left the tunnel entry") + } +} From 9b28975e63dbb26969eb68e850711195a0844c1d Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 14:32:11 +0300 Subject: [PATCH 06/10] daemon: a rekey that gives up does not reset a peer nobody uses MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit onRekeyGaveUp resets the peer's path (resolve, punch, a fresh PILA). For a peer that went away the new key exchange gives up too, and the next reset follows once gaveUpResetCooldown (30 s) has passed. With a few hundred such peers in rekey, the keyexchange loop's cap of 64 retransmits per 4 s tick stretched each peer's five attempts past that cooldown, so the cycle never ended: on 2026-10-08, 90 minutes after a restart, catfact-ninja had logged 25,147 "rekey gave up (session desync) — reset peer path" and hn-algolia-search 8,267. Each round is a registry Resolve in a goroutine, a punch and five PILAs, and those PILAs stamp lastOutboundSend, so the stale-peer reaper never dropped these peers either. The idle-peer change stopped keepalives, path probes and upgrades to unused peers, but not this. - onRekeyGaveUp resets only a peer in use: an open connection, or application traffic within PeerIdleAfter (peerInUse). A peer with no activity record is not in use here, unlike for the keepalive and upgrade loops (idleFilter), where only ready sessions are considered and every new session records activity: one it only ever got a key request from, or never heard from, is exactly the case to leave alone. Left alone, it goes quiet and the reaper drops it; whoever uses it next re-keys it. - resetPeerPath keeps the peer's activity record across its RemovePeer: a reset re-keys the same peer. Dropping the record made a reset peer look untracked, which the keepalive and upgrade loops treat as active. TestOnRekeyGaveUpResetsWithCooldown now marks its peer in use. New tests: TestRekeyGaveUpLeavesUnusedPeerToReaper (fails without the gate) and TestResetPeerPathKeepsActivity. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 11 ++++ pkg/daemon/pathwatch.go | 23 +++++++ pkg/daemon/peeractivity.go | 17 +++++ pkg/daemon/zz_pathwatch_test.go | 3 +- .../zz_rekey_gaveup_unused_peer_test.go | 66 +++++++++++++++++++ 5 files changed, 119 insertions(+), 1 deletion(-) create mode 100644 pkg/daemon/zz_rekey_gaveup_unused_peer_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a36663c..e21a6dd3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -363,6 +363,17 @@ Detailed per-release notes are on the 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. - **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 diff --git a/pkg/daemon/pathwatch.go b/pkg/daemon/pathwatch.go index 2d0920e1..44a568a1 100644 --- a/pkg/daemon/pathwatch.go +++ b/pkg/daemon/pathwatch.go @@ -257,9 +257,19 @@ func (d *Daemon) resetPeerPath(nodeID uint32) peerPathReset { d.forgetPeerResolution(nodeID) + // A reset re-keys the same peer; it does not forget who uses it. + // RemovePeer drops the activity record with the rest, so put it back. + var lastUse time.Time + var hadUse bool + if d.tunnels.activity != nil { + lastUse, hadUse = d.tunnels.activity.last(nodeID) + } if res.HadTunnel { d.tunnels.RemovePeer(nodeID) } + if hadUse { + d.tunnels.activity.note(nodeID, lastUse) + } if res.WasRelayActive { d.tunnels.SetRelayPeer(nodeID, true) } @@ -300,6 +310,19 @@ const gaveUpResetCooldown = 30 * time.Second // the keyexchange retransmit loop goroutine. func (d *Daemon) onRekeyGaveUp(nodeID uint32) { now := time.Now() + // Only for a peer someone is using. Resetting an unused one starts a + // new key exchange, and when it never answers (it went away) that ends + // in another give-up and another reset. Each round's key-exchange + // frames count as contact, so the stale-peer reaper never dropped such + // a peer, and with many of them waiting the per-tick retransmit cap + // stretched a round past gaveUpResetCooldown, so the cooldown did not + // stop it: service agents logged up to 25,000 of these resets an hour + // for clients long gone. Left alone, the peer goes quiet and the reaper + // drops it; whoever uses it next re-keys it, as first contact does. + if !d.tunnels.peerInUse(nodeID, now) { + slog.Debug("rekey gave up on an unused peer — leaving it to the reaper", "peer_node_id", nodeID) + return + } d.gaveUpResetMu.Lock() if last, ok := d.lastGaveUpReset[nodeID]; ok && now.Sub(last) < gaveUpResetCooldown { d.gaveUpResetMu.Unlock() diff --git a/pkg/daemon/peeractivity.go b/pkg/daemon/peeractivity.go index 718c6ea3..b06cd984 100644 --- a/pkg/daemon/peeractivity.go +++ b/pkg/daemon/peeractivity.go @@ -108,6 +108,23 @@ func (tm *TunnelManager) idleFilter(now time.Time) func(nodeID uint32) bool { } } +// peerInUse reports whether a peer has an open connection or carried +// application traffic within PeerIdleAfter. Unlike idleFilter, a peer with +// no recorded activity is not in use: it is one this node only ever +// exchanged upkeep with (a key request it never answered, a frame it could +// not decrypt), or one a path reset has just re-keyed. With PeerIdleAfter +// off every peer is in use, as before idleness was tracked. +func (tm *TunnelManager) peerInUse(nodeID uint32, now time.Time) bool { + if PeerIdleAfter <= 0 || tm.activity == nil { + return true + } + if tm.openConnPeers != nil && tm.openConnPeers()[nodeID] { + return true + } + last, ok := tm.activity.last(nodeID) + return ok && now.Sub(last) <= PeerIdleAfter +} + // PeerIdle reports whether one peer is idle now. func (tm *TunnelManager) PeerIdle(nodeID uint32) bool { return tm.idleFilter(time.Now())(nodeID) diff --git a/pkg/daemon/zz_pathwatch_test.go b/pkg/daemon/zz_pathwatch_test.go index 671efa01..b54c2b45 100644 --- a/pkg/daemon/zz_pathwatch_test.go +++ b/pkg/daemon/zz_pathwatch_test.go @@ -252,7 +252,8 @@ func TestOnRekeyGaveUpResetsWithCooldown(t *testing.T) { t.Cleanup(func() { gaveUpResetPeer = prev }) const peer = 77 - d.onRekeyGaveUp(peer) // fires an async reset + d.tunnels.noteAppActivity(peer) // a peer someone is using + d.onRekeyGaveUp(peer) // fires an async reset // second giveup immediately: cooldown must suppress it d.onRekeyGaveUp(peer) diff --git a/pkg/daemon/zz_rekey_gaveup_unused_peer_test.go b/pkg/daemon/zz_rekey_gaveup_unused_peer_test.go new file mode 100644 index 00000000..a7623b93 --- /dev/null +++ b/pkg/daemon/zz_rekey_gaveup_unused_peer_test.go @@ -0,0 +1,66 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "net" + "sync/atomic" + "testing" + "time" +) + +// A peer nobody uses is not reset when its rekey gives up: the reset's own +// key exchange, never answered by a peer that went away, ended in another +// give-up, and its frames kept the reaper from ever dropping the peer. +// A peer in use still is (TestOnRekeyGaveUpResetsWithCooldown). +func TestRekeyGaveUpLeavesUnusedPeerToReaper(t *testing.T) { + var resets atomic.Int32 + prev := gaveUpResetPeer + gaveUpResetPeer = func(_ *Daemon, _ uint32) peerPathReset { + resets.Add(1) + return peerPathReset{} + } + t.Cleanup(func() { gaveUpResetPeer = prev }) + + d := churnDaemon(t) + d.lastGaveUpReset = make(map[uint32]time.Time) + const unknown, idle = 0x00400001, 0x00400002 + addr := &net.UDPAddr{IP: net.IPv4(203, 0, 113, 9), Port: 4000} + d.tunnels.AddPeer(unknown, addr) // only ever sent a key request + d.tunnels.AddPeer(idle, addr) + d.tunnels.activity.note(idle, time.Now().Add(-PeerIdleAfter-time.Minute)) + + d.onRekeyGaveUp(unknown) + d.onRekeyGaveUp(idle) + time.Sleep(100 * time.Millisecond) // a reset would run in a goroutine + if n := resets.Load(); n != 0 { + t.Fatalf("%d path resets for peers nobody uses; want none", n) + } + + // With nothing sent to them any more, the reaper drops both. + long := time.Now().Add(-10 * time.Minute) + for _, id := range []uint32{unknown, idle} { + d.tunnels.routing.RecordOutboundSend(id, long) + } + d.reapStalePeers() + if d.tunnels.HasPeer(unknown) || d.tunnels.HasPeer(idle) { + t.Fatal("unused peers were not reaped once nothing was sent to them") + } +} + +// A path reset re-keys a peer; it keeps the record of when the peer was +// last used. Losing it made every reset peer look unused (or, to the +// keepalive and upgrade loops, untracked and therefore active). +func TestResetPeerPathKeepsActivity(t *testing.T) { + const peer = 48 + d := newPathWatchTestDaemon(t, peer) + used := time.Now().Add(-30 * time.Second) + d.tunnels.activity.note(peer, used) + + d.resetPeerPath(peer) + + last, ok := d.tunnels.activity.last(peer) + if !ok || !last.Equal(time.Unix(0, used.UnixNano())) { + t.Fatalf("activity after reset = %v, %v; want %v", last, ok, used) + } +} From 4da2cc6a304fe7f9d9352249ab4a23f5c3fa30dd Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 14:55:48 +0300 Subject: [PATCH 07/10] daemon: reap per-peer state left for nodes with no tunnel reapStalePeers walks the tunnel's peer table, so state kept for a node that has no entry there is never reaped. A path reset produces it: resetPeerPath removes the peer, puts its relay flag back so the recovery PILA travels by relay, and re-adds the peer only if ensureTunnel's registry resolve succeeds. When it fails (the peer went away, or the registry is slow), the relay flag stays with no tunnel behind it. The relay probe loop then attempted a direct upgrade for it every 15 s, and it counted against MaxRelayPeers (4,096) for the daemon's lifetime; a service agent holds ~3,600 relayed peers, so the cap is within reach, and past it new relayed peers are dropped ("relay peers cap reached"). In a local soak run of this branch (silent clients, 23 minutes), health showed 875 relay flags for 381 peers. - reapStalePeers also sweeps every node that routing, key exchange or the session store holds state for (new PeerIDs listings) but that has no tunnel entry and no open connection: it is forgotten once its last contact is older than peerReapIdleTimeout, or, with no contact ever recorded, peerReapIdleTimeout after the sweep first finds it (orphanSeen). - The last-contact computation is shared (lastPeerContact) and the timeout is a package constant. - relayProbeLoop skips relay flags with no tunnel peer: there is no session to upgrade. TestReaperForgetsOrphanedPeerState covers both kinds of orphan and that state with recent contact stays. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 8 ++ pkg/daemon/daemon.go | 99 ++++++++++++++++++++---- pkg/daemon/keyexchange/keyexchange.go | 33 ++++++++ pkg/daemon/routing/relay.go | 28 +++++++ pkg/daemon/zz_peer_churn_bounded_test.go | 43 ++++++++++ 5 files changed, 195 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e21a6dd3..dfcd7006 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -374,6 +374,14 @@ Detailed per-release notes are on the 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. - **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 diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index dff3f99c..09786083 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -579,6 +579,11 @@ type Daemon struct { netPolicyMu sync.RWMutex netPolicies map[uint16][]uint16 + // orphanSeen records when reapOrphanedPeerState first found per-peer + // state, with no contact ever recorded, for a node with no tunnel + // entry. Only the idle sweep (reapStalePeers) touches it. + orphanSeen map[uint32]time.Time + // gaveUpResetMu guards lastGaveUpReset, the per-peer cooldown for // rekey-gave-up-triggered path resets (onRekeyGaveUp). gaveUpResetMu sync.Mutex @@ -6451,14 +6456,30 @@ func (d *Daemon) restoreRegistryState(nodeID uint32, trust bool) bool { return complete } +// peerReapIdleTimeout is how long a peer with no open connection may go +// without contact before reapStalePeers forgets it. +const peerReapIdleTimeout = 5 * time.Minute + +// lastPeerContact is the latest frame either way with a peer: our last +// send, its last direct packet, or any authenticated inbound frame (relayed +// ones too, which LastDirectRecv skips). Zero when none was ever recorded. +func (d *Daemon) lastPeerContact(nodeID uint32) time.Time { + latest := d.tunnels.LastDirectRecv(nodeID) + if lastOutbound, ok := d.tunnels.LastOutboundSend(nodeID); ok && lastOutbound.After(latest) { + latest = lastOutbound + } + if lastIn, ok := d.tunnels.LastInboundDecrypt(nodeID); ok && lastIn.After(latest) { + latest = lastIn + } + return latest +} + // reapStalePeers removes tunnel peers that have no active connections // and haven't been contacted for peerReapIdleTimeout. Called periodically // from idleSweepLoop to prevent unbounded growth of per-peer maps // (tm.peers, routing relay/blackhole/send-err maps, keyexchange // pubkey/rekey state) in long-running daemons with peer churn. func (d *Daemon) reapStalePeers() { - const peerReapIdleTimeout = 5 * time.Minute - active := d.ports.ActiveNodeIDs() now := time.Now() peers := d.tunnels.PeerList() @@ -6468,20 +6489,10 @@ func (d *Daemon) reapStalePeers() { continue // has active connection — keep } - lastDirect := d.tunnels.LastDirectRecv(p.NodeID) - lastOutbound, ok := d.tunnels.LastOutboundSend(p.NodeID) - - // Find the latest known contact. Any authenticated inbound frame - // counts, relayed ones too (LastDirectRecv skips those): a peer - // still talking to us is not stale, and dropping it would only - // make its next frame trigger a rekey. - latest := lastDirect - if ok && lastOutbound.After(latest) { - latest = lastOutbound - } - if lastIn, ok := d.tunnels.LastInboundDecrypt(p.NodeID); ok && lastIn.After(latest) { - latest = lastIn - } + // Any authenticated inbound frame counts as contact, relayed ones + // too: a peer still talking to us is not stale, and dropping it + // would only make its next frame trigger a rekey. + latest := d.lastPeerContact(p.NodeID) if latest.IsZero() { // Never contacted — fresh peer entry, don't reap yet. @@ -6495,6 +6506,57 @@ func (d *Daemon) reapStalePeers() { d.forgetPeer(p.NodeID) } } + d.reapOrphanedPeerState(now, active) +} + +// reapOrphanedPeerState forgets routing, key-exchange and session state +// held for a node that has no tunnel entry, so the loop above never sees +// it. A path reset leaves such state when the peer cannot be re-resolved: +// it removes the peer, puts its relay flag back for the recovery PILA, and +// re-adds the peer only if the registry answers. The orphaned relay flag +// then kept the node in the 15 s direct-upgrade loop and counted against +// MaxRelayPeers for the daemon's lifetime (a local soak: 875 relay flags +// for 381 peers after 23 minutes). State with no contact recorded at all +// is given peerReapIdleTimeout from when this sweep first finds it. +func (d *Daemon) reapOrphanedPeerState(now time.Time, active map[uint32]bool) { + ids := map[uint32]struct{}{} + for _, id := range d.tunnels.routing.PeerIDs() { + ids[id] = struct{}{} + } + for _, id := range d.tunnels.kx.PeerIDs() { + ids[id] = struct{}{} + } + for _, id := range d.tunnels.envelope.PeerIDs() { + ids[id] = struct{}{} + } + if d.orphanSeen == nil { + d.orphanSeen = make(map[uint32]time.Time) + } + for id := range d.orphanSeen { + if _, ok := ids[id]; !ok { + delete(d.orphanSeen, id) + } + } + for id := range ids { + if active[id] || d.tunnels.HasPeer(id) { + delete(d.orphanSeen, id) + continue + } + latest := d.lastPeerContact(id) + if latest.IsZero() { + first, seen := d.orphanSeen[id] + if !seen { + d.orphanSeen[id] = now + continue + } + latest = first + } + if now.Sub(latest) > peerReapIdleTimeout { + slog.Debug("reaping orphaned peer state", "node_id", id, "last_contact", latest.Round(time.Second)) + d.forgetPeer(id) + delete(d.orphanSeen, id) + } + } } // forgetPeer drops everything the daemon keeps for a peer it no longer @@ -6679,6 +6741,11 @@ func (d *Daemon) relayProbeLoop() { // and five probes; nobody is using this path. continue } + if !d.tunnels.HasPeer(nodeID) { + // A relay flag with no tunnel behind it (see + // reapOrphanedPeerState): no session to upgrade. + continue + } go d.tryDirectUpgrade(nodeID) } } diff --git a/pkg/daemon/keyexchange/keyexchange.go b/pkg/daemon/keyexchange/keyexchange.go index 62192d3f..a55d4181 100644 --- a/pkg/daemon/keyexchange/keyexchange.go +++ b/pkg/daemon/keyexchange/keyexchange.go @@ -753,6 +753,39 @@ func (m *Manager) PeerStateEntries() int { return n } +// PeerIDs lists every node this manager holds per-peer state for, except +// rekeyGaveUp, which expires on its own. +func (m *Manager) PeerIDs() []uint32 { + seen := map[uint32]struct{}{} + m.pubKeysMu.RLock() + for id := range m.peerPubKeys { + seen[id] = struct{}{} + } + for id := range m.negPubKeys { + seen[id] = struct{}{} + } + m.pubKeysMu.RUnlock() + m.rkPendingMu.Lock() + for id := range m.pendingRekey { + seen[id] = struct{}{} + } + for id := range m.lastInboundDecrypt { + seen[id] = struct{}{} + } + for id := range m.lastKeyExchangeReply { + seen[id] = struct{}{} + } + for id := range m.keyRequestDue { + seen[id] = struct{}{} + } + m.rkPendingMu.Unlock() + ids := make([]uint32, 0, len(seen)) + for id := range seen { + ids = append(ids, id) + } + return ids +} + // RemovePeer wipes per-peer L5 state (called from TunnelManager.RemovePeer). func (m *Manager) RemovePeer(nodeID uint32) { m.pubKeysMu.Lock() diff --git a/pkg/daemon/routing/relay.go b/pkg/daemon/routing/relay.go index 425a03af..d8584c0c 100644 --- a/pkg/daemon/routing/relay.go +++ b/pkg/daemon/routing/relay.go @@ -4,6 +4,7 @@ package routing import ( "net" + "time" ) // SetRelayPeer marks a peer as needing relay through the beacon. Pinning @@ -179,6 +180,33 @@ func (m *Manager) PeerStateEntries() int { len(m.blackholeMissCount) + len(m.directClearCount) } +// PeerIDs lists every node this manager holds per-peer state for. +func (m *Manager) PeerIDs() []uint32 { + m.mu.RLock() + defer m.mu.RUnlock() + seen := make(map[uint32]struct{}, len(m.lastOutboundSend)) + for _, t := range []map[uint32]time.Time{m.lastOutboundSend, m.firstOutboundSend, m.lastDirectRecv} { + for id := range t { + seen[id] = struct{}{} + } + } + for _, b := range []map[uint32]bool{m.relayPeers, m.relayPinned} { + for id := range b { + seen[id] = struct{}{} + } + } + for _, c := range []map[uint32]int{m.sendErrCount, m.blackholeMissCount, m.directClearCount} { + for id := range c { + seen[id] = struct{}{} + } + } + ids := make([]uint32, 0, len(seen)) + for id := range seen { + ids = append(ids, id) + } + return ids +} + // RemovePeer wipes per-peer L4 state. func (m *Manager) RemovePeer(nodeID uint32) { m.mu.Lock() diff --git a/pkg/daemon/zz_peer_churn_bounded_test.go b/pkg/daemon/zz_peer_churn_bounded_test.go index 3902bee1..fc24cd6a 100644 --- a/pkg/daemon/zz_peer_churn_bounded_test.go +++ b/pkg/daemon/zz_peer_churn_bounded_test.go @@ -244,3 +244,46 @@ func TestPendingQueueForSilentPeerDoesNotOutliveIt(t *testing.T) { t.Fatal("forgetPeer left the tunnel entry") } } + +// Per-peer state for a node with no tunnel entry is reaped too: a path +// reset that cannot re-resolve its peer leaves the relay flag it restored +// (and whatever else the peer had), which the loop over tunnel peers never +// visits. State with recent contact stays; state with none recorded gets +// peerReapIdleTimeout from when the sweep first finds it. +func TestReaperForgetsOrphanedPeerState(t *testing.T) { + d := churnDaemon(t) + now := time.Now() + const flagOnly, longQuiet, recent = 0x00500001, 0x00500002, 0x00500003 + d.tunnels.SetRelayPeer(flagOnly, true) // what a failed path reset leaves + d.tunnels.SetRelayPeer(longQuiet, true) + d.tunnels.routing.RecordOutboundSend(longQuiet, now.Add(-10*time.Minute)) + d.tunnels.kx.SetLastInboundDecryptForTest(longQuiet, now.Add(-10*time.Minute)) + d.tunnels.routing.RecordOutboundSend(recent, now) + + d.reapStalePeers() + if d.tunnels.IsRelayPeer(longQuiet) { + t.Fatal("state for a node with no tunnel, quiet for 10 minutes, was kept") + } + if _, ok := d.tunnels.LastInboundDecrypt(longQuiet); ok { + t.Fatal("key-exchange state for the quiet orphan was kept") + } + if !d.tunnels.IsRelayPeer(flagOnly) { + t.Fatal("an orphan with no recorded contact was dropped the first time it was seen") + } + if _, ok := d.tunnels.LastOutboundSend(recent); !ok { + t.Fatal("state with contact just now was dropped") + } + + // peerReapIdleTimeout after it was first seen, the flag-only orphan goes. + d.orphanSeen[flagOnly] = now.Add(-peerReapIdleTimeout - time.Second) + d.reapStalePeers() + if d.tunnels.IsRelayPeer(flagOnly) { + t.Fatal("an orphaned relay flag outlived peerReapIdleTimeout") + } + if ids := d.tunnels.routing.PeerIDs(); len(ids) != 1 || ids[0] != recent { + t.Fatalf("routing holds state for %v, want only the recent node %d", ids, recent) + } + if len(d.orphanSeen) != 0 { + t.Fatalf("orphanSeen still holds %d entries", len(d.orphanSeen)) + } +} From e4aff229e9b2a4b6da14dc93c96172a4a5f91eda Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 15:02:00 +0300 Subject: [PATCH 08/10] tests: soak test for an agent's memory under client churn (off by default) TestSoakSwarmAgentMemory runs one agent daemon as its own process, from PILOT_SOAK_AGENT_BIN with the service-agent fleet's flags (-encrypt -public -trust-auto-approve -idle-timeout 3600s -keepalive 300s) and -pprof, against an in-process registry and beacon. A swarm of in-process clients churns through it the way the fleet's traffic does: each client trust-handshakes, sends a message and gets a reply the way the fleet's responder sends one (a handshake, then a dial back to port 1001). Some stay up idle for the whole run; some register an endpoint nobody listens on, so the agent reaches them over the relay. Every sample records the agent's live heap after a forced GC, goroutines, RSS, peers, relayed peers and connections to a CSV; it saves a heap profile at peak RSS, a goroutine dump the first time goroutines pass 2,000, and a heap profile and goroutine dump at the end. Knobs are in the doc comment. It skips unless PILOT_SOAK_AGENT_BIN is set. Before and after the idle-peer and reaping changes on this branch (30 minutes, 4 new clients per second for the first 20, 60 s client lifetime): main: peers grew one for one with clients ever seen; rekey give-ups reset dead peers' paths ~5 times a second; the registry calls behind them piled up to 21,149 goroutines; RSS 238 MB at 11 minutes and 436 MB at 23. this branch: peers stayed at the clients of the last few minutes and fell after churn stopped (126 at the end); goroutines 38-120; RSS 48-57 MB. Co-Authored-By: Claude Opus 5.5 --- tests/zz_soak_swarm_memory_test.go | 496 +++++++++++++++++++++++++++++ 1 file changed, 496 insertions(+) create mode 100644 tests/zz_soak_swarm_memory_test.go diff --git a/tests/zz_soak_swarm_memory_test.go b/tests/zz_soak_swarm_memory_test.go new file mode 100644 index 00000000..88e333fe --- /dev/null +++ b/tests/zz_soak_swarm_memory_test.go @@ -0,0 +1,496 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package tests + +import ( + "bufio" + "context" + "fmt" + "io" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "regexp" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/pilot-protocol/common/driver" + "github.com/pilot-protocol/common/protocol" + "github.com/pilot-protocol/dataexchange" + "github.com/pilot-protocol/pilotprotocol/pkg/daemon" + acceptpkg "github.com/pilot-protocol/rendezvous/accept" + pluginsruntime "github.com/pilot-protocol/runtime" +) + +// TestSoakSwarmAgentMemory reproduces the service-agent memory growth: one +// agent daemon (a separate process started from PILOT_SOAK_AGENT_BIN with +// the fleet's flags and -pprof) serves a churning swarm of short-lived +// clients. Each client trust-handshakes, sends a message, and gets a reply +// the way the fleet's responder sends one (handshake, then a dial back to +// port 1001). Some clients stay up and idle for the whole run, some register +// an endpoint nobody listens on so the agent reaches them over the relay. +// +// Every PILOT_SOAK_SAMPLE the test records the agent's live heap (after a +// forced GC), goroutines, RSS, peers and connections, and at the end saves +// a heap profile and a goroutine dump next to PILOT_SOAK_OUT. Off unless +// PILOT_SOAK_AGENT_BIN is set. Knobs (environment): +// +// PILOT_SOAK_DURATION run length (20m) +// PILOT_SOAK_STOP_NEW_AFTER no new clients after this (the duration); the +// rest of the run shows what the agent releases +// PILOT_SOAK_RATE new clients per second (2) +// PILOT_SOAK_LIFETIME how long a client stays up (90s) +// PILOT_SOAK_STAY_FRACTION clients that stay up, idle, all run (0.15) +// PILOT_SOAK_RELAY_FRACTION clients the agent must reach by relay (0.5) +// PILOT_SOAK_SWARM_IDLE_AFTER the clients' PeerIdleAfter; 0 keeps their +// sessions warm forever like older daemons +// PILOT_SOAK_AGENT_GODEBUG GODEBUG for the agent, e.g. memprofilerate=4096 +// +// go build -o /tmp/agent ./cmd/daemon +// PILOT_SOAK_AGENT_BIN=/tmp/agent PILOT_SOAK_OUT=/tmp/soak.csv \ +// go test ./tests -run TestSoakSwarmAgentMemory -timeout 0 -v +func TestSoakSwarmAgentMemory(t *testing.T) { + bin := os.Getenv("PILOT_SOAK_AGENT_BIN") + if bin == "" { + t.Skip("set PILOT_SOAK_AGENT_BIN to an agent daemon binary built with -pprof") + } + dur := soakDuration("PILOT_SOAK_DURATION", 20*time.Minute) + sample := soakDuration("PILOT_SOAK_SAMPLE", 30*time.Second) + lifetime := soakDuration("PILOT_SOAK_LIFETIME", 90*time.Second) + ratePerSec := soakFloat("PILOT_SOAK_RATE", 2) + stayFrac := soakFloat("PILOT_SOAK_STAY_FRACTION", 0.15) + relayFrac := soakFloat("PILOT_SOAK_RELAY_FRACTION", 0.5) + stopNew := soakDuration("PILOT_SOAK_STOP_NEW_AFTER", dur) // churn stops here; the rest shows what is released + outPath := os.Getenv("PILOT_SOAK_OUT") + + home, err := os.MkdirTemp("/tmp", "soak-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(home) }) + // The in-process swarm's plugins write under $HOME/.pilot; keep them off + // the real one. + t.Setenv("HOME", filepath.Join(home, "swarm-home")) + daemon.TunnelKeepaliveInterval = 25 * time.Second + // The swarm runs this tree's daemon code. PILOT_SOAK_SWARM_IDLE_AFTER=0 + // keeps its sessions warm forever, as clients before the idle-peer + // change do; the default lets them fall silent when idle, as the + // fleet's v1.8.0 swarm does (it has no per-peer keepalive at all). + if v := os.Getenv("PILOT_SOAK_SWARM_IDLE_AFTER"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + daemon.PeerIdleAfter = d + } + } + + env := NewTestEnv(t) + // Every swarm node reaches the registry from 127.0.0.1; without this + // the per-IP and global request caps throttle the swarm, not the agent. + if err := env.Registry.SetRateLimitWhitelist([]acceptpkg.WhitelistEntry{{CIDR: "127.0.0.0/8", Rate: 1 << 20}}); err != nil { + t.Fatal(err) + } + + // --- agent process ------------------------------------------------- + agentHome := filepath.Join(home, "agent-home") + if err := os.MkdirAll(agentHome, 0o700); err != nil { + t.Fatal(err) + } + agentSock := filepath.Join(home, "agent.sock") + pprofPort := freeTCPPort(t) + pprofAddr := fmt.Sprintf("127.0.0.1:%d", pprofPort) + logf, err := os.Create(filepath.Join(home, "agent.log")) + if err != nil { + t.Fatal(err) + } + cmd := exec.Command(bin, + "-registry", env.RegistryAddr, "-beacon", env.BeaconAddr, + "-listen", "127.0.0.1:0", "-socket", agentSock, + "-identity", filepath.Join(agentHome, "identity.json"), + "-hostname", "soak-agent", "-email", "soak@pilot.local", + "-encrypt", "-public", "-trust-auto-approve", + "-idle-timeout", "3600s", "-keepalive", "300s", + "-no-skillinject", "-motd-feed-url", "", "-telemetry-url", "", + "-pprof", pprofAddr, + ) + cmd.Env = append(os.Environ(), "HOME="+agentHome) + if g := os.Getenv("PILOT_SOAK_AGENT_GODEBUG"); g != "" { + cmd.Env = append(cmd.Env, "GODEBUG="+g) // e.g. memprofilerate=1 for an exact heap profile + } + cmd.Stdout, cmd.Stderr = logf, logf + if err := cmd.Start(); err != nil { + t.Fatalf("start agent: %v", err) + } + t.Cleanup(func() { + _ = cmd.Process.Signal(os.Interrupt) + done := make(chan struct{}) + go func() { _ = cmd.Wait(); close(done) }() + select { + case <-done: + case <-time.After(15 * time.Second): + _ = cmd.Process.Kill() + } + logf.Close() + if outPath != "" { + _ = copyFile(filepath.Join(home, "agent.log"), outPath+".agent.log") + } + }) + var agentDrv *driver.Driver + deadline := time.Now().Add(30 * time.Second) + for { + agentDrv, err = driver.Connect(agentSock) + if err == nil { + break + } + if time.Now().After(deadline) { + t.Fatalf("agent socket: %v", err) + } + time.Sleep(200 * time.Millisecond) + } + defer agentDrv.Close() + var agentNode uint32 + for { + info, err := agentDrv.Info() + if err == nil { + if v, ok := info["node_id"].(float64); ok && v > 0 { + agentNode = uint32(v) + break + } + } + if time.Now().After(deadline) { + t.Fatalf("agent never registered: %v", err) + } + time.Sleep(200 * time.Millisecond) + } + agentAddr := protocol.Addr{Network: 0, Node: agentNode} + t.Logf("agent node %d pid %d pprof %s", agentNode, cmd.Process.Pid, pprofAddr) + + // --- responder: what pilot-agents/deploy/responder.py does ---------- + type replyTo struct{ addr protocol.Addr } + replyCh := make(chan replyTo, 4096) + var repliesOK, repliesFail, sendsOK, sendsFail, started, stopped atomic.Int64 + var respWG sync.WaitGroup + for i := 0; i < 8; i++ { + respWG.Add(1) + go func() { + defer respWG.Done() + for r := range replyCh { + _, _ = agentDrv.Handshake(r.addr.Node, "reply") + c, err := dataexchange.Dial(agentDrv, r.addr) + if err != nil { + repliesFail.Add(1) + continue + } + if err := c.SendText("reply"); err != nil { + repliesFail.Add(1) + } else { + repliesOK.Add(1) + } + c.Close() + } + }() + } + inbox := filepath.Join(agentHome, ".pilot", "inbox") + stopInboxSweep := make(chan struct{}) + go func() { + tk := time.NewTicker(5 * time.Second) + defer tk.Stop() + for { + select { + case <-stopInboxSweep: + return + case <-tk.C: + ents, _ := os.ReadDir(inbox) + for _, e := range ents { + _ = os.Remove(filepath.Join(inbox, e.Name())) + } + } + } + }() + + // --- swarm ----------------------------------------------------------- + ctx, cancel := context.WithTimeout(context.Background(), dur) + defer cancel() + var swarmWG sync.WaitGroup + var alive atomic.Int64 + idx := 0 + stopNewAt := time.Now().Add(stopNew) + go func() { + interval := time.Duration(float64(time.Second) / ratePerSec) + tk := time.NewTicker(interval) + defer tk.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tk.C: + } + if time.Now().After(stopNewAt) { + continue + } + idx++ + n := idx + stay := float64(n%100) < stayFrac*100 + relay := float64((n/7)%100) < relayFrac*100 + swarmWG.Add(1) + go func() { + defer swarmWG.Done() + sn, err := startSoakSwarmNode(env, home, n, relay) + if err != nil { + t.Logf("swarm %d: %v", n, err) + return + } + started.Add(1) + alive.Add(1) + defer func() { + sn.stop() + alive.Add(-1) + stopped.Add(1) + }() + _, _ = sn.drv.Handshake(agentNode, "soak") + c, err := dataexchange.Dial(sn.drv, agentAddr) + if err != nil { + sendsFail.Add(1) + } else { + if err := c.SendText(fmt.Sprintf("query %d", n)); err != nil { + sendsFail.Add(1) + } else { + sendsOK.Add(1) + select { + case replyCh <- replyTo{sn.d.Addr()}: + default: + } + } + c.Close() + } + life := lifetime + if stay { + life = dur + } + select { + case <-ctx.Done(): + case <-time.After(life): + } + }() + } + }() + + // --- sampler --------------------------------------------------------- + var out *os.File + if outPath != "" { + out, err = os.Create(outPath) + if err != nil { + t.Fatal(err) + } + defer out.Close() + fmt.Fprintln(out, "elapsed_s,heap_alloc_mb,heap_inuse_mb,heap_objects,goroutines,rss_mb,peers,relay_peers,encrypted_peers,conns,swarm_started,swarm_alive,sends_ok,sends_fail,replies_ok,replies_fail") + } + start := time.Now() + dumpedGoroutines := false + peakRSS := 0.0 + takeSample := func() { + ms := pprofMemStats(pprofAddr) + gr := pprofGoroutines(pprofAddr) + rss := processRSSMB(cmd.Process.Pid) + if outPath != "" && gr > 2000 && !dumpedGoroutines { + // A pile-up: keep its stacks. + dumpedGoroutines = true + _ = fetchToFile("http://"+pprofAddr+"/debug/pprof/goroutine?debug=1", outPath+".goroutines-pileup.txt") + } + if outPath != "" && rss > peakRSS { + peakRSS = rss + _ = fetchToFile("http://"+pprofAddr+"/debug/pprof/heap", outPath+".heap-peak.pb.gz") + } + h, _ := agentDrv.Health() + num := func(k string) int64 { + if v, ok := h[k].(float64); ok { + return int64(v) + } + return -1 + } + line := fmt.Sprintf("%d,%.1f,%.1f,%d,%d,%.1f,%d,%d,%d,%d,%d,%d,%d,%d,%d,%d", + int(time.Since(start).Seconds()), float64(ms["HeapAlloc"])/1e6, float64(ms["HeapInuse"])/1e6, ms["HeapObjects"], + gr, rss, num("peers"), num("relay_peer_count"), num("encrypted_peers"), num("connections"), + started.Load(), alive.Load(), sendsOK.Load(), sendsFail.Load(), repliesOK.Load(), repliesFail.Load()) + t.Log(line) + if out != nil { + fmt.Fprintln(out, line) + } + } + tk := time.NewTicker(sample) + defer tk.Stop() + takeSample() +loop: + for { + select { + case <-ctx.Done(): + break loop + case <-tk.C: + takeSample() + } + } + takeSample() + if outPath != "" { + _ = fetchToFile("http://"+pprofAddr+"/debug/pprof/heap?gc=1", outPath+".heap.pb.gz") + _ = fetchToFile("http://"+pprofAddr+"/debug/pprof/goroutine?debug=1", outPath+".goroutines.txt") + } + cancel() + swarmWG.Wait() + close(replyCh) + respWG.Wait() + close(stopInboxSweep) +} + +type soakSwarmNode struct { + d *daemon.Daemon + rt *pluginsruntime.Runtime + drv *driver.Driver +} + +func (s *soakSwarmNode) stop() { + s.drv.Close() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + _ = s.rt.StopPlugins(ctx) + cancel() + _ = s.d.Stop() +} + +// startSoakSwarmNode starts one swarm client. With relay set it registers an +// endpoint nothing listens on, so the agent's direct sends fail and it falls +// back to the beacon relay, as it does for the fleet's NATed clients. +func startSoakSwarmNode(env *TestEnv, dir string, n int, relay bool) (*soakSwarmNode, error) { + sock := filepath.Join(dir, fmt.Sprintf("s%d.sock", n)) + cfg := daemon.Config{ + RegistryAddr: env.RegistryAddr, + BeaconAddr: env.BeaconAddr, + ListenAddr: "127.0.0.1:0", + SocketPath: sock, + IdentityPath: filepath.Join(dir, fmt.Sprintf("s%d.json", n)), + Email: fmt.Sprintf("swarm-%d@pilot.local", n), + Encrypt: true, + TrustAutoApprove: true, + DisableEventStream: true, + DisablePolicyRunner: true, + } + if relay { + cfg.Endpoint = "127.0.0.1:9" + } + d := daemon.New(cfg) + rt := registerStandardPlugins(soakFatal{}, d, &cfg) + if err := rt.StartPlugins(context.Background()); err != nil { + return nil, err + } + if err := d.Start(); err != nil { + return nil, err + } + drv, err := driver.Connect(sock) + if err != nil { + _ = d.Stop() + return nil, err + } + return &soakSwarmNode{d: d, rt: rt, drv: drv}, nil +} + +type soakFatal struct{} + +func (soakFatal) Fatalf(format string, args ...interface{}) { panic(fmt.Sprintf(format, args...)) } + +func soakDuration(key string, def time.Duration) time.Duration { + if v := os.Getenv(key); v != "" { + if d, err := time.ParseDuration(v); err == nil { + return d + } + } + return def +} + +func soakFloat(key string, def float64) float64 { + if v := os.Getenv(key); v != "" { + if f, err := strconv.ParseFloat(v, 64); err == nil { + return f + } + } + return def +} + +func freeTCPPort(t *testing.T) int { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + return ln.Addr().(*net.TCPAddr).Port +} + +var memStatLine = regexp.MustCompile(`^# (\w+) = (\d+)`) + +// pprofMemStats reads runtime.MemStats from the heap profile's debug text, +// after the forced GC that gc=1 asks for. +func pprofMemStats(addr string) map[string]int64 { + out := map[string]int64{} + resp, err := http.Get("http://" + addr + "/debug/pprof/heap?gc=1&debug=1") + if err != nil { + return out + } + defer resp.Body.Close() + sc := bufio.NewScanner(resp.Body) + sc.Buffer(make([]byte, 1<<20), 1<<24) + for sc.Scan() { + if m := memStatLine.FindStringSubmatch(sc.Text()); m != nil { + v, _ := strconv.ParseInt(m[2], 10, 64) + out[m[1]] = v + } + } + return out +} + +func pprofGoroutines(addr string) int64 { + resp, err := http.Get("http://" + addr + "/debug/pprof/goroutine?debug=1") + if err != nil { + return -1 + } + defer resp.Body.Close() + line, _ := bufio.NewReader(resp.Body).ReadString('\n') + f := strings.Fields(line) // "goroutine profile: total N" + if len(f) == 0 { + return -1 + } + n, _ := strconv.ParseInt(f[len(f)-1], 10, 64) + return n +} + +func processRSSMB(pid int) float64 { + out, err := exec.Command("ps", "-o", "rss=", "-p", strconv.Itoa(pid)).Output() + if err != nil { + return -1 + } + kb, _ := strconv.ParseFloat(strings.TrimSpace(string(out)), 64) + return kb / 1024 +} + +func fetchToFile(url, path string) error { + resp, err := http.Get(url) + if err != nil { + return err + } + defer resp.Body.Close() + f, err := os.Create(path) + if err != nil { + return err + } + defer f.Close() + _, err = io.Copy(f, resp.Body) + return err +} + +func copyFile(src, dst string) error { + b, err := os.ReadFile(src) + if err != nil { + return err + } + return os.WriteFile(dst, b, 0o644) +} From 529679aa58aac50051e6b24f41effba2e9425616 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 15:04:46 +0300 Subject: [PATCH 09/10] daemon: at most one direct-upgrade attempt per relayed peer at a time relayProbeLoop started tryDirectUpgrade for every relayed peer every RelayProbeInterval (15 s). An attempt without a cached resolve asks the registry through withRegistryDeadline, which stops waiting after 8 s but leaves the call running: the registry client then waits for a free pooled connection with no deadline. While the registry was slow, every tick added one more waiting call per relayed peer, so goroutines grew with the length of the slowdown rather than with the number of peers. In the local soak test on main this reached 21,149 goroutines at one sample (the gave-up path resets of the previous commits feed the same registry); afterwards 30 MB of the agent's live heap was runtime.malg, i.e. about 70,000 goroutine descriptors that Go keeps for the life of the process. On the fleet the registry connections do stall: since the 10:47 restart catfact-ninja logged 79 "registry pool conn reconnect failed" and 9 half-open reconnects. - relayProbeTick (the loop's body, now testable) skips a peer whose previous attempt is still running (upgradeInFlight). - tryDirectUpgrade's resolve is waited for in full instead of through withRegistryDeadline, so the attempt, and the guard, last as long as the registry call. It already runs in its own goroutine. TestRelayProbeTickRunsOneUpgradePerPeer: with the registry stuck, 8 ticks over 25 relayed peers start 25 attempts (178 without the guard), and the next tick after they finish starts 25 again. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 9 ++++ pkg/daemon/daemon.go | 62 +++++++++++++++++------- pkg/daemon/zz_peer_churn_bounded_test.go | 52 ++++++++++++++++++++ 3 files changed, 106 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dfcd7006..19c144ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -382,6 +382,15 @@ Detailed per-release notes are on the 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 diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 09786083..437b918a 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -579,6 +579,10 @@ type Daemon struct { netPolicyMu sync.RWMutex netPolicies map[uint16][]uint16 + // upgradeInFlight holds the peers with a direct-upgrade attempt + // running (nodeID → struct{}); see relayProbeTick. + upgradeInFlight sync.Map + // orphanSeen records when reapOrphanedPeerState first found per-peer // state, with no contact ever recorded, for a node with no tunnel // entry. Only the idle sweep (reapStalePeers) touches it. @@ -6734,21 +6738,43 @@ func (d *Daemon) relayProbeLoop() { case <-d.stopCh: return case <-ticker.C: - idle := d.tunnels.idleFilter(time.Now()) - for _, nodeID := range d.tunnels.RelayPeerIDs() { - if idle(nodeID) { - // An upgrade costs a registry lookup, a beacon punch - // and five probes; nobody is using this path. - continue - } - if !d.tunnels.HasPeer(nodeID) { - // A relay flag with no tunnel behind it (see - // reapOrphanedPeerState): no session to upgrade. - continue - } - go d.tryDirectUpgrade(nodeID) - } + d.relayProbeTick(time.Now()) + } + } +} + +// directUpgradePeer runs one direct-upgrade attempt; tests swap it. +var directUpgradePeer = func(d *Daemon, nodeID uint32) { d.tryDirectUpgrade(nodeID) } + +// relayProbeTick starts a direct-upgrade attempt for each relayed peer in +// use, unless one for that peer is still running. An attempt that has to +// resolve the peer waits for the registry, and with the registry slow +// (or a pooled connection half-open) a new attempt every +// RelayProbeInterval stacked up behind the old ones: goroutines grew with +// how long the registry stayed slow, not with the number of peers, and Go +// keeps a goroutine's descriptor for the life of the process, so the heap +// stayed up after the backlog drained. With the guard, at most one attempt +// per peer waits. +func (d *Daemon) relayProbeTick(now time.Time) { + idle := d.tunnels.idleFilter(now) + for _, nodeID := range d.tunnels.RelayPeerIDs() { + if idle(nodeID) { + // An upgrade costs a registry lookup, a beacon punch + // and five probes; nobody is using this path. + continue + } + if !d.tunnels.HasPeer(nodeID) { + // A relay flag with no tunnel behind it (see + // reapOrphanedPeerState): no session to upgrade. + continue } + if _, running := d.upgradeInFlight.LoadOrStore(nodeID, struct{}{}); running { + continue + } + go func(nodeID uint32) { + defer d.upgradeInFlight.Delete(nodeID) + directUpgradePeer(d, nodeID) + }(nodeID) } } @@ -6778,9 +6804,11 @@ func (d *Daemon) tryDirectUpgrade(nodeID uint32) { if rc == nil { return } - r, err := withRegistryDeadline(registryCallDeadline, func() (map[string]interface{}, error) { - return rc.Resolve(nodeID, d.NodeID()) - }) + // Waited for in full, not cut off at registryCallDeadline: this + // runs in its own goroutine, and relayProbeTick starts no other + // attempt for the peer until this one returns. Cutting it off left + // the call running anyway, and the next tick added another. + r, err := rc.Resolve(nodeID, d.NodeID()) if err != nil { return } diff --git a/pkg/daemon/zz_peer_churn_bounded_test.go b/pkg/daemon/zz_peer_churn_bounded_test.go index fc24cd6a..04fc89c6 100644 --- a/pkg/daemon/zz_peer_churn_bounded_test.go +++ b/pkg/daemon/zz_peer_churn_bounded_test.go @@ -7,6 +7,7 @@ import ( "crypto/rand" "net" "runtime" + "sync/atomic" "testing" "time" @@ -287,3 +288,54 @@ func TestReaperForgetsOrphanedPeerState(t *testing.T) { t.Fatalf("orphanSeen still holds %d entries", len(d.orphanSeen)) } } + +// While a peer's direct-upgrade attempt is still running (its registry +// resolve is waiting on a slow registry), later ticks start no other for +// it: attempts are bounded by relayed peers, not by how long the registry +// stays slow. Before, every tick added one per relayed peer. +func TestRelayProbeTickRunsOneUpgradePerPeer(t *testing.T) { + release := make(chan struct{}) + var running, started atomic.Int32 + prev := directUpgradePeer + directUpgradePeer = func(_ *Daemon, _ uint32) { + started.Add(1) + running.Add(1) + <-release // a registry that does not answer + running.Add(-1) + } + t.Cleanup(func() { directUpgradePeer = prev }) + + d := churnDaemon(t) + const peers = 25 + addr := &net.UDPAddr{IP: net.IPv4(203, 0, 113, 10), Port: 4000} + for i := uint32(0); i < peers; i++ { + id := 0x00600000 + i + d.tunnels.AddPeer(id, addr) + d.tunnels.SetRelayPeer(id, true) + d.tunnels.noteAppActivity(id) // in use + } + for tick := 0; tick < 8; tick++ { // two minutes of RelayProbeInterval ticks + d.relayProbeTick(time.Now()) + } + deadline := time.Now().Add(2 * time.Second) + for running.Load() < peers && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if got := started.Load(); got != peers { + t.Fatalf("%d upgrade attempts after 8 ticks with the registry stuck; want one per relayed peer (%d)", got, peers) + } + close(release) + deadline = time.Now().Add(2 * time.Second) + for running.Load() > 0 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + // Once they finish, the next tick tries again. + d.relayProbeTick(time.Now()) + deadline = time.Now().Add(2 * time.Second) + for started.Load() < 2*peers && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if got := started.Load(); got != 2*peers { + t.Fatalf("after the attempts finished, a tick started %d more; want %d", got-peers, peers) + } +} From efe80b061670f3b142e3a471317466772e03b9da Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 15:05:34 +0300 Subject: [PATCH 10/10] daemon: peerInUse comment: a path reset keeps the activity record Co-Authored-By: Claude Opus 5.5 --- pkg/daemon/peeractivity.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/daemon/peeractivity.go b/pkg/daemon/peeractivity.go index b06cd984..708cbb11 100644 --- a/pkg/daemon/peeractivity.go +++ b/pkg/daemon/peeractivity.go @@ -112,8 +112,8 @@ func (tm *TunnelManager) idleFilter(now time.Time) func(nodeID uint32) bool { // application traffic within PeerIdleAfter. Unlike idleFilter, a peer with // no recorded activity is not in use: it is one this node only ever // exchanged upkeep with (a key request it never answered, a frame it could -// not decrypt), or one a path reset has just re-keyed. With PeerIdleAfter -// off every peer is in use, as before idleness was tracked. +// not decrypt). With PeerIdleAfter off every peer is in use, as before +// idleness was tracked. func (tm *TunnelManager) peerInUse(nodeID uint32, now time.Time) bool { if PeerIdleAfter <= 0 || tm.activity == nil { return true