diff --git a/CHANGELOG.md b/CHANGELOG.md index 8468314a..19c144ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -341,6 +341,56 @@ Detailed per-release notes are on the A `--data` over 65535 bytes is now refused before the daemon is contacted. +- **Peers nobody is using no longer cost anything.** The daemon kept every + peer it had ever exchanged keys with warm forever: a NAT keepalive every + 25 s, path-watchdog probes and resets when the peer went quiet, and for + each relayed peer a direct-upgrade attempt (registry lookup, beacon punch, + five probes) every 15 s. A public service agent that had answered a few + thousand clients sent about 1,600 packets a second on this while serving + a few requests a minute. A peer with no application traffic for 2 minutes + and no open connection now gets none of it. Sending to it works as + before, re-establishing the path through the usual fallbacks if a NAT + mapping expired meanwhile. Five minutes after the last frame either way, + the stale-peer reaper now drops the peer, as it was meant to: the + keepalives counted as contact, so it never fired, and busy nodes held + every peer they had ever seen (service agents: 3,500 to 10,900). The next + contact runs a fresh key exchange. The reaper also counts relayed inbound + frames as contact now, so a peer still sending to us is kept. +- **A reaped peer leaves nothing behind, and frames for a peer that never + answers no longer wait forever.** Frames queued for a key exchange were + dropped only when the exchange completed; for a peer that never answered + they stayed until the daemon exited, reaper or not. Once 256 such peers + had piled up, every first send to a new peer failed with "too many + pending key exchanges". A queue now goes after 2 minutes (the exchange + itself gives up after 20 s), and with its peer when the reaper drops it. +- **A rekey that gives up no longer resets a peer nobody is using.** The + reset starts a new key exchange; for a client that had gone away it gave + up again, and the next reset followed, for as long as the daemon ran. + With many such peers waiting, the cap of 64 key-exchange retransmits per + 4 s stretched each round past the 30 s reset cooldown, so nothing broke + the cycle: on 2026-10-08 one service agent logged 25,147 of these resets + in 90 minutes, each a registry lookup and a new key exchange, and the + key-exchange frames counted as contact, so the reaper never dropped those + peers. The reset now runs only for a peer with an open connection or + application traffic in the last 2 minutes. A path reset also keeps the + record of when the peer was last used. +- **State left behind for a node with no tunnel is reaped too.** A path + reset that could not re-resolve its peer left the peer's relay flag (and + whatever else it had) with no tunnel entry, which the stale-peer reaper, + walking tunnel peers, never visited: the flag kept the node in the 15 s + direct-upgrade loop and counted against the 4,096 relay peers a daemon + admits, for as long as the daemon ran. Such state now goes 5 minutes + after the last frame with the node, and the upgrade loop skips nodes with + no tunnel. +- **A slow registry no longer piles up goroutines.** Every 15 s the daemon + tries to move each relayed peer to a direct path, and an attempt that + must resolve the peer waits for the registry. It gave up waiting after + 8 s but left the call running, and the next round started another, so + while the registry was slow the waiting calls grew by one per relayed + peer every 15 s. A local soak test reached 21,000 goroutines, and since + Go keeps a goroutine's descriptor for the life of the process, the heap + stayed ~30 MB higher afterwards. A peer now has at most one attempt + running. - **The tunnel socket asks the kernel for 4 MB buffers** in each direction instead of the default (about 200 KB on Linux). Every tunnel shares the one socket, and several streams sending at once overflowed it. The kernel caps 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) + } +} diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index d0829569..437b918a 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -579,6 +579,15 @@ 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. + orphanSeen map[uint32]time.Time + // gaveUpResetMu guards lastGaveUpReset, the per-peer cooldown for // rekey-gave-up-triggered path resets (onRekeyGaveUp). gaveUpResetMu sync.Mutex @@ -701,6 +710,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 @@ -6450,14 +6460,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() @@ -6467,14 +6493,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. - latest := lastDirect - if ok && lastOutbound.After(latest) { - latest = lastOutbound - } + // 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. @@ -6485,9 +6507,70 @@ 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) } } + 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 +// 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 @@ -6556,6 +6639,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. @@ -6651,13 +6738,46 @@ func (d *Daemon) relayProbeLoop() { case <-d.stopCh: return case <-ticker.C: - for _, nodeID := range d.tunnels.RelayPeerIDs() { - 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) + } +} + // tryDirectUpgrade attempts to promote a relay-flagged peer to a direct path // via coordinated NAT hole-punching, then verifies on the rig that the path // actually carries traffic. @@ -6684,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/keyexchange/keyexchange.go b/pkg/daemon/keyexchange/keyexchange.go index 345fa0f8..a55d4181 100644 --- a/pkg/daemon/keyexchange/keyexchange.go +++ b/pkg/daemon/keyexchange/keyexchange.go @@ -738,6 +738,54 @@ 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 +} + +// 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/pathwatch.go b/pkg/daemon/pathwatch.go index 8aabe524..44a568a1 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 { @@ -250,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) } @@ -293,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 new file mode 100644 index 00000000..708cbb11 --- /dev/null +++ b/pkg/daemon/peeractivity.go @@ -0,0 +1,131 @@ +// 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 + } +} + +// 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). 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/routing/relay.go b/pkg/daemon/routing/relay.go index 25cd15f7..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 @@ -168,6 +169,44 @@ 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) +} + +// 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/tunnel.go b/pkg/daemon/tunnel.go index 01375335..3fae6b22 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: @@ -132,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; @@ -406,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), @@ -413,6 +421,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) @@ -516,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 @@ -988,6 +1070,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 +1078,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 +1671,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 +1958,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}: @@ -2014,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 { @@ -2192,6 +2288,16 @@ 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 — 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) @@ -2245,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]) @@ -2329,6 +2438,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..e77d9c79 --- /dev/null +++ b/pkg/daemon/zz_idle_peer_upkeep_test.go @@ -0,0 +1,183 @@ +// 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") + } + + // 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, + 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") + } +} 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_peer_churn_bounded_test.go b/pkg/daemon/zz_peer_churn_bounded_test.go new file mode 100644 index 00000000..04fc89c6 --- /dev/null +++ b/pkg/daemon/zz_peer_churn_bounded_test.go @@ -0,0 +1,341 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "crypto/ecdh" + "crypto/rand" + "net" + "runtime" + "sync/atomic" + "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") + } +} + +// 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)) + } +} + +// 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) + } +} 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) + } +} 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 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) +}