[client] Drop agentConnecting whenever ICE session state clears (#7327)

* [client] Drop agentConnecting whenever ICE session state clears

Closing a WorkerICE raced a blocked dial goroutine: Close released the
agent while connect() was still inside Dial, and the goroutine's own
cleanup skipped its flag reset because w.agent no longer matched. With
agentConnecting stuck on true, evalConnStatus read the peer as
connected, the reconnection guard stopped sending offers and
same-session offers were dropped, so the peer could not recover without
a restart. An aborted recreate in OnNewOffer reaches the same wedged
state without any race.

Route every teardown path through one abandonNegotiation helper so the
agent and flag fields always clear together; Close now also cleans up
residual state left by an aborted recreate.

* [client] Drive the ICE teardown race test through the real dial goroutine

The regression test simulated the stale goroutine by calling closeAgent
directly, so it pinned the symptom rather than the mechanism. Rework it
to start a real negotiation, tear it down mid-flight and let the actual
goroutine run its own cleanup: with no remote responder the dial can
only fail once Close cancels it, so the interleaving stays deterministic
without sleeps or injection points.

Assert the full idle state that abandonNegotiation owns (agent nil,
connecting false, remote session ID empty) instead of only InProgress,
and make the stale-cleanup ownership test verify that the newer session
survives field by field.

* [client] Assert live remote session ID after stale ICE cleanup

The stale-cleanup test compared a snapshot captured before closeAgent
ran, so clearing the field during cleanup would have gone unnoticed.
Read the field under the mutex after the cleanup instead.

* [client] Give the ICE race tests a no-op signal client

The candidate callback fires from a real gather and dereferences the
signaler, so a nil one crashes the test package intermittently when
gather wins the race against Close. Build the worker with a stub
signal.Client instead.

* [client] Read the ICE dial cancel func from an argument in connect

The error paths read w.agentDialerCancel without holding muxAgent while
OnNewOffer rewrites the field for a newer negotiation, a data race the
new teardown test trips under -race. Reading a stale value also let an
old goroutine cancel another session's dial. Capture the cancel func at
goroutine spawn, like the dial context already is.

* [client] Guard the ICE dial success path against stale negotiations

The stale-cleanup guard in closeAgent only protected teardown. Its
success-path counterpart was missing: an older negotiation could complete
agentDial after a newer one replaced w.agent, then clear the newer
session's agentConnecting, record lastSuccess and publish its dead
connection via onICEConnectionIsReady.

Verify ownership under muxAgent twice: right after the dial returns, so a
stale goroutine drops its connection before touching a closed agent, and
again at the state-commit point, atomic with the agentConnecting and
lastSuccess writes, so a replacement arriving in the meantime cannot get
its state clobbered. Both paths close the stale connection and return
without modifying worker state. A regression test holds session A's dial
open until session B is installed, then releases it; the stale connection
must be discarded and B's agent, connecting flag and remote session ID
must survive.

* [client] Fix ICE teardown test leak and document the stale delivery window

A code review of the stale-negotiation guard found a leftover resource
leak in TestWorkerICE_StaleCloseAgentKeepsCurrentSession: session B is
never closed, so its ICE sockets and blocked dial goroutine live as long
as the test process. Register t.Cleanup(w.Close).

The delivery race flagged after the success-path guard is pre-existing
and self-correcting - the newer negotiation overwrites the transient
endpoint - so document it in the existing todo instead of locking the
callback, which would invert lock order against Conn.Close. Adjust the
teardown test comment to match the now-synchronous Close flag clearing.

client/internal/peer/worker_ice.go

@@ -64,6 +64,9 @@ type WorkerICE struct {
 
 	// portForwardAttempted tracks if we've already tried port forwarding this session
 	portForwardAttempted bool
+
+	// dialFunc, when non-nil, replaces agentDial in connect(). Only for tests.
+	dialFunc func(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) (net.Conn, error)
 }
 
 func NewWorkerICE(ctx context.Context, log *log.Entry, config ConnConfig, conn *Conn, signaler *Signaler, ifaceDiscover stdnet.ExternalIFaceDiscover, statusRecorder *Status, hasRelayOnLocally bool) (*WorkerICE, error) {
@@ -123,7 +126,7 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 			w.log.Errorf("failed to create new session ID: %s", err)
 		}
 		w.sessionID = sessionID
-		w.agent = nil
+		w.abandonNegotiation()
 	}
 
 	var preferredCandidateTypes []ice.CandidateType
@@ -151,7 +154,9 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 		w.remoteSessionID = ""
 	}
 
-	go w.connect(dialerCtx, agent, remoteOfferAnswer)
+	// Capture the cancel func at spawn time: connect reads it from the argument
+	// instead of the field, which a newer OnNewOffer may already have replaced.
+	go w.connect(dialerCtx, dialerCancel, agent, remoteOfferAnswer)
 }
 
 // OnRemoteCandidate Handles ICE connection Candidate provided by the remote peer.
@@ -200,16 +205,16 @@ func (w *WorkerICE) Close() {
 	w.muxAgent.Lock()
 	defer w.muxAgent.Unlock()
 
-	if w.agent == nil {
-		return
+	if w.agent != nil {
+		w.agentDialerCancel()
+		if err := w.agent.Close(); err != nil {
+			w.log.Warnf("failed to close ICE agent: %s", err)
+		}
 	}
-
-	w.agentDialerCancel()
-	if err := w.agent.Close(); err != nil {
-		w.log.Warnf("failed to close ICE agent: %s", err)
-	}
-
-	w.agent = nil
+	// Unconditional: a dial goroutine racing this Close skips its own cleanup
+	// (closeAgent finds a nil agent), so the flags must be dropped here too or
+	// the reconnection guard reads the stale state as Connected forever.
+	w.abandonNegotiation()
 }
 
 func (w *WorkerICE) reCreateAgent(dialerCancel context.CancelFunc, candidates []ice.CandidateType) (*icemaker.ThreadSafeAgent, error) {
@@ -247,31 +252,52 @@ func (w *WorkerICE) SessionID() ICESessionID {
 // will block until connection succeeded
 // but it won't release if ICE Agent went into Disconnected or Failed state,
 // so we have to cancel it with the provided context once agent detected a broken connection
-func (w *WorkerICE) connect(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) {
+func (w *WorkerICE) connect(ctx context.Context, dialerCancel context.CancelFunc, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) {
 	w.log.Debugf("gather candidates")
 	if err := agent.GatherCandidates(); err != nil {
 		w.log.Warnf("failed to gather candidates: %s", err)
-		w.closeAgent(agent, w.agentDialerCancel)
+		w.closeAgent(agent, dialerCancel)
 		return
 	}
 
 	w.log.Debugf("agent dial")
-	remoteConn, err := w.agentDial(ctx, agent, remoteOfferAnswer)
+	dial := func(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) (net.Conn, error) {
+		return w.agentDial(ctx, agent, remoteOfferAnswer)
+	}
+	if w.dialFunc != nil {
+		dial = w.dialFunc
+	}
+	remoteConn, err := dial(ctx, agent, remoteOfferAnswer)
 	if err != nil {
 		w.log.Debugf("failed to dial the remote peer: %s", err)
-		w.closeAgent(agent, w.agentDialerCancel)
+		w.closeAgent(agent, dialerCancel)
 		return
 	}
 	w.log.Debugf("agent dial succeeded")
 
+	// A newer negotiation may have replaced our agent while agentDial was
+	// blocked. Drop the dead connection before running pair retrieval, port
+	// punching or candidate work against a closed agent. The commit-point
+	// check below still guards a replacement arriving after this point.
+	w.muxAgent.Lock()
+	stale := w.agent != agent
+	w.muxAgent.Unlock()
+	if stale {
+		if err := remoteConn.Close(); err != nil {
+			w.log.Warnf("failed to close stale ICE connection: %s", err)
+		}
+		w.log.Warnf("discarding connection from a stale ICE negotiation")
+		return
+	}
+
 	pair, err := agent.GetSelectedCandidatePair()
 	if err != nil {
-		w.closeAgent(agent, w.agentDialerCancel)
+		w.closeAgent(agent, dialerCancel)
 		return
 	}
 	if pair == nil {
 		w.log.Warnf("selected candidate pair is nil, cannot proceed")
-		w.closeAgent(agent, w.agentDialerCancel)
+		w.closeAgent(agent, dialerCancel)
 		return
 	}
 
@@ -301,11 +327,27 @@ func (w *WorkerICE) connect(ctx context.Context, agent *icemaker.ThreadSafeAgent
 
 	w.log.Infof("connection succeeded with offer session: %s", remoteOfferAnswer.SessionIDString())
 	w.muxAgent.Lock()
+	// Authoritative ownership guard: a negotiation that lost w.agent to a newer
+	// one between the post-dial check and the commit must not clear agentConnecting,
+	// record lastSuccess or report the connection, so the state commit has to be
+	// atomic with the check.
+	if w.agent != agent {
+		w.muxAgent.Unlock()
+		if err := remoteConn.Close(); err != nil {
+			w.log.Warnf("failed to close stale ICE connection: %s", err)
+		}
+		w.log.Warnf("discarding connection from a stale ICE negotiation")
+		return
+	}
 	w.agentConnecting = false
 	w.lastSuccess = time.Now()
 	w.muxAgent.Unlock()
 
 	// todo: the potential problem is a race between the onConnectionStateChange
+	// and the delivery below: after this unlock, a newer offer can replace
+	// w.agent before onICEConnectionIsReady runs, delivering this (now stale)
+	// connection. The newer negotiation overwrites it with its own delivery,
+	// so the window only ever downgrades an endpoint transiently.
 	w.conn.onICEConnectionIsReady(selectedPriority(pair), ci)
 }
 
@@ -321,20 +363,32 @@ func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.C
 	sessionChanged := w.remoteSessionChanged
 	w.remoteSessionChanged = false
 
+	// Only the owner of the current session may reset its state: a stale dial
+	// goroutine waking after a newer attempt must not clobber it.
 	if w.agent == agent {
-		// consider to remove from here and move to the OnNewOffer
 		sessionID, err := NewICESessionID()
 		if err != nil {
 			w.log.Errorf("failed to create new session ID: %s", err)
 		}
 		w.sessionID = sessionID
-		w.agent = nil
-		w.agentConnecting = false
-		w.remoteSessionID = ""
+		w.abandonNegotiation()
 	}
 	return sessionChanged
 }
 
+// abandonNegotiation drops all recorded ICE session state so the worker treats the
+// next offer as a fresh start instead of a duplicate of a dead negotiation. The
+// agent and agentConnecting flags must change together: leaving one stale wedges
+// the reconnection guard into reporting Connected forever. It neither cancels an
+// in-flight dial nor closes an agent — callers dispose of those themselves first,
+// so a stale goroutine can never cancel another session's dial through this path.
+// Caller must hold muxAgent.
+func (w *WorkerICE) abandonNegotiation() {
+	w.agent = nil
+	w.agentConnecting = false
+	w.remoteSessionID = ""
+}
+
 func (w *WorkerICE) punchRemoteWGPort(pair *ice.CandidatePair, remoteWgPort int) {
 	// wait local endpoint configuration
 	time.Sleep(time.Second)

client/internal/peer/worker_ice_close_test.go

@@ -0,0 +1,257 @@
+package peer
+
+import (
+	"context"
+	"net"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+
+	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
+	signal "github.com/netbirdio/netbird/shared/signal/client"
+	sProto "github.com/netbirdio/netbird/shared/signal/proto"
+)
+
+// stubSignalClient satisfies signal.Client as a no-op so the candidate
+// goroutine spawned by a real GatherCandidates never dereferences a nil
+// signaler in tests.
+type stubSignalClient struct{}
+
+func (stubSignalClient) Close() error                                               { return nil }
+func (stubSignalClient) StreamConnected() bool                                      { return false }
+func (stubSignalClient) GetStatus() signal.Status                                   { return signal.StreamDisconnected }
+func (stubSignalClient) Receive(context.Context, func(*sProto.Message) error) error { return nil }
+func (stubSignalClient) Ready() bool                                                { return false }
+func (stubSignalClient) IsHealthy() bool                                            { return false }
+func (stubSignalClient) WaitStreamConnected(context.Context)                        {}
+func (stubSignalClient) SendToStream(*sProto.EncryptedMessage) error                { return nil }
+func (stubSignalClient) Send(*sProto.Message) error                                 { return nil }
+func (stubSignalClient) SetOnReconnectedListener(func())                            {}
+
+// newTestWorkerICE builds a worker with real pion plumbing and no-op signaling.
+func newTestWorkerICE(t *testing.T) *WorkerICE {
+	t.Helper()
+
+	config := connConf
+	stunTurn := &icemaker.StunTurn{}
+	stunTurn.Store(nil)
+	config.ICEConfig.StunTurn = stunTurn
+
+	w, err := NewWorkerICE(context.Background(), log.WithField("test", t.Name()), config, nil,
+		NewSignaler(stubSignalClient{}, wgtypes.Key{}), nil, nil, false)
+	require.NoError(t, err, "worker setup must succeed")
+	return w
+}
+
+// TestWorkerICE_CloseDuringDial_ClearsConnectingFlag drives the teardown race
+// through the real dial goroutine instead of simulating its cleanup.
+//
+// The real-world sequence this models:
+//  1. OnNewOffer starts a negotiation: agent set, agentConnecting = true,
+//     go connect()
+//  2. The network dies and connect() stays blocked inside GatherCandidates/Dial
+//  3. A WG handshake timeout calls Close(): the agent is released and the dial
+//     context cancelled, but agentConnecting is not reset
+//  4. The real goroutine wakes with an error and runs its own cleanup
+//     (closeAgent), where `w.agent == agent` is now false, so the flag reset
+//     is skipped
+//
+// There is no remote responder, so Dial can never succeed: whatever point the
+// goroutine is at, closing first forces it down the error path. Before the fix
+// the flag stays true forever and the deadline below expires.
+func TestWorkerICE_CloseDuringDial_ClearsConnectingFlag(t *testing.T) {
+	w := newTestWorkerICE(t)
+
+	sid := ICESessionID("test-session-id")
+	w.OnNewOffer(&OfferAnswer{
+		IceCredentials: IceCredentials{
+			UFrag: "testufrag",
+			Pwd:   "testpwdtestpwdtestpwd12",
+		},
+		SessionID: &sid,
+	})
+	require.True(t, w.InProgress(), "OnNewOffer must mark the negotiation as in progress")
+
+	// Teardown wins the race while connect() is still running.
+	w.Close()
+
+	// Close drops the flags synchronously, so the assertion below does not
+	// converge on the goroutine: the deadline only absorbs the dial goroutine
+	// waking up in the background, proving nothing re-wedges it afterwards.
+	require.Eventually(t, func() bool {
+		return !w.InProgress()
+	}, 10*time.Second, 50*time.Millisecond,
+		"Close must leave the negotiation idle even while the dial goroutine is still winding down")
+
+	// abandonNegotiation owns these three fields together; the worker is idle
+	// only when all of them are dropped.
+	w.muxAgent.Lock()
+	defer w.muxAgent.Unlock()
+	assert.Nil(t, w.agent, "no agent may survive the teardown")
+	assert.False(t, w.agentConnecting, "the connecting flag must match the nil agent")
+	assert.Empty(t, w.remoteSessionID, "a dead session's remote ID must not linger")
+}
+
+// TestWorkerICE_CloseClearsResidualConnectingState covers Close on a worker whose
+// agent is already gone but whose flag is stuck on true, e.g. after an aborted
+// recreate in OnNewOffer or after a first Close raced a dial goroutine.
+func TestWorkerICE_CloseClearsResidualConnectingState(t *testing.T) {
+	w := newTestWorkerICE(t)
+
+	w.muxAgent.Lock()
+	w.agentConnecting = true
+	w.muxAgent.Unlock()
+
+	w.Close()
+
+	assert.False(t, w.InProgress(), "Close must drop residual connecting state even without a live agent")
+
+	w.muxAgent.Lock()
+	defer w.muxAgent.Unlock()
+	assert.Nil(t, w.agent)
+	assert.False(t, w.agentConnecting)
+	assert.Empty(t, w.remoteSessionID)
+}
+
+// TestWorkerICE_StaleCloseAgentKeepsCurrentSession pins the ownership guard in
+// closeAgent: a late-waking dial goroutine from an older session must not reset
+// the state of a newer negotiation that reused the worker. The newer session
+// must survive wholesale - agent, flag and remote session identity alike.
+func TestWorkerICE_StaleCloseAgentKeepsCurrentSession(t *testing.T) {
+	w := newTestWorkerICE(t)
+	t.Cleanup(w.Close)
+
+	sidA := ICESessionID("session-a")
+	w.OnNewOffer(&OfferAnswer{
+		IceCredentials: IceCredentials{UFrag: "ufragaaaa", Pwd: "pwdpwdpwdpwdpwdpwdpwdp1"},
+		SessionID:      &sidA,
+	})
+	w.muxAgent.Lock()
+	oldAgent := w.agent
+	oldCancel := w.agentDialerCancel
+	w.muxAgent.Unlock()
+	require.NotNil(t, oldAgent, "OnNewOffer must have created an ICE agent")
+
+	w.Close()
+
+	sidB := ICESessionID("session-b")
+	w.OnNewOffer(&OfferAnswer{
+		IceCredentials: IceCredentials{UFrag: "ufragbbbb", Pwd: "pwdpwdpwdpwdpwdpwdpwdp2"},
+		SessionID:      &sidB,
+	})
+	require.True(t, w.InProgress(), "the second negotiation must be in flight")
+
+	w.muxAgent.Lock()
+	newAgent := w.agent
+	w.muxAgent.Unlock()
+
+	// The old dial goroutine finally wakes and cleans up its captured agent.
+	w.closeAgent(oldAgent, oldCancel)
+
+	w.muxAgent.Lock()
+	defer w.muxAgent.Unlock()
+	assert.Same(t, newAgent, w.agent, "the current agent must be untouched by the stale cleanup")
+	assert.True(t, w.agentConnecting, "the current negotiation must stay in flight")
+	// Read live under the lock: a snapshot captured before the stale cleanup
+	// would pass even if the cleanup wiped current state.
+	assert.Equal(t, sidB, w.remoteSessionID, "the remote session identity must be preserved")
+}
+
+// closeTrackConn records Close calls so a test can assert that a discarded
+// connection was actually released.
+type closeTrackConn struct {
+	net.Conn
+	closed atomic.Bool
+}
+
+func (c *closeTrackConn) Close() error {
+	c.closed.Store(true)
+	return c.Conn.Close()
+}
+
+// TestWorkerICE_StaleDialSuccessKeepsNewerNegotiation pins the ownership guard
+// in connect()'s success path: a dial that came back after a newer negotiation
+// replaced the agent must discard its connection and leave the newer session's
+// state - agent, agentConnecting, remoteSessionID, lastSuccess - intact.
+//
+// The dial hook holds session A's goroutine open until session B is installed,
+// then returns a live connection, mimicking the vendored pion dial which hands
+// out a live *ice.Conn when a pair is selected without checking afterwards
+// whether the agent was replaced meanwhile. Releasing A's dial therefore
+// exercises the stale-success commit path deterministically instead of racing
+// real ICE.
+func TestWorkerICE_StaleDialSuccessKeepsNewerNegotiation(t *testing.T) {
+	w := newTestWorkerICE(t)
+	t.Cleanup(w.Close)
+
+	dialStarted := make(chan struct{})
+	releaseDial := make(chan struct{})
+	staleConn := &closeTrackConn{}
+
+	var calls atomic.Int32
+	w.dialFunc = func(ctx context.Context, _ *icemaker.ThreadSafeAgent, _ *OfferAnswer) (net.Conn, error) {
+		if calls.Add(1) == 1 {
+			// Session A: hold the goroutine open until session B is installed,
+			// then return a live connection, mimicking the vendored pion dial
+			// which hands out a live *ice.Conn once a pair is selected without
+			// re-checking whether the agent was replaced meanwhile. Releasing
+			// the dial therefore exercises the stale-success commit path
+			// deterministically instead of racing real ICE.
+			close(dialStarted)
+			<-releaseDial
+			client, _ := net.Pipe()
+			staleConn.Conn = client
+			return staleConn, nil
+		}
+		// A newer negotiation parks on its dialer context, cancelled by the
+		// t.Cleanup Close at test end.
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+
+	sidA := ICESessionID("session-a")
+	w.OnNewOffer(&OfferAnswer{
+		IceCredentials: IceCredentials{UFrag: "ufragaaaa", Pwd: "pwdpwdpwdpwdpwdpwdpwdp1"},
+		SessionID:      &sidA,
+	})
+	require.True(t, w.InProgress(), "session A must be in flight")
+
+	// Session A's goroutine is now parked in the dial hook.
+	<-dialStarted
+
+	sidB := ICESessionID("session-b")
+	w.OnNewOffer(&OfferAnswer{
+		IceCredentials: IceCredentials{UFrag: "ufragbbbb", Pwd: "pwdpwdpwdpwdpwdpwdpwdp2"},
+		SessionID:      &sidB,
+	})
+
+	w.muxAgent.Lock()
+	agentB := w.agent
+	w.lastSuccess = time.Time{}
+	w.muxAgent.Unlock()
+	require.NotNil(t, agentB, "session B must have created an ICE agent")
+	require.True(t, w.InProgress(), "session B must be in flight")
+
+	// Release session A's dial: it must be recognized as stale and discarded.
+	close(releaseDial)
+	require.Eventually(t, func() bool {
+		return staleConn.closed.Load()
+	}, 10*time.Second, 10*time.Millisecond,
+		"the stale connection must be closed by the ownership guard")
+
+	w.muxAgent.Lock()
+	defer w.muxAgent.Unlock()
+	assert.Same(t, agentB, w.agent, "session A must not uninstall session B's agent")
+	assert.True(t, w.agentConnecting, "session A must not clear session B's connecting flag")
+	assert.Equal(t, sidB, w.remoteSessionID, "session A must not clear session B's remote session identity")
+	assert.True(t, w.lastSuccess.IsZero(), "session A must not record a success for session B")
+	// The commit block guards agentConnecting, lastSuccess and
+	// onICEConnectionIsReady together, so the state assertions above imply the
+	// callback never ran for session A; the nil conn would have panicked the
+	// stale goroutine on any invocation.
+}