daemon: stop maintaining paths to peers nobody is using

CHANGELOG.md

@@ -245,6 +245,21 @@ Detailed per-release notes are on the
   unknown type; before it answered that no verifier was configured.
 
 ### Changed
+- **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

pkg/daemon/daemon.go

@@ -667,6 +667,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
@@ -5922,11 +5923,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.
@@ -6100,7 +6107,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)
 			}
 		}

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 {

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)
+}

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:
@@ -404,6 +408,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)
@@ -974,13 +979,17 @@ 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 {
 		last, ok := tm.routing.LastOutboundSend(nodeID)
 		if ok && now.Sub(last) < TunnelKeepaliveInterval {
 			continue
 		}
+		if idle(nodeID) {
+			continue
+		}
 		pc := tm.envelope.Get(nodeID)
 		if pc == nil || !pc.Ready {
 			continue
@@ -1478,6 +1487,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)
 	}
@@ -1759,6 +1774,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}:
@@ -2079,6 +2097,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)
@@ -2216,6 +2237,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