daemon: a dial timeout while another dial succeeds is not a transport wedge

CHANGELOG.md

@@ -439,6 +439,15 @@ Detailed per-release notes are on the
     one per 5s after that, whatever arrives.
   An idle node still makes one poll per minute, as before. Nodes and
   servers that are not updated keep working at the old pace for their part.
+- **Dial timeouts while other dials succeed no longer read as a wedged
+  transport.** The watchdog's `outbound-dial-wedged` check counted every
+  dial timeout, so a burst of parallel sends to one peer whose SYN limiter
+  dropped some of them (`dial_timeouts=32`) made a healthy node re-register
+  with the beacon and registry, and could escalate to a restart. A timeout
+  is no longer counted when another dial completed while it ran: that dial's
+  SYN-ACK shows the outbound path was working. Traffic merely received from
+  the peer does not count as evidence, because a node whose outbound path is
+  dead still receives its peers' keepalives.
 - **Proxy credential hints no longer send an operator who already set
   `proxy_cmd` off to set it.** When the daemon re-reads its credentials with
   a proxy command and the proxy still rejects them (407, or Meta Muse's

pkg/daemon/daemon.go

@@ -451,7 +451,12 @@ type Daemon struct {
 	// it. consecutiveDialTimeouts counts back-to-back dial timeouts
 	// (reset to 0 on any successful dial); a run of them to distinct
 	// peers means our outbound path is wedged, not that one peer is dead.
+	// A timeout is not counted when another dial succeeded while it ran
+	// (see dialConnectionLocked): dialOKSeq counts completed dials, and a
+	// dial compares it with the count it saw when it started. A counter,
+	// not lastDialOKNano, so a wall-clock step cannot hide timeouts.
 	lastDialOKNano          atomic.Int64
+	dialOKSeq               atomic.Uint64
 	consecutiveDialTimeouts atomic.Uint64
 
 	startTime       time.Time
@@ -3897,6 +3902,7 @@ func (d *Daemon) dialConnectionLocked(ctx context.Context, dstAddr protocol.Addr
 	if err := ctx.Err(); err != nil {
 		return nil, err
 	}
+	dialsOKAtStart := d.dialOKSeq.Load()
 
 	// Enforce outbound port policy: prevent dialing ports blocked by the network
 	if !d.evaluatePortPolicy(PolicyEventDial, dstAddr.Network, dstPort, dstAddr.Node, 0, "") {
@@ -4088,6 +4094,7 @@ func (d *Daemon) dialConnectionLocked(ctx context.Context, dstAddr protocol.Addr
 			// A completed handshake proves the outbound path is alive —
 			// clears the rx-watchdog's partial-wedge signal.
 			d.lastDialOKNano.Store(time.Now().UnixNano())
+			d.dialOKSeq.Add(1)
 			d.consecutiveDialTimeouts.Store(0)
 			return conn, nil, true
 		}
@@ -4196,7 +4203,19 @@ func (d *Daemon) dialConnectionLocked(ctx context.Context, dstAddr protocol.Addr
 				// Full direct+relay retry budget exhausted with no SYN-ACK.
 				// Feeds the rx-watchdog's partial-wedge detector: a run of
 				// these to distinct peers means our outbound is wedged.
-				d.consecutiveDialTimeouts.Add(1)
+				// Not when another dial completed while this one ran: that
+				// SYN reached its peer and the SYN-ACK came back, so our
+				// outbound path worked during this dial and this one was
+				// declined (a peer's SYN limiter under a burst of our own
+				// dials, a full accept path). Traffic merely received from
+				// the peer is no evidence — its keepalives still arrive
+				// when it is our outbound path that is dead.
+				if d.dialOKSeq.Load() != dialsOKAtStart {
+					slog.Debug("dial timed out while another dial succeeded; not counted as a transport wedge",
+						"peer_node_id", dstAddr.Node, "dst_port", dstPort)
+				} else {
+					d.consecutiveDialTimeouts.Add(1)
+				}
 				if keyMissing {
 					// Not one SYN left this node: the peer never completed
 					// the key exchange. Say so, so callers can tell "busy

pkg/daemon/rxwatchdog.go

@@ -77,7 +77,10 @@ const (
 	// dead, not the peers. This is the PARTIAL wedge the rx-silence check
 	// misses: rx keeps trickling (keepalives) so silence never trips, but
 	// no new connection can be opened. 4 = at least four back-to-back
-	// user-visible dial failures with zero successes in between.
+	// user-visible dial failures with zero successes in between. A timeout
+	// during which another dial succeeded is not counted: 32 parallel sends
+	// to one peer whose SYN limiter dropped some of them once read as a
+	// wedge and re-registered a healthy node.
 	dialWedgeThreshold = 4
 
 	// rxWatchdogSoftMax is how many consecutive soft recoveries run

pkg/daemon/zz_rx_watchdog_dial_wedge_test.go

@@ -0,0 +1,177 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package daemon
+
+import (
+	"context"
+	"errors"
+	"net"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/pilot-protocol/common/protocol"
+)
+
+// dialTimeoutBurst runs dialWedgeThreshold concurrent dials to one peer that
+// never completes a handshake and returns the daemon once every dial has
+// spent its full retry budget. before, if set, runs before the dials start;
+// during, if set, one second into them.
+func dialTimeoutBurst(t *testing.T, before, during func(d *Daemon, peerNode uint32)) *Daemon {
+	t.Helper()
+	d := New(Config{})
+	t.Cleanup(func() { d.tunnels.Close() })
+
+	const peerNode uint32 = 0xCAFEBABE
+	peerConn := addPeerOnDaemon(t, d, peerNode)
+	t.Cleanup(func() { peerConn.Close() })
+	d.setNodeID_testhelper(0x33330000)
+	if before != nil {
+		before(d, peerNode)
+	}
+
+	dst := protocol.Addr{Network: 0, Node: peerNode}
+	var wg sync.WaitGroup
+	for i := 0; i < dialWedgeThreshold; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if _, err := d.DialConnectionContext(context.Background(), dst, 1001); !errors.Is(err, protocol.ErrDialTimeout) {
+				t.Errorf("dial: err = %v, want ErrDialTimeout", err)
+			}
+		}()
+	}
+	if during != nil {
+		time.Sleep(time.Second)
+		during(d, peerNode)
+	}
+	wg.Wait()
+	return d
+}
+
+// TestRxWatchdogDialTimeoutsWhileAnotherDialSucceedsAreNotAWedge: a burst of
+// dials times out while another dial completes — its SYN reached the peer
+// and the SYN-ACK came back, so this node's outbound path demonstrably
+// worked (a peer's SYN limiter answered some of our burst and dropped the
+// rest). That says something about the peer, not about this node's
+// transport, so the watchdog must not start its recovery (beacon and
+// registry re-registration, escalating to a process exit). Before the fix
+// every such timeout bumped the global consecutiveDialTimeouts and four of
+// them tripped "outbound-dial-wedged".
+func TestRxWatchdogDialTimeoutsWhileAnotherDialSucceedsAreNotAWedge(t *testing.T) {
+	if testing.Short() {
+		t.Skip("long retry-budget test")
+	}
+	t.Parallel()
+
+	now := time.Now()
+	d := dialTimeoutBurst(t, nil, func(d *Daemon, peerNode uint32) {
+		completeDial(t, d, 0xCAFED00D)
+	})
+	if c := d.consecutiveDialTimeouts.Load(); c != 0 {
+		t.Errorf("consecutiveDialTimeouts = %d after timeouts during which another dial succeeded, want 0", c)
+	}
+
+	st := &rxWatchdogState{lastProgress: now, recvSeen: atomic.LoadUint64(&d.tunnels.PktsRecv)}
+	atomic.AddUint64(&d.tunnels.PktsRecv, 25)
+	if got := d.rxWatchdogTick(st, now.Add(rxWatchdogTickInterval)); got != rxActionProgress {
+		t.Fatalf("dial timeouts while another dial succeeded: action = %q, want %q", got, rxActionProgress)
+	}
+}
+
+// TestRxWatchdogPeerKeepalivesDoNotHideAWedge: a node whose outbound path is
+// dead still receives its peers' keepalives, so traffic from the dialled peer
+// during the dials is not evidence that the transport works. An earlier
+// version of this fix treated it as such, and a node with every outbound
+// frame dropped never reached the threshold.
+func TestRxWatchdogPeerKeepalivesDoNotHideAWedge(t *testing.T) {
+	if testing.Short() {
+		t.Skip("long retry-budget test")
+	}
+	t.Parallel()
+
+	now := time.Now()
+	d := dialTimeoutBurst(t, nil, func(d *Daemon, peerNode uint32) {
+		d.tunnels.kx.RecordInboundDecrypt(peerNode) // the peer's keepalive arrives
+	})
+	if c := d.consecutiveDialTimeouts.Load(); c != dialWedgeThreshold {
+		t.Fatalf("consecutiveDialTimeouts = %d with only the peer's keepalives arriving, want %d", c, dialWedgeThreshold)
+	}
+
+	st := &rxWatchdogState{lastProgress: now, recvSeen: atomic.LoadUint64(&d.tunnels.PktsRecv)}
+	atomic.AddUint64(&d.tunnels.PktsRecv, 3)
+	if got := d.rxWatchdogTick(st, now.Add(rxWatchdogTickInterval)); got != rxActionSoftRecover {
+		t.Fatalf("outbound dead, keepalives arriving: action = %q, want %q", got, rxActionSoftRecover)
+	}
+}
+
+// TestRxWatchdogDialSuccessBeforeTheBurstDoesNotHideIt: only a dial that
+// completes while a timed-out dial ran says the outbound path worked during
+// it. One that completed earlier says nothing about the dials after it.
+func TestRxWatchdogDialSuccessBeforeTheBurstDoesNotHideIt(t *testing.T) {
+	if testing.Short() {
+		t.Skip("long retry-budget test")
+	}
+	t.Parallel()
+
+	d := dialTimeoutBurst(t, func(d *Daemon, peerNode uint32) {
+		completeDial(t, d, 0xCAFED00E)
+	}, nil)
+	if c := d.consecutiveDialTimeouts.Load(); c != dialWedgeThreshold {
+		t.Fatalf("consecutiveDialTimeouts = %d after timeouts that all started after the last success, want %d", c, dialWedgeThreshold)
+	}
+}
+
+// completeDial dials a second, answering peer and completes the handshake
+// the way the network would: its SYN is answered with a SYN-ACK.
+func completeDial(t *testing.T, d *Daemon, peer uint32) {
+	t.Helper()
+	pc, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { pc.Close() })
+	d.AddTunnelPeer(peer, pc.LocalAddr().(*net.UDPAddr))
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := d.DialConnectionContext(context.Background(), protocol.Addr{Network: 0, Node: peer}, 1001)
+		done <- err
+	}()
+	var syn *protocol.Packet
+	for deadline := time.Now().Add(5 * time.Second); syn == nil && time.Now().Before(deadline); {
+		if p := recvPacket(t, pc, time.Second); p != nil && p.Flags&protocol.FlagSYN != 0 {
+			syn = p
+		}
+	}
+	if syn == nil {
+		t.Fatal("no SYN reached the answering peer")
+	}
+	d.handleStreamPacket(streamPacket(protocol.FlagSYN|protocol.FlagACK, peer, d.NodeID(), syn.DstPort, syn.SrcPort, 999, syn.Seq+1))
+	if err := <-done; err != nil {
+		t.Fatalf("dial to the answering peer: %v", err)
+	}
+}
+
+// TestRxWatchdogDialTimeoutsToSilentPeerStillCount keeps the partial-wedge
+// detector (2026-07-15) intact: with no dial succeeding meanwhile, every
+// timeout counts and the watchdog recovers even though rx is trickling.
+func TestRxWatchdogDialTimeoutsToSilentPeerStillCount(t *testing.T) {
+	if testing.Short() {
+		t.Skip("long retry-budget test")
+	}
+	t.Parallel()
+
+	now := time.Now()
+	d := dialTimeoutBurst(t, nil, nil)
+	if c := d.consecutiveDialTimeouts.Load(); c != dialWedgeThreshold {
+		t.Fatalf("consecutiveDialTimeouts = %d after timeouts to a silent peer, want %d", c, dialWedgeThreshold)
+	}
+
+	st := &rxWatchdogState{lastProgress: now, recvSeen: atomic.LoadUint64(&d.tunnels.PktsRecv)}
+	atomic.AddUint64(&d.tunnels.PktsRecv, 3) // keepalive trickle from other peers
+	if got := d.rxWatchdogTick(st, now.Add(rxWatchdogTickInterval)); got != rxActionSoftRecover {
+		t.Fatalf("dial timeouts to a silent peer: action = %q, want %q", got, rxActionSoftRecover)
+	}
+}