From ca485915fe6a6adc5d855667e550e2214222e459 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Wed, 22 Jul 2026 20:15:07 +0300 Subject: [PATCH] daemon: event-driven path reset on rekey give-up (corrects #415, verified live) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #415 tried to recover the T2 desync by polling the rekey-gave-up state from the path watchdog. A LIVE test (2026-07-22) proved it never fired: a desynced peer has no usable key, so it is not Ready, so ReadyPeerIDs() — the set the watchdog scans — excludes it entirely. 0 firings across dozens of real give-ups. (The unit test passed only because it injected a Ready peer + gave-up state, a combination that can't occur.) Correct fix: event-driven. Add an onGaveUp hook fired from the keyexchange retransmit loop at the exact give-up site, wired through the tunnel to a daemon handler (onRekeyGaveUp) that runs resetPeerPath — the only thing that recovers a desync, because it re-resolves a FRESH endpoint and re-punches (plain rekey-retransmit just resends to the same stale cached endpoint forever). Per-peer 30s cooldown prevents a persistently-offline peer from storming resolves; reset runs async (registry I/O) off the rekey loop goroutine. Reverts #415's ineffective path-watchdog poll. Verified LIVE on a real NAT'd laptop: give-up -> full path reset (fresh resolve + punch + PILA) in ~1.2s, reproduced across peers 17461 and 16392, vs the prior indefinite silent-drop. This is the 'no reply on whatever' fix. Tests: TestOnRekeyGaveUpResetsWithCooldown (hook resets + cooldown suppresses a second give-up). pkg/daemon + keyexchange green under -race. Co-Authored-By: Claude Opus 4.8 --- pkg/daemon/daemon.go | 11 +++- pkg/daemon/keyexchange/keyexchange.go | 9 +++ pkg/daemon/keyexchange/loop.go | 11 ++++ pkg/daemon/pathwatch.go | 86 +++++++++++++++++++-------- pkg/daemon/tunnel.go | 16 +++++ pkg/daemon/zz_pathwatch_test.go | 71 +++++++++++++--------- 6 files changed, 148 insertions(+), 56 deletions(-) diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 0e5a9c2f..129a34b6 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -468,6 +468,11 @@ type Daemon struct { netPolicyMu sync.RWMutex netPolicies map[uint16][]uint16 + // gaveUpResetMu guards lastGaveUpReset, the per-peer cooldown for + // rekey-gave-up-triggered path resets (onRekeyGaveUp). + gaveUpResetMu sync.Mutex + lastGaveUpReset map[uint32]time.Time + // Managed network engines: netID -> engine (older "managed network" // feature — doesn't reference pkg/policy, stays in pkg/daemon). managedMu sync.Mutex @@ -550,12 +555,16 @@ func New(cfg Config) *Daemon { epCache: make(map[uint32]*endpointEntry), resolveCache: make(map[uint32]*resolveEntry), hostnameCache: make(map[string]*hostnameCacheEntry), - netPolicies: make(map[uint16][]uint16), + netPolicies: make(map[uint16][]uint16), + lastGaveUpReset: make(map[uint32]time.Time), managed: make(map[uint16]*ManagedEngine), memberTags: make(map[uint16][]string), } d.ctx, d.cancelCtx = context.WithCancel(context.Background()) d.bus = newInProcessBus(d.NodeID) + // Event-driven T2 recovery: reset a peer's path the moment the rekey + // machinery gives up on it (see onRekeyGaveUp / pathwatch.go). + d.tunnels.SetRekeyGaveUpHook(d.onRekeyGaveUp) d.ipc = NewIPCServer(cfg.SocketPath, d) // HandshakeService is wired post-construction by the composition // root via RegisterHandshakeService (T3.3 — handshake plugin moved diff --git a/pkg/daemon/keyexchange/keyexchange.go b/pkg/daemon/keyexchange/keyexchange.go index 5f685e19..e8d34f86 100644 --- a/pkg/daemon/keyexchange/keyexchange.go +++ b/pkg/daemon/keyexchange/keyexchange.go @@ -262,6 +262,7 @@ type Manager struct { publisher EventPublisher postInstall PostInstallHook preRetx PreRetransmitHook + onGaveUp func(nodeID uint32) } // New returns a fresh Manager. The Manager installs into store; pass @@ -306,6 +307,14 @@ func (m *Manager) SetPostInstallHook(h PostInstallHook) { m.postInstall = h } // routing concerns into the keyexchange package. func (m *Manager) SetPreRetransmitHook(h PreRetransmitHook) { m.preRetx = h } +// SetOnGaveUpHook wires a callback fired once per peer at the moment the +// rekey machinery gives up (MaxRekeyAttempts retransmits to a stale +// cached endpoint, all unanswered). The daemon uses this to trigger a +// full path reset (fresh resolve + hole-punch) — the ONLY thing that +// recovers a desynced peer, since it is no longer Ready and so is invisible +// to the inbound-silence path watchdog. Invoked outside rkPendingMu. +func (m *Manager) SetOnGaveUpHook(h func(nodeID uint32)) { m.onGaveUp = h } + // SetLocalNodeIDFn supplies the closure used to read our own node ID // (atomic read living in the daemon). func (m *Manager) SetLocalNodeIDFn(f func() uint32) { m.localNodeIDFn = f } diff --git a/pkg/daemon/keyexchange/loop.go b/pkg/daemon/keyexchange/loop.go index bd8b0601..6b2afb99 100644 --- a/pkg/daemon/keyexchange/loop.go +++ b/pkg/daemon/keyexchange/loop.go @@ -73,6 +73,17 @@ func (m *Manager) RekeyRetransmitTick() { } m.rkPendingMu.Unlock() + // Fire the gave-up hook for each peer we just abandoned — OUTSIDE the + // lock (the daemon's handler does registry I/O). This is the only + // recovery signal for a desynced peer: it is no longer Ready, so the + // inbound-silence path watchdog can't see it; the hook triggers a full + // path reset (fresh resolve + hole-punch). + if m.onGaveUp != nil { + for _, id := range toGiveUp { + m.onGaveUp(id) + } + } + // Cap retransmits per tick to prevent UDP flooding when many peers // are pending simultaneously. Remaining entries are retried next tick. if len(toRetry) > maxRetransmitsPerTick { diff --git a/pkg/daemon/pathwatch.go b/pkg/daemon/pathwatch.go index d4dfba4b..62b7e375 100644 --- a/pkg/daemon/pathwatch.go +++ b/pkg/daemon/pathwatch.go @@ -149,35 +149,13 @@ func (d *Daemon) pathWatchPeer(nodeID uint32, st *pathPeerState, now time.Time) if !ok { // Session marked Ready but no decrypt recorded yet — key // exchange just completed. Leave it to the rekey machinery. + // (A DESYNCED peer with no usable key is not Ready and so is not + // scanned here at all — that case is handled event-driven by the + // rekey-gave-up hook, onRekeyGaveUp, not by this silence loop.) return pathActionSkip } silence := now.Sub(last) - // Fast path for the T2 desync (2026-07-20 probe): when the rekey - // machinery has GIVEN UP on a peer, the session is unambiguously dead - // — both sides hold mismatched keys and every inbound frame is - // dropped undecryptable. Plain rekey-retransmit can't recover it (it - // resends the key exchange to the same stale cached endpoint); a path - // reset can, because it re-resolves a fresh endpoint and re-punches. - // Reset immediately rather than waiting out the ~85s inbound-silence - // probe budget — but still honour the post-reset cooldown so a truly - // offline peer can't cause a reset storm. - inCooldown := !st.lastResetAt.IsZero() && now.Sub(st.lastResetAt) < pathResetCooldown - if d.tunnels.PeerInRekeyGaveUp(nodeID) && !inCooldown { - slog.Warn("path watchdog: peer rekey gave up (session desync) — resetting path now", - "peer_node_id", nodeID, - "silent_for", silence.Truncate(time.Second).String()) - d.publishEvent("tunnel.path_suspect", map[string]any{ - "peer_node_id": nodeID, - "silent_for_seconds": int64(silence.Seconds()), - "reason": "rekey_gave_up", - }) - pathWatchResetPeer(d, nodeID) - st.probesSent = 0 - st.lastResetAt = now - return pathActionReset - } - if silence < pathSilenceThreshold { if st.probesSent > 0 { slog.Info("path watchdog: peer path recovered", @@ -196,7 +174,7 @@ func (d *Daemon) pathWatchPeer(nodeID uint32, st *pathPeerState, now time.Time) // Inside the post-reset cooldown: don't re-probe/re-reset a peer // that is likely just offline. - if inCooldown { + if !st.lastResetAt.IsZero() && now.Sub(st.lastResetAt) < pathResetCooldown { return pathActionCooldown } @@ -296,3 +274,59 @@ func (d *Daemon) resetPeerPath(nodeID uint32) peerPathReset { } return res } + +// gaveUpResetCooldown bounds how often a single peer's path is reset in +// response to a rekey-gave-up event. The rekey machinery re-arms roughly +// every RekeyRetransmitInterval*MaxRekeyAttempts + rekeyGaveUpCooldown +// (~30s) for a persistently-desynced peer; this cooldown prevents a +// genuinely-offline peer from triggering a resolve/punch every cycle. +const gaveUpResetCooldown = 30 * time.Second + +// onRekeyGaveUp is the daemon-owned handler wired to the keyexchange +// gave-up hook (SetRekeyGaveUpHook). A peer the rekey machinery has +// abandoned is desynced — no usable key, so it is NOT Ready and the +// inbound-silence path watchdog never scans it. This is the recovery +// path: reset the peer (drop cached endpoint + tunnel, re-resolve fresh, +// re-punch, push a fresh PILA) so a working session can re-establish. +// Runs the reset asynchronously (it does registry I/O) and rate-limits +// per peer so a flapping/offline peer can't storm resolves. Fired from +// the keyexchange retransmit loop goroutine. +func (d *Daemon) onRekeyGaveUp(nodeID uint32) { + now := time.Now() + d.gaveUpResetMu.Lock() + if last, ok := d.lastGaveUpReset[nodeID]; ok && now.Sub(last) < gaveUpResetCooldown { + d.gaveUpResetMu.Unlock() + return + } + d.lastGaveUpReset[nodeID] = now + // Opportunistic prune so the map can't grow without bound. + if len(d.lastGaveUpReset) > 1024 { + for id, t := range d.lastGaveUpReset { + if now.Sub(t) > 5*gaveUpResetCooldown { + delete(d.lastGaveUpReset, id) + } + } + } + d.gaveUpResetMu.Unlock() + + go func() { + defer recoverLayer("L4", "onRekeyGaveUp", d.bus, nil) + res := gaveUpResetPeer(d, nodeID) + slog.Warn("rekey gave up (session desync) — reset peer path", + "peer_node_id", nodeID, + "had_tunnel", res.HadTunnel, + "was_relay_active", res.WasRelayActive, + "pila_pushed", res.PilaPushed, + "resolve_error", res.ResolveErr) + d.publishEvent("tunnel.path_suspect", map[string]any{ + "peer_node_id": nodeID, + "reason": "rekey_gave_up", + }) + }() +} + +// gaveUpResetPeer is swapped by tests to observe the reset without a live +// registry/beacon (mirrors pathWatchResetPeer). +var gaveUpResetPeer = func(d *Daemon, nodeID uint32) peerPathReset { + return d.resetPeerPath(nodeID) +} diff --git a/pkg/daemon/tunnel.go b/pkg/daemon/tunnel.go index 9e24d19e..cec46527 100644 --- a/pkg/daemon/tunnel.go +++ b/pkg/daemon/tunnel.go @@ -163,6 +163,10 @@ type TunnelManager struct { // (preserves the leaf-lock invariant from 03-INVARIANTS.md §3). routing *routing.Manager + // onRekeyGaveUp is the daemon-owned handler fired when the rekey + // machinery abandons a peer (see SetRekeyGaveUpHook / resetPeerPath). + onRekeyGaveUp func(nodeID uint32) + // Event bus — replaces inline tm.webhook.Emit calls. Set via // SetEventBus from daemon during construction. May be nil in // tests; tm.publishEvent handles that. Webhook delivery is a @@ -314,9 +318,21 @@ func NewTunnelManager() *TunnelManager { tm.kx.SetLocalNodeIDFn(tm.loadNodeID) tm.kx.SetPostInstallHook(tm.onKeyInstalled) tm.kx.SetPreRetransmitHook(tm.maybeForceRelayOnRekey) + tm.kx.SetOnGaveUpHook(func(nodeID uint32) { + if fn := tm.onRekeyGaveUp; fn != nil { + fn(nodeID) + } + }) return tm } +// SetRekeyGaveUpHook wires the daemon's handler for the rekey-gave-up +// event (owned by the daemon because recovery is a path reset). Called +// once during daemon setup. +func (tm *TunnelManager) SetRekeyGaveUpHook(fn func(nodeID uint32)) { + tm.onRekeyGaveUp = fn +} + // maybeForceRelayOnRekey is the cross-layer policy bridge between the // keyexchange retransmit loop and the routing layer's relay path. // Invoked once per peer per pending retransmit, BEFORE the frame goes diff --git a/pkg/daemon/zz_pathwatch_test.go b/pkg/daemon/zz_pathwatch_test.go index f3e53294..b4af5aeb 100644 --- a/pkg/daemon/zz_pathwatch_test.go +++ b/pkg/daemon/zz_pathwatch_test.go @@ -4,6 +4,7 @@ package daemon import ( "net" + "sync" "testing" "time" @@ -226,40 +227,52 @@ func TestResetPeerPathPreservesRelayActive(t *testing.T) { } } -// TestPathWatchRekeyGaveUpResetsImmediately pins the T2 fast-path: a peer -// whose rekey machinery has given up is a dead session, so the watchdog -// resets it on the current tick instead of waiting out the inbound- -// silence probe budget — but still honours the post-reset cooldown. -func TestPathWatchRekeyGaveUpResetsImmediately(t *testing.T) { - const peer = 91 - d := newPathWatchTestDaemon(t, peer) - resets := swapPathResetForTest(t) - now := time.Now() - // A gave-up peer is inbound-silent (that's why it desynced). Set - // silence PAST the threshold but leave probesSent=0: without the fast - // path this tick would only send the first probe (probesSent 0->1); - // with it, the gave-up signal resets immediately. Asserting Reset with - // probesSent still 0 proves the fast path beats the probe budget. - d.tunnels.kx.SetLastInboundDecryptForTest(peer, now.Add(-2*pathSilenceThreshold)) - d.tunnels.kx.MarkRekeyGaveUpForTest(peer) - st := &pathPeerState{} - if got := d.pathWatchPeer(peer, st, now); got != pathActionReset { - t.Fatalf("rekey-gave-up action = %q, want %q (should reset now, not probe)", got, pathActionReset) - } - if len(*resets) != 1 || (*resets)[0] != peer { - t.Fatalf("resets = %v, want [%d]", *resets, peer) + +// TestOnRekeyGaveUpResetsWithCooldown pins the event-driven T2 recovery: +// the rekey-gave-up hook resets a peer's path, and a per-peer cooldown +// prevents a persistently-desynced/offline peer from storming resolves. +// The 2026-07-22 live test proved the earlier poll-from-pathwatch +// approach never fired (a desynced peer isn't Ready, so pathwatch never +// scanned it) — this hook fires straight from the giveup site. +func TestOnRekeyGaveUpResetsWithCooldown(t *testing.T) { + d := &Daemon{ + tunnels: NewTunnelManager(), + startTime: time.Now(), + stopCh: make(chan struct{}), + lastGaveUpReset: make(map[uint32]time.Time), } - if st.lastResetAt.IsZero() { - t.Fatal("reset must stamp cooldown") + var resets []uint32 + var mu sync.Mutex + prev := gaveUpResetPeer + gaveUpResetPeer = func(_ *Daemon, id uint32) peerPathReset { + mu.Lock() + resets = append(resets, id) + mu.Unlock() + return peerPathReset{} } + t.Cleanup(func() { gaveUpResetPeer = prev }) + + const peer = 77 + d.onRekeyGaveUp(peer) // fires an async reset + // second giveup immediately: cooldown must suppress it + d.onRekeyGaveUp(peer) - // Immediately after, still gave-up but inside cooldown → no second reset. - if got := d.pathWatchPeer(peer, st, now.Add(time.Second)); got != pathActionCooldown { - t.Fatalf("in-cooldown action = %q, want %q (no reset storm)", got, pathActionCooldown) + // let the async reset goroutine run + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + mu.Lock() + n := len(resets) + mu.Unlock() + if n >= 1 { + break + } + time.Sleep(20 * time.Millisecond) } - if len(*resets) != 1 { - t.Fatalf("cooldown must suppress a second reset, resets=%v", *resets) + mu.Lock() + defer mu.Unlock() + if len(resets) != 1 || resets[0] != peer { + t.Fatalf("resets = %v, want exactly [%d] (cooldown must suppress the 2nd giveup)", resets, peer) } }