From 699cbdc2b3938b1b23d1402a01b80359ecf2396f Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Sat, 18 Jul 2026 11:27:42 +0300 Subject: [PATCH] =?UTF-8?q?daemon:=20per-peer=20path=20watchdog=20?= =?UTF-8?q?=E2=80=94=20heal=20a=20single=20dead=20peer=20path=20in=20place?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 2026-07-17 incident: one peer's NAT/relay session died (list-agents unreachable for hours) while 8 other peers stayed healthy — so the global rx-watchdog never fired (any live peer keeps PktsRecv moving), and the only remedy was a full daemon restart, which drops EVERY session and costs minutes of re-establishment on a NAT'd link. Nothing health-checks an individual peer mapping: the tunnel keepalive is fire-and-forget (the receiver silently swallows it), so it keeps NAT mappings warm but verifies nothing. Add pathwatch.go (L4): every Ready peer already refreshes lastInboundDecrypt at least every ~30s via bidirectional keepalives. A peer silent past 55s gets up to 3 pong-soliciting probes; still silent → that ONE peer's path is reset in place via the same sequence as `pilotctl prefer-direct` (drop cached resolve + tunnel, preserve relay-active, clear rekey cooldowns, proactively push a fresh PILA) — extracted from the IPC handler into the shared resetPeerPath. Worst- case detection ≈85s, recovery ~1 RTT, healthy sessions untouched. Reset is rate-limited per peer (2min cooldown); disable with -no-path-watch. Events: tunnel.path_suspect / tunnel.path_recovered. Backward compatible with the whole deployed fleet — no wire changes: the probe is a plain PortPing control frame with a non-empty payload. Verified against the v1.10.0, v1.10.0-rc5 and -rc7 trees (the fleet runs rc6-dev): their isTunnelKeepalive only swallows EMPTY frames, so the probe reaches handleControlPacket and gets a pong; old peers' own keepalives independently keep refreshing our liveness signal, so an old peer is never falsely reset. The compat invariant is pinned by TestPathProbeIsNotSwallowedByOldKeepaliveFilter. Tests: probe/reset/recovery/cooldown/prune state machine, relay-state preservation in resetPeerPath, keepalive-filter compat. pkg/daemon green under -race for the new tests and under the -short CI gate. Co-Authored-By: Claude Opus 4.8 --- cmd/daemon/main.go | 2 + pkg/daemon/daemon.go | 16 ++ pkg/daemon/ipc.go | 107 +--------- pkg/daemon/keyexchange/keyexchange.go | 11 ++ pkg/daemon/pathwatch.go | 272 ++++++++++++++++++++++++++ pkg/daemon/tunnel.go | 68 +++++++ pkg/daemon/zz_pathwatch_test.go | 227 +++++++++++++++++++++ 7 files changed, 606 insertions(+), 97 deletions(-) create mode 100644 pkg/daemon/pathwatch.go create mode 100644 pkg/daemon/zz_pathwatch_test.go diff --git a/cmd/daemon/main.go b/cmd/daemon/main.go index 1c3ede04..0d84a85e 100644 --- a/cmd/daemon/main.go +++ b/cmd/daemon/main.go @@ -95,6 +95,7 @@ func main() { trustAutoApprove := flag.Bool("trust-auto-approve", false, "automatically approve all incoming trust handshakes") beaconRTTProbe := flag.Bool("beacon-rtt-probe", false, "probe beacon RTT before selection; override hash pick when >2× slower than best (ablation test, default off)") noRxWatchdog := flag.Bool("no-rx-watchdog", false, "disable the inbound-path watchdog that soft-recovers (beacon+registry re-registration) and, on a persistent wedge, exits non-zero for supervisor respawn") + noPathWatch := flag.Bool("no-path-watch", false, "disable the per-peer path watchdog that probes inbound-silent peers and resets a dead peer path in place (prefer-direct sequence) without a daemon restart") transportMode := flag.String("transport", "udp", "tunnel transport: 'udp' (default) or 'compat' (WSS to beacon, opt-in, for UDP-blocked environments)") compatBeacon := flag.String("compat-beacon", "wss://beacon.pilotprotocol.network/v1/compat", "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) or 'pinned' (Pilot CA root embedded in the daemon binary; will become the default in a future release once production root ships)") @@ -250,6 +251,7 @@ func main() { TrustAutoApprove: *trustAutoApprove, BeaconRTTProbe: *beaconRTTProbe, DisableRxWatchdog: *noRxWatchdog, + DisablePathWatch: *noPathWatch, TransportMode: *transportMode, CompatBeaconURL: *compatBeacon, CompatTLSTrust: *tlsTrust, diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 31f6f00a..0e5a9c2f 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -147,6 +147,16 @@ type Config struct { // (watchdog on) — the wedge otherwise requires a manual restart. DisableRxWatchdog bool + // DisablePathWatch turns off the per-peer path watchdog + // (pathwatch.go). The watchdog detects a SINGLE peer's NAT/relay + // mapping dying while other peers stay healthy (so the global + // rx-watchdog never fires), probes the peer with pong-soliciting + // pings that every deployed daemon version answers, and on + // sustained silence resets that one peer's path in place (same + // sequence as `pilotctl prefer-direct`) instead of requiring a + // full daemon restart. Default false (watchdog on). + DisablePathWatch bool + // Telemetry consent gate. When set to the telemetry endpoint URL, // the daemon initialises a telemetry client that emits signed events // (install, usage, view, review). When empty (default), the client @@ -1228,6 +1238,12 @@ func (d *Daemon) Start() error { d.bgWG.Add(1) go func() { defer d.bgWG.Done(); d.rxWatchdogLoop() }() + // 8c. Start per-peer path watchdog (L4). Detects a single peer's + // NAT/relay mapping dying while others stay healthy and resets that + // peer's path in place — see pathwatch.go. + d.bgWG.Add(1) + go func() { defer d.bgWG.Done(); d.pathWatchLoop() }() + // 9. Start idle connection sweeper d.bgWG.Add(1) go func() { defer d.bgWG.Done(); d.idleSweepLoop() }() diff --git a/pkg/daemon/ipc.go b/pkg/daemon/ipc.go index f486d8f9..6e40cedf 100644 --- a/pkg/daemon/ipc.go +++ b/pkg/daemon/ipc.go @@ -1225,106 +1225,19 @@ func (s *IPCServer) handlePreferDirect(conn *ipcConn, reqID uint64, payload []by } nodeID := binary.BigEndian.Uint32(payload) - d := s.daemon - hadTunnel := d.tunnels.HasPeer(nodeID) - wasRelayActive := d.tunnels.IsRelayPeer(nodeID) - wasRelayPinned := d.tunnels.IsRelayPinned(nodeID) - - // Unpin relay so the next successful direct receive can promote the - // peer out via the routing layer's ClearRelayOnDirect path. We do NOT - // unset the active relay flag — that would cause our proactive PILA - // below to be sprayed at the peer's direct (NAT'd, unreachable) - // endpoint and lost. Keeping relay active means the PILA reaches - // the peer through the beacon, restores crypto state, and the - // routing layer can then promote the path on its own merits. - // - // Reproduces the Mac↔GCP-VM dual-NAT case where direct never works: - // unsetting relay there forced the next dial onto a dead direct - // path and stalled send-file at the dataexchange timeout (~17 s) - // before the routing layer's silence detector could re-flip. - if wasRelayPinned { - d.tunnels.SetRelayPeerPinned(nodeID, false) - } - - // Drop the cached resolve / endpoint so the next ensureTunnel hits - // the registry fresh. Without this, ensureTunnel would short-circuit - // on HasPeer=true and reuse the existing (relay-mediated) tunnel. - d.forgetPeerResolution(nodeID) - - // Drop the tunnel + per-peer state so ensureTunnel re-runs the full - // resolve + punch flow on the next dial. RemovePeer is safe under - // concurrent traffic — see TunnelManager.RemovePeer's docstring for the - // per-peer metadata it cleans up. - // - // SIDE EFFECT we have to undo: RemovePeer calls routing.RemovePeer, - // which wipes relayPeers and relayPinned. For a peer where the only - // working transport is relay (the Mac↔GCP-VM dual-NAT case), this is - // catastrophic — the proactive PILA we push at the end of this handler - // would then be sprayed at the peer's direct address and lost. We - // captured the original relay state above; re-apply it after the wipe - // so writeFrame still picks the working path. - if hadTunnel { - d.tunnels.RemovePeer(nodeID) - } - if wasRelayActive { - d.tunnels.SetRelayPeer(nodeID, true) - // Note: we deliberately do NOT restore relayPinned. The point of - // prefer-direct is to give the next direct receive a chance to - // promote the path out of relay; restoring the pin would defeat - // that. Relay active without pin = "use relay for now, but a - // direct packet from this peer will demote the relay flag". - } - - // Clear the per-peer cooldowns that survive RemovePeer. - // - // rekeyMu.lastRekeyReq lives on TunnelManager and is NOT touched by - // tunnels.RemovePeer. Without this, the next "encrypted packet but no - // key" from the peer hits the 3-second gate (rekeyRequestInterval) - // and silently skips the PILA reply. That blank window is exactly - // when send-file is about to dial — losing it means another retransmit - // cycle before recovery. - d.tunnels.ClearLastRekeyReq(nodeID) - // rekeyGaveUp similarly survives RemovePeer (kx.RemovePeer doesn't - // clear it because giving up doesn't imply the peer is gone). After - // an operator-initiated reset, treat the peer as fresh — let - // MarkPendingRekey arm a new retransmit cycle without the 5-second - // cooldown delay. - d.tunnels.ClearRekeyGaveUp(nodeID) - - // Proactively push a fresh PILA. Without this, recovery waits for the - // peer to send us their next packet, which can take up to the peer's - // own keepalive cadence (currently ~30s in the worst case). Doing it - // from here makes prefer-direct recovery deterministic: by the time - // the IPC reply lands, our PILA has already left the wire. - // - // Sequence: - // 1. ensureTunnel re-resolves the peer's address from the registry - // (RemovePeer just wiped tm.peers, so sendKeyExchangeToNode would - // otherwise have no destination and silently no-op). - // 2. sendKeyExchangeToNode builds the signed PILA and writes it - // out via tm.writeFrame, picking direct vs relay based on the - // freshly-set routing state. - // - // Both steps are best-effort. If the registry is unreachable or the - // peer has no public endpoint registered, we still report tunnel - // state was reset and the caller can fall back to relay-pinned - // behaviour on retry. - pilaPushed := false - resolveErr := "" - if err := d.ensureTunnel(nodeID); err != nil { - resolveErr = err.Error() - } else { - d.tunnels.sendKeyExchangeToNode(nodeID) - pilaPushed = true - } + // The reset + re-establish sequence is shared with the per-peer path + // watchdog (pathwatch.go) — see resetPeerPath for the full rationale + // of each step (relay-state preservation, cooldown clears, proactive + // PILA push). + res := s.daemon.resetPeerPath(nodeID) data, err := json.Marshal(map[string]interface{}{ "node_id": nodeID, - "had_tunnel": hadTunnel, - "was_relay_active": wasRelayActive, - "was_relay_pinned": wasRelayPinned, - "pila_pushed": pilaPushed, - "resolve_error": resolveErr, + "had_tunnel": res.HadTunnel, + "was_relay_active": res.WasRelayActive, + "was_relay_pinned": res.WasRelayPinned, + "pila_pushed": res.PilaPushed, + "resolve_error": res.ResolveErr, }) if err != nil { s.sendError(conn, reqID, fmt.Sprintf("prefer_direct marshal: %v", err)) diff --git a/pkg/daemon/keyexchange/keyexchange.go b/pkg/daemon/keyexchange/keyexchange.go index 6cd473f8..87263757 100644 --- a/pkg/daemon/keyexchange/keyexchange.go +++ b/pkg/daemon/keyexchange/keyexchange.go @@ -563,6 +563,17 @@ func (m *Manager) LastInboundDecryptHas(peerNodeID uint32) bool { return ok } +// LastInboundDecrypt returns the timestamp of the last successful +// decrypt from peerNodeID. ok=false means no decrypt has ever been +// recorded this process. Used by the per-peer path watchdog to measure +// how long a peer's inbound path has been silent. +func (m *Manager) LastInboundDecrypt(peerNodeID uint32) (time.Time, bool) { + m.rkPendingMu.Lock() + defer m.rkPendingMu.Unlock() + t, ok := m.lastInboundDecrypt[peerNodeID] + return t, ok +} + // InboundDecryptStale returns true if we haven't successfully // decrypted any packet from peerNodeID within the staleness window. func (m *Manager) InboundDecryptStale(peerNodeID uint32) bool { diff --git a/pkg/daemon/pathwatch.go b/pkg/daemon/pathwatch.go new file mode 100644 index 00000000..53ef2a99 --- /dev/null +++ b/pkg/daemon/pathwatch.go @@ -0,0 +1,272 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "log/slog" + "math/rand" + "time" +) + +// Per-peer path watchdog (L4). +// +// Failure mode this exists for (2026-07-17 incident): a peer session's +// NAT/relay mapping dies for ONE peer while the rest stay healthy. The +// global rx-watchdog (rxwatchdog.go) never fires — any single live peer +// keeps PktsRecv moving — so the dead peer stays dark until an operator +// notices and restarts something. Observed shape: 9 peers healthy, +// list-agents unreachable for hours; a restart "fixed" it only by +// rebuilding EVERY session (and cost minutes of total re-establishment). +// +// Detection: every Ready peer exchanges encrypted keepalives both ways +// at TunnelKeepaliveInterval (25s), so a live path refreshes the peer's +// lastInboundDecrypt at least every ~30s regardless of the peer's +// version. If a peer goes inbound-silent past pathSilenceThreshold, the +// watchdog sends up to pathProbeMax pong-soliciting probes (see +// TunnelManager.SendPathProbe — plain PortPing control frames that every +// deployed version answers; no new wire format). Any authenticated +// inbound — pong or otherwise — clears the suspicion. +// +// Recovery: still silent after the probe budget → the peer's path is +// declared dead and resetPeerPath runs the same in-place re-establish +// sequence as `pilotctl prefer-direct`: drop cached resolve/endpoint, +// drop the tunnel, clear rekey cooldowns, re-resolve, and proactively +// push a fresh PILA. Per-peer, in-place, seconds — instead of a full +// daemon restart that drops every healthy session too. +// +// Backward compatibility (v1.10.0 fleet): old peers send keepalives on +// the same cadence (their keepalives bump our lastInboundDecrypt), and +// their handleControlPacket answers non-empty PortPing frames with a +// pong (verified against the v1.10.0 tree: isTunnelKeepalive only +// swallows EMPTY payloads). A peer that neither sends nor pongs within +// the full window is genuinely unreachable — resetting its path state +// is correct on every version, and rate-limited by pathResetCooldown. +const ( + // pathWatchTickInterval is how often the watchdog scans peers. + pathWatchTickInterval = 10 * time.Second + + // pathSilenceThreshold is how long a Ready peer must be + // inbound-silent before probing starts. Above 2× the 25s keepalive + // interval so a single lost keepalive never triggers probing. + pathSilenceThreshold = 55 * time.Second + + // pathProbeMax is how many pong-soliciting probes are sent (one per + // tick) before the path is declared dead. With a 10s tick this puts + // worst-case detection at ~threshold + 3 ticks ≈ 85s. + pathProbeMax = 3 + + // pathResetCooldown rate-limits resets per peer. A peer that is + // genuinely offline (not a path problem) would otherwise be reset + // on every probe-budget cycle; the cooldown caps that churn while + // still retrying often enough to catch the peer coming back. + pathResetCooldown = 2 * time.Minute +) + +// pathPeerState is the watchdog's per-peer memory between ticks. +type pathPeerState struct { + probesSent int + lastResetAt time.Time +} + +// pathWatchAction names what a tick decided for one peer — returned +// from pathWatchPeer for tests and logs. +type pathWatchAction string + +const ( + pathActionHealthy pathWatchAction = "healthy" + pathActionProbe pathWatchAction = "probe" + pathActionReset pathWatchAction = "reset" + pathActionCooldown pathWatchAction = "cooldown" + pathActionSkip pathWatchAction = "skip" +) + +// pathWatchResetPeer is swapped by tests to observe resets without a +// live registry/beacon (mirrors the rxWatchdogExit pattern). +var pathWatchResetPeer = func(d *Daemon, nodeID uint32) { + res := d.resetPeerPath(nodeID) + slog.Warn("path watchdog: peer path reset", + "peer_node_id", nodeID, + "had_tunnel", res.HadTunnel, + "was_relay_active", res.WasRelayActive, + "pila_pushed", res.PilaPushed, + "resolve_error", res.ResolveErr) +} + +func (d *Daemon) pathWatchLoop() { + if d.config.DisablePathWatch { + return + } + // Independent jitter so this loop does not align with the others. + // #nosec G404 -- startup-jitter scheduling only (same pattern as the + // sibling loops); not security-sensitive randomness + time.Sleep(time.Duration(rand.Int63n(int64(5 * time.Second)))) + + states := make(map[uint32]*pathPeerState) + ticker := time.NewTicker(pathWatchTickInterval) + defer ticker.Stop() + for { + select { + case <-d.stopCh: + return + case <-ticker.C: + d.pathWatchTick(states, time.Now()) + } + } +} + +// pathWatchTick runs one scan over all Ready peers. Extracted for +// testability — production drives it from pathWatchLoop. +// +// L4 panic boundary (architecture-notes/03-INVARIANTS.md §8): the tick +// calls into resolve/rekey machinery; a panic drops this tick and the +// next one rescans from clean state. +func (d *Daemon) pathWatchTick(states map[uint32]*pathPeerState, now time.Time) { + defer recoverLayer("L4", "pathWatchTick", d.bus, nil) + + ready := d.tunnels.ReadyPeerIDs() + live := make(map[uint32]bool, len(ready)) + for _, nodeID := range ready { + live[nodeID] = true + st := states[nodeID] + if st == nil { + st = &pathPeerState{} + states[nodeID] = st + } + d.pathWatchPeer(nodeID, st, now) + } + // Drop state for peers that no longer have a session so the map + // doesn't grow unboundedly. + for nodeID := range states { + if !live[nodeID] { + delete(states, nodeID) + } + } +} + +// pathWatchPeer evaluates one peer: healthy / probe / reset / cooldown. +func (d *Daemon) pathWatchPeer(nodeID uint32, st *pathPeerState, now time.Time) pathWatchAction { + last, ok := d.tunnels.LastInboundDecrypt(nodeID) + if !ok { + // Session marked Ready but no decrypt recorded yet — key + // exchange just completed. Leave it to the rekey machinery. + return pathActionSkip + } + silence := now.Sub(last) + if silence < pathSilenceThreshold { + if st.probesSent > 0 { + slog.Info("path watchdog: peer path recovered", + "peer_node_id", nodeID, + "silent_for", silence.Truncate(time.Second).String(), + "probes_sent", st.probesSent) + d.publishEvent("tunnel.path_recovered", map[string]any{ + "peer_node_id": nodeID, + "silent_for_seconds": int64(silence.Seconds()), + "probes_sent": st.probesSent, + }) + } + st.probesSent = 0 + return pathActionHealthy + } + + // Inside the post-reset cooldown: don't re-probe/re-reset a peer + // that is likely just offline. + if !st.lastResetAt.IsZero() && now.Sub(st.lastResetAt) < pathResetCooldown { + return pathActionCooldown + } + + if st.probesSent < pathProbeMax { + st.probesSent++ + if err := d.tunnels.SendPathProbe(nodeID); err != nil { + slog.Debug("path watchdog: probe send failed", + "peer_node_id", nodeID, "attempt", st.probesSent, "err", err) + } + return pathActionProbe + } + + // Probe budget exhausted with no authenticated inbound: the path is + // dead. Reset it in place. + slog.Warn("path watchdog: peer inbound-silent past probe budget — resetting path", + "peer_node_id", nodeID, + "silent_for", silence.Truncate(time.Second).String(), + "probes_sent", st.probesSent) + d.publishEvent("tunnel.path_suspect", map[string]any{ + "peer_node_id": nodeID, + "silent_for_seconds": int64(silence.Seconds()), + "probes_sent": st.probesSent, + }) + pathWatchResetPeer(d, nodeID) + st.probesSent = 0 + st.lastResetAt = now + return pathActionReset +} + +// peerPathReset reports what resetPeerPath found and did. +type peerPathReset struct { + HadTunnel bool + WasRelayActive bool + WasRelayPinned bool + PilaPushed bool + ResolveErr string +} + +// resetPeerPath drops a peer's tunnel + sticky routing state and +// proactively re-establishes, so the next traffic takes a freshly +// resolved + punched path. Shared by the prefer-direct IPC handler +// (operator-initiated) and the path watchdog (automatic). +// +// Step-by-step rationale (kept from the original prefer-direct +// implementation — see the Mac↔GCP-VM dual-NAT postmortem): +// +// - Unpin relay (but keep it ACTIVE) so the proactive PILA below still +// travels via the beacon relay when that is the only working path, +// while a future direct receive can promote the path. +// - forgetPeerResolution drops the cached resolve/endpoint so +// ensureTunnel hits the registry fresh instead of short-circuiting. +// - RemovePeer wipes the tunnel + per-peer metadata; it also wipes the +// relay flags as a side effect, so the captured relay-active state is +// re-applied (without the pin) right after. +// - ClearLastRekeyReq / ClearRekeyGaveUp lift the per-peer rekey +// cooldowns that survive RemovePeer — without this the recovery PILA +// would be silently skipped by the 3s rate gate. +// - ensureTunnel + sendKeyExchangeToNode push a fresh signed PILA +// immediately, making recovery deterministic instead of waiting for +// the peer's next keepalive. Best-effort: with the registry +// unreachable the state reset still happened and the next inbound +// packet from the peer triggers the rekey path. +func (d *Daemon) resetPeerPath(nodeID uint32) peerPathReset { + res := peerPathReset{ + HadTunnel: d.tunnels.HasPeer(nodeID), + WasRelayActive: d.tunnels.IsRelayPeer(nodeID), + WasRelayPinned: d.tunnels.IsRelayPinned(nodeID), + } + + if res.WasRelayPinned { + d.tunnels.SetRelayPeerPinned(nodeID, false) + } + + d.forgetPeerResolution(nodeID) + + if res.HadTunnel { + d.tunnels.RemovePeer(nodeID) + } + if res.WasRelayActive { + d.tunnels.SetRelayPeer(nodeID, true) + } + + d.tunnels.ClearLastRekeyReq(nodeID) + d.tunnels.ClearRekeyGaveUp(nodeID) + + // Guarded for watchdog-driven callers in minimal configurations + // (tests, registry-less setups): the IPC path always has a regConn. + if d.regConn == nil { + res.ResolveErr = "no registry connection" + return res + } + if err := d.ensureTunnel(nodeID); err != nil { + res.ResolveErr = err.Error() + } else { + d.tunnels.sendKeyExchangeToNode(nodeID) + res.PilaPushed = true + } + return res +} diff --git a/pkg/daemon/tunnel.go b/pkg/daemon/tunnel.go index 4b98f6ec..30a721f2 100644 --- a/pkg/daemon/tunnel.go +++ b/pkg/daemon/tunnel.go @@ -886,6 +886,74 @@ func isTunnelKeepalive(pkt *protocol.Packet) bool { len(pkt.Payload) == 0 } +// ReadyPeerIDs returns a snapshot of peers that have an established +// crypto session (pc.Ready). These are the peers the path watchdog +// health-checks: a Ready peer exchanges keepalives both ways every +// TunnelKeepaliveInterval, so sustained inbound silence from one is a +// per-peer path-death signal, not idleness. +func (tm *TunnelManager) ReadyPeerIDs() []uint32 { + tm.mu.RLock() + defer tm.mu.RUnlock() + ids := make([]uint32, 0, len(tm.peers)) + for nodeID := range tm.peers { + if pc := tm.envelope.Get(nodeID); pc != nil && pc.Ready { + ids = append(ids, nodeID) + } + } + return ids +} + +// LastInboundDecrypt is the tunnel-level shim over +// keyexchange.Manager.LastInboundDecrypt for the path watchdog. +func (tm *TunnelManager) LastInboundDecrypt(peerNodeID uint32) (time.Time, bool) { + return tm.kx.LastInboundDecrypt(peerNodeID) +} + +// pathProbePayload marks a path-watch probe. MUST be non-empty: +// isTunnelKeepalive on every deployed version (v1.10.0+) only swallows +// EMPTY ProtoControl/PortPing frames, so a payload-carrying probe passes +// the peer's keepalive filter, reaches handleControlPacket, and gets a +// pong back — from old and new peers alike. The pong (or any other +// authenticated inbound frame) bumps lastInboundDecrypt, which is the +// liveness signal the path watchdog reads. No new wire format. +var pathProbePayload = []byte("pathprobe.v1") + +// SendPathProbe sends one pong-soliciting ping to a Ready peer over the +// existing encrypted session. Returns an error if the peer has no +// session or the write fails. The reply needs no dedicated handling: +// the pong is an encrypted frame, and decrypting it refreshes the +// peer's lastInboundDecrypt — proving the bidirectional path. +// +// SrcPort is PortPing so the peer's pong comes back with +// DstPort==PortPing + FlagACK, which handleControlPacket ignores by +// design (anti-amplification) — the application layer never sees it. +func (tm *TunnelManager) SendPathProbe(peerNodeID uint32) error { + tm.mu.RLock() + addr := tm.peers[peerNodeID] + tm.mu.RUnlock() + if addr == nil { + return fmt.Errorf("path probe: no address for peer %d", peerNodeID) + } + pc := tm.envelope.Get(peerNodeID) + if pc == nil || !pc.Ready { + return fmt.Errorf("path probe: no ready session for peer %d", peerNodeID) + } + probe := &protocol.Packet{ + Version: protocol.Version, + Protocol: protocol.ProtoControl, + SrcPort: protocol.PortPing, + DstPort: protocol.PortPing, + Src: protocol.Addr{Node: tm.loadNodeID()}, + Payload: pathProbePayload, + } + plaintext, err := probe.Marshal() + if err != nil { + return fmt.Errorf("path probe marshal: %w", err) + } + frame := tm.encryptFrame(pc, plaintext) + return tm.writeFrame(peerNodeID, addr, frame) +} + // getPeerPubKey returns the cached Ed25519 public key for a peer, // fetching from registry on cache miss. Thin shim over // keyexchange.Manager.GetPeerPubKey. diff --git a/pkg/daemon/zz_pathwatch_test.go b/pkg/daemon/zz_pathwatch_test.go new file mode 100644 index 00000000..38c89aa1 --- /dev/null +++ b/pkg/daemon/zz_pathwatch_test.go @@ -0,0 +1,227 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "net" + "testing" + "time" + + "github.com/pilot-protocol/common/protocol" +) + +// newPathWatchTestDaemon builds the minimal Daemon the path-watch tick +// needs: a TunnelManager with one Ready peer installed, no socket, no +// registry (resetPeerPath's network steps are guarded/best-effort). +func newPathWatchTestDaemon(t *testing.T, peerID uint32) *Daemon { + t.Helper() + d := &Daemon{ + tunnels: NewTunnelManager(), + startTime: time.Now(), + stopCh: make(chan struct{}), + } + addr := &net.UDPAddr{IP: net.IPv4(203, 0, 113, 7), Port: 4000} + d.tunnels.AddPeer(peerID, addr) + pc := fakePC(t) // real AEAD so SendPathProbe's encryptFrame works + pc.Authenticated = true + d.tunnels.envelope.Install(peerID, pc) + return d +} + +// swapPathResetForTest stubs the reset hook and returns a counter of +// invocations per peer. Tests using it must NOT run in parallel with +// each other (global hook), mirroring swapExitForTest. +func swapPathResetForTest(t *testing.T) *[]uint32 { + t.Helper() + var resets []uint32 + prev := pathWatchResetPeer + pathWatchResetPeer = func(d *Daemon, nodeID uint32) { resets = append(resets, nodeID) } + t.Cleanup(func() { pathWatchResetPeer = prev }) + return &resets +} + +func TestPathWatchHealthyPeerNoAction(t *testing.T) { + const peer = 42 + d := newPathWatchTestDaemon(t, peer) + resets := swapPathResetForTest(t) + now := time.Now() + + // Fresh inbound — like a live peer's keepalives (any daemon version + // sends them every 25s). + d.tunnels.kx.SetLastInboundDecryptForTest(peer, now.Add(-10*time.Second)) + + st := &pathPeerState{} + if got := d.pathWatchPeer(peer, st, now); got != pathActionHealthy { + t.Fatalf("action = %q, want %q", got, pathActionHealthy) + } + if st.probesSent != 0 || len(*resets) != 0 { + t.Fatalf("healthy peer must not accrue probes/resets (probes=%d resets=%d)", + st.probesSent, len(*resets)) + } +} + +func TestPathWatchStalePeerProbesThenResets(t *testing.T) { + const peer = 43 + d := newPathWatchTestDaemon(t, peer) + resets := swapPathResetForTest(t) + now := time.Now() + + // Inbound-silent well past the threshold, and staying silent. + d.tunnels.kx.SetLastInboundDecryptForTest(peer, now.Add(-2*pathSilenceThreshold)) + + st := &pathPeerState{} + for i := 1; i <= pathProbeMax; i++ { + if got := d.pathWatchPeer(peer, st, now); got != pathActionProbe { + t.Fatalf("tick %d action = %q, want %q", i, got, pathActionProbe) + } + if st.probesSent != i { + t.Fatalf("tick %d probesSent = %d, want %d", i, st.probesSent, i) + } + now = now.Add(pathWatchTickInterval) + } + + // Budget exhausted, still silent: reset fires exactly once. + if got := d.pathWatchPeer(peer, st, now); got != pathActionReset { + t.Fatalf("post-budget action = %q, want %q", got, pathActionReset) + } + if len(*resets) != 1 || (*resets)[0] != peer { + t.Fatalf("resets = %v, want exactly [%d]", *resets, peer) + } + if st.probesSent != 0 || st.lastResetAt.IsZero() { + t.Fatalf("reset must clear probes and stamp cooldown (probes=%d resetAt=%v)", + st.probesSent, st.lastResetAt) + } +} + +func TestPathWatchRecoveryClearsProbeState(t *testing.T) { + const peer = 44 + d := newPathWatchTestDaemon(t, peer) + resets := swapPathResetForTest(t) + now := time.Now() + + // Silent → one probe... + d.tunnels.kx.SetLastInboundDecryptForTest(peer, now.Add(-pathSilenceThreshold-time.Second)) + st := &pathPeerState{} + if got := d.pathWatchPeer(peer, st, now); got != pathActionProbe { + t.Fatalf("action = %q, want %q", got, pathActionProbe) + } + + // ...then the pong (or any authenticated inbound) lands. + d.tunnels.kx.SetLastInboundDecryptForTest(peer, now) + if got := d.pathWatchPeer(peer, st, now.Add(time.Second)); got != pathActionHealthy { + t.Fatalf("post-recovery action = %q, want %q", got, pathActionHealthy) + } + if st.probesSent != 0 || len(*resets) != 0 { + t.Fatalf("recovery must clear probe state without reset (probes=%d resets=%d)", + st.probesSent, len(*resets)) + } +} + +func TestPathWatchCooldownPreventsResetStorm(t *testing.T) { + const peer = 45 + d := newPathWatchTestDaemon(t, peer) + resets := swapPathResetForTest(t) + now := time.Now() + + // Peer is genuinely offline: permanently silent. + d.tunnels.kx.SetLastInboundDecryptForTest(peer, now.Add(-time.Hour)) + st := &pathPeerState{lastResetAt: now.Add(-pathResetCooldown / 2)} + + if got := d.pathWatchPeer(peer, st, now); got != pathActionCooldown { + t.Fatalf("in-cooldown action = %q, want %q", got, pathActionCooldown) + } + if len(*resets) != 0 { + t.Fatalf("cooldown must suppress resets, got %v", *resets) + } + + // After the cooldown expires, the probe cycle starts over (not an + // instant reset — the peer gets a fresh probe budget). + later := now.Add(pathResetCooldown) + if got := d.pathWatchPeer(peer, st, later); got != pathActionProbe { + t.Fatalf("post-cooldown action = %q, want %q", got, pathActionProbe) + } +} + +func TestPathWatchTickPrunesGonePeers(t *testing.T) { + const peer = 46 + d := newPathWatchTestDaemon(t, peer) + _ = swapPathResetForTest(t) + now := time.Now() + d.tunnels.kx.SetLastInboundDecryptForTest(peer, now) + + states := make(map[uint32]*pathPeerState) + d.pathWatchTick(states, now) + if states[peer] == nil { + t.Fatalf("tick must create state for ready peer %d", peer) + } + + // Peer session goes away → state pruned on the next tick. + d.tunnels.RemovePeer(peer) + d.pathWatchTick(states, now.Add(pathWatchTickInterval)) + if states[peer] != nil { + t.Fatalf("tick must prune state for removed peer %d", peer) + } +} + +// TestPathProbeIsNotSwallowedByOldKeepaliveFilter pins the backward- +// compatibility invariant this whole design rests on: the probe must +// NOT match isTunnelKeepalive — the filter every deployed version +// (v1.10.0+) uses to silently swallow keepalives before they reach the +// pong-answering control handler. isTunnelKeepalive only matches EMPTY +// ProtoControl/PortPing frames, so the probe's non-empty payload is +// what earns it a pong from old peers. If someone "optimises" the +// payload away, old peers stop answering and the watchdog would reset +// healthy old-peer paths — this test is the tripwire. +func TestPathProbeIsNotSwallowedByOldKeepaliveFilter(t *testing.T) { + t.Parallel() + probe := &protocol.Packet{ + Version: protocol.Version, + Protocol: protocol.ProtoControl, + SrcPort: protocol.PortPing, + DstPort: protocol.PortPing, + Payload: pathProbePayload, + } + if isTunnelKeepalive(probe) { + t.Fatal("path probe matches isTunnelKeepalive — old peers would swallow it instead of ponging") + } + if len(pathProbePayload) == 0 { + t.Fatal("pathProbePayload must be non-empty (see isTunnelKeepalive)") + } + // And the real keepalive still matches, so we haven't broken the + // filter from the other side. + tm := NewTunnelManager() + if !isTunnelKeepalive(tm.newKeepalivePacket()) { + t.Fatal("regular keepalive no longer matches isTunnelKeepalive") + } +} + +// TestResetPeerPathPreservesRelayActive pins the relay-state contract +// resetPeerPath inherited from prefer-direct: relay ACTIVE survives the +// reset (the recovery PILA must travel via the beacon when relay is the +// only working path), while the PIN is cleared (a direct receive may +// promote the path). +func TestResetPeerPathPreservesRelayActive(t *testing.T) { + const peer = 47 + d := newPathWatchTestDaemon(t, peer) + d.tunnels.SetRelayPeer(peer, true) + d.tunnels.SetRelayPeerPinned(peer, true) + + res := d.resetPeerPath(peer) + + if !res.HadTunnel || !res.WasRelayActive || !res.WasRelayPinned { + t.Fatalf("captured state wrong: %+v", res) + } + if !d.tunnels.IsRelayPeer(peer) { + t.Fatal("relay ACTIVE must be restored after reset") + } + if d.tunnels.IsRelayPinned(peer) { + t.Fatal("relay PIN must be cleared after reset") + } + if d.tunnels.HasPeer(peer) { + t.Fatal("tunnel must be dropped by reset") + } + // No registry in this fixture: the network step reports, not panics. + if res.ResolveErr == "" { + t.Fatal("expected resolve error with no registry connection") + } +}