daemon: event-driven path reset on rekey give-up (corrects #415, verified live)

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

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 }

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 {

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

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

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