daemon: inbound-path watchdog — auto-recover the long-uptime NAT wedge

CHANGELOG.md

@@ -12,6 +12,21 @@ Detailed per-release notes are on the
 Reliable P2P data transfer across NAT. Tag intentionally held for review.
 
 ### Added
+- **Inbound-path watchdog — the long-uptime NAT wedge now auto-recovers.**
+  A daemon could run for days transmitting into a stale NAT/relay mapping
+  while receiving nothing (2026-07-13 incident: 43.5 MB sent vs 102 KB
+  received over 2d19h, every `send-message` failing with "cannot connect
+  (data exchange port 1001)") — the registry heartbeat is TCP and kept
+  succeeding, so nothing noticed until a manual restart. The daemon now
+  watches for delivered-packet silence while transmit stays active, first
+  soft-recovers (beacon re-registration — whose discover reply doubles as
+  an active inbound probe — plus registry re-registration), and if the
+  wedge persists on a supervised daemon, exits with code 86 so
+  launchd/systemd respawns it with a fresh transport. Guarded against
+  flapping: never exits when the registry is also unreachable (machine
+  offline), when inbound never worked this process, or within 30 min of
+  start. Emits `tunnel.rx_silence` / `tunnel.rx_recovered` /
+  `tunnel.rx_wedged_exit` webhook events. Disable with `-no-rx-watchdog`.
 - **Chunked, ACK'd, resumable file transfer (`TypeFileStream`).** `pilotctl
   send-file` now streams files in 48 KiB chunks with per-chunk ACKs, an
   end-to-end SHA-256 integrity check, and automatic resume from the last
@@ -39,6 +54,12 @@ Reliable P2P data transfer across NAT. Tag intentionally held for review.
   `--motd-feed-url` / `$PILOT_MOTD_URL` as before. (motd)
 
 ### Fixed
+- **`pilotctl daemon start` no longer reports a false failure on slow boots.**
+  The launchd path waited only 10 s for the IPC socket; a daemon with
+  installed app-store apps spawns them before IPC comes up and can take
+  longer, producing "socket did not become ready within 10s" for a start
+  that succeeds moments later. The wait is now 30 s and the timeout message
+  says launchd is still supervising the boot.
 - **NAT traversal now actually establishes (and holds) a direct path.** The
   relay→direct upgrade sent a one-way probe that a stateful NAT/firewall
   always dropped, so peers stayed on the beacon relay indefinitely. The

cmd/daemon/main.go

@@ -94,6 +94,7 @@ func main() {
 	networks := flag.String("networks", "", "comma-separated network IDs to auto-join at startup")
 	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")
 	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)")
@@ -248,6 +249,7 @@ func main() {
 		Version:               version,
 		TrustAutoApprove:      *trustAutoApprove,
 		BeaconRTTProbe:        *beaconRTTProbe,
+		DisableRxWatchdog:     *noRxWatchdog,
 		TransportMode:         *transportMode,
 		CompatBeaconURL:       *compatBeacon,
 		CompatTLSTrust:        *tlsTrust,

cmd/pilotctl/main.go

@@ -2791,8 +2791,12 @@ func cmdDaemonStart(args []string) {
 		if err := exec.Command("launchctl", "bootstrap", fmt.Sprintf("gui/%d", os.Getuid()), plist).Run(); err != nil {
 			fatalCode("internal", "launchctl bootstrap: %v", err)
 		}
-		// Poll socket until daemon is responsive.
-		waitDeadline := time.Now().Add(10 * time.Second)
+		// Poll socket until daemon is responsive. 30s, not 10s: a daemon
+		// with installed app-store apps spawns them before IPC comes up,
+		// which can push readiness past 10s on a loaded machine — and a
+		// premature "not ready" here reads as a failed start even though
+		// launchd keeps supervising and the daemon finishes booting fine.
+		waitDeadline := time.Now().Add(30 * time.Second)
 		for time.Now().Before(waitDeadline) {
 			if d, err := driver.Connect(getSocket()); err == nil {
 				if _, err := d.Info(); err == nil {
@@ -2808,7 +2812,9 @@ func cmdDaemonStart(args []string) {
 			}
 			time.Sleep(200 * time.Millisecond)
 		}
-		fatalCode("timeout", "launchd loaded the agent but the socket did not become ready within 10s")
+		fatalHint("timeout",
+			"launchd is still supervising it — check `pilotctl daemon status` and the daemon log",
+			"launchd loaded the agent but the socket did not become ready within 30s")
 	}
 
 	// Check if already running

go.mod

@@ -1,6 +1,6 @@
 module github.com/pilot-protocol/pilotprotocol
 
-go 1.25.11
+go 1.25.12
 
 require (
 	github.com/coder/websocket v1.8.15

pkg/daemon/daemon.go

@@ -136,6 +136,17 @@ type Config struct {
 	// Feature flags — ablation testing. All default false (current behavior).
 	BeaconRTTProbe bool // probe beacon RTT; override hash pick when >2× slower than best
 
+	// DisableRxWatchdog turns off the inbound-path watchdog (rxwatchdog.go).
+	// The watchdog detects the long-uptime wedge where the daemon keeps
+	// transmitting but the inbound UDP path has silently died (stale NAT /
+	// relay mapping — 2026-07-13 incident: 43.5 MB sent vs 102 KB received
+	// over 2d19h, every send-message failing with "cannot connect"). It
+	// first soft-recovers (beacon + registry re-registration) and, if the
+	// wedge persists on a supervised daemon, exits non-zero so
+	// launchd/systemd respawns with a clean transport. Default false
+	// (watchdog on) — the wedge otherwise requires a manual restart.
+	DisableRxWatchdog 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
@@ -332,6 +343,13 @@ type Daemon struct {
 	netSubMu   sync.Mutex
 	netSubStop func()
 
+	// lastRegistryOKNano is the unix-nano time of the last successful
+	// registry interaction (initial registration, heartbeat, or
+	// re-registration). The rx watchdog uses it to distinguish "data
+	// plane wedged, control plane fine" (restart helps — the incident
+	// signature) from "whole network unreachable" (restart just loops).
+	lastRegistryOKNano atomic.Int64
+
 	startTime       time.Time
 	stopCh          chan struct{}         // closed on Stop() to signal goroutines
 	beaconSelection *beaconSelectionState // multi-beacon discovery state
@@ -653,6 +671,12 @@ func (d *Daemon) SetRekeyWhitelist(nodeIDs []uint32) {
 }
 
 func (d *Daemon) Start() error {
+	// Stamp startTime before any background goroutine launches — the rx
+	// watchdog reads it from its own goroutine for the min-uptime guard,
+	// and the historical assignment at the bottom of Start left a window
+	// where a loop could read the zero value.
+	d.startTime = time.Now()
+
 	// Warm the hostname cache from disk before the IPC server comes up,
 	// so the first send-message after restart hits the cache instead of
 	// pounding regConn for a fresh resolve_hostname.
@@ -1056,6 +1080,7 @@ func (d *Daemon) Start() error {
 	d.addrMu.Unlock()
 
 	slog.Info("daemon registered", "node_id", d.nodeID, "addr", d.addr, "endpoint", registrationAddr)
+	d.lastRegistryOKNano.Store(time.Now().UnixNano())
 
 	// T4.1: webhook is now a bus subscriber owned by plugins/webhook.
 	// cmd/daemon (composition root) constructs the plugin and starts
@@ -1183,6 +1208,11 @@ func (d *Daemon) Start() error {
 	d.bgWG.Add(1)
 	go func() { defer d.bgWG.Done(); d.observabilityHeartbeatLoop() }()
 
+	// 8b. Start inbound-path watchdog (L4). Detects the "transmitting into
+	// a dead NAT mapping" wedge and recovers — see rxwatchdog.go.
+	d.bgWG.Add(1)
+	go func() { defer d.bgWG.Done(); d.rxWatchdogLoop() }()
+
 	// 9. Start idle connection sweeper
 	d.bgWG.Add(1)
 	go func() { defer d.bgWG.Done(); d.idleSweepLoop() }()
@@ -1260,7 +1290,6 @@ func (d *Daemon) Start() error {
 	// runtime.StartPlugins(ctx) AFTER this Start returns. This
 	// inversion is T7.1: pkg/daemon no longer imports pkg/coreapi.
 
-	d.startTime = time.Now()
 	slog.Info("daemon running", "node_id", d.nodeID, "addr", d.addr)
 	return nil
 }
@@ -4950,6 +4979,7 @@ func (d *Daemon) trustRepublishLoop() {
 				}
 				consecutiveFailures = 0
 				reregBackoff = 100 * time.Millisecond
+				d.lastRegistryOKNano.Store(time.Now().UnixNano())
 			}
 		}
 	}
@@ -5102,6 +5132,7 @@ func (d *Daemon) reRegister() {
 	nodeID := d.nodeID
 	slog.Info("re-registered", "node_id", nodeID, "addr", d.addr)
 	d.addrMu.Unlock()
+	d.lastRegistryOKNano.Store(time.Now().UnixNano())
 	d.publishEvent("node.reregistered", map[string]interface{}{
 		"address": d.addr.String(),
 	})