daemon: per-peer path watchdog — heal a single dead peer path in place

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,

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

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

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 {