diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index b508af630..56e6afb2b 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -242,18 +242,20 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error { conn.metricsStages = &metricsstages.MetricsStages{} conn.ctx, conn.ctxCancel = context.WithCancel(engineCtx) + mb := newMailbox() conn.workerRelay = worker.NewWorkerRelay(conn.Log, conn.config.Key, conn.config.IsController(), conn.onRelayConnectionIsReady, conn.onRelayDisconnected, conn.relayManager) if !IsForceRelayed() { relayIsSupportedLocally := conn.workerRelay.RelayIsSupportedLocally() - workerICE, err := worker.NewICE(conn.Log, conn.config.Key, conn.config.ICEConfig, conn.config.IsController(), conn.onICEConnectionIsReady, conn.onICEStateDisconnected, worker.ICEDependencies{ + workerICE, err := worker.NewICE(conn.Log, conn.config.Key, conn.config.ICEConfig, conn.config.IsController(), mb.post, worker.ICEDependencies{ Signaler: conn.signaler, IFaceDiscover: conn.iFaceDiscover, StatusRecorder: conn.statusRecorder, PortForwardManager: conn.portForwardManager, }, relayIsSupportedLocally) if err != nil { + conn.ctxCancel() return err } conn.workerICE = workerICE @@ -288,7 +290,6 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error { conn.Log.Warnf("error while updating the state err: %v", err) } - mb := newMailbox() conn.loopDone = make(chan struct{}) conn.mailbox.Store(mb) @@ -446,10 +447,16 @@ func (conn *Conn) handleEvent(ev event) { conn.handleRemoteAnswer(&e.answer) case evRemoteCandidate: conn.handleRemoteCandidate(e) - case evICEReady: - conn.handleICEReady(e.priority, e.info) - case evICEDown: - conn.handleICEDisconnected(e.sessionChanged) + case worker.ICEDialDone: + conn.handleICEDialDone(e) + case worker.ICEStateChanged: + if disconnected, sessionChanged := conn.workerICE.OnConnectionStateChange(e); disconnected { + conn.handleICEDisconnected(sessionChanged) + } + case worker.ICECandidate: + conn.workerICE.OnLocalCandidate(e) + case worker.ICESelectedPair: + conn.workerICE.OnSelectedCandidatePair(e) case evRelayReady: conn.handleRelayReady(e.info) case evRelayDown: @@ -533,6 +540,12 @@ func (conn *Conn) teardown(mb *mailbox, leftover []event, signalToRemote bool, d func (conn *Conn) releaseEvents(evs []event) { for _, ev := range evs { switch e := ev.(type) { + case worker.ICEDialDone: + if e.Conn != nil { + if err := e.Conn.Close(); err != nil { + conn.Log.Debugf("close unused ICE connection: %v", err) + } + } case evRelayReady: if err := e.info.RelayedConn.Close(); err != nil { conn.Log.Warnf("failed to close unnecessary relayed connection: %v", err) @@ -623,6 +636,16 @@ func (conn *Conn) handleGuardTick() { }() } +func (conn *Conn) handleICEDialDone(e worker.ICEDialDone) { + if conn.ctx.Err() != nil { + conn.releaseEvents([]event{e}) + return + } + if priority, info, ready := conn.workerICE.OnDialDone(e); ready { + conn.handleICEReady(priority, info) + } +} + // handleICEReady starts proxying traffic from/to local WireGuard and sets the // connection status to StatusConnected. func (conn *Conn) handleICEReady(priority worker.ConnPriority, iceConnInfo worker.ICEConnInfo) { @@ -927,14 +950,6 @@ func (conn *Conn) handleWGCheckSuccess() { conn.wgTimeouts = 0 } -func (conn *Conn) onICEConnectionIsReady(priority worker.ConnPriority, iceConnInfo worker.ICEConnInfo) { - conn.post(evICEReady{priority: priority, info: iceConnInfo}) -} - -func (conn *Conn) onICEStateDisconnected(sessionChanged bool) { - conn.post(evICEDown{sessionChanged: sessionChanged}) -} - // onRelayConnectionIsReady closes the relayed connection when the event loop // is gone and nobody will take ownership of it. func (conn *Conn) onRelayConnectionIsReady(rci worker.RelayConnInfo) { diff --git a/client/internal/peer/conn_ice_events_test.go b/client/internal/peer/conn_ice_events_test.go new file mode 100644 index 000000000..f6330bca7 --- /dev/null +++ b/client/internal/peer/conn_ice_events_test.go @@ -0,0 +1,50 @@ +package peer + +import ( + "context" + "net" + "testing" + + log "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/netbirdio/netbird/client/internal/peer/worker" +) + +type iceEventConn struct { + net.Conn + closed bool +} + +func (c *iceEventConn) Close() error { + c.closed = true + return c.Conn.Close() +} + +func TestConn_DiscardedICEDialResultClosesConnection(t *testing.T) { + for _, reason := range []string{"cancelled conn", "shutdown queue", "shutdown batch"} { + t.Run(reason, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + t.Cleanup(cancel) + conn := &Conn{ctx: ctx, Log: log.WithField("test", t.Name())} + client, server := net.Pipe() + t.Cleanup(func() { _ = client.Close(); _ = server.Close() }) + remote := &iceEventConn{Conn: client} + e := worker.ICEDialDone{Conn: remote} + mb := newMailbox() + switch reason { + case "cancelled conn": + cancel() + conn.handleEvent(e) + case "shutdown queue": + require.True(t, mb.post(e), "the result must enter the queue before shutdown") + conn.releaseEvents(mb.closeAndDrain()) + case "shutdown batch": + require.True(t, mb.post(e), "the result must enter the batch before shutdown") + conn.releaseEvents(mb.drain()) + } + assert.True(t, remote.closed, "discarded results must release their connections") + }) + } +} diff --git a/client/internal/peer/event.go b/client/internal/peer/event.go index c77838ff4..925ee42e8 100644 --- a/client/internal/peer/event.go +++ b/client/internal/peer/event.go @@ -14,13 +14,10 @@ import ( // event is a message processed by the Conn event loop. All mutable Conn state // is owned by that loop; producers deliver events through the mailbox and // never mutate Conn state directly. -type event any +type event = any -// staleableEvent is implemented by events tied to the lifetime of a transport -// component (WG watcher, ICE agent, relay connection). Each such component runs -// under its own context, cancelled when the component is superseded; an event -// carrying a cancelled context is dropped at dispatch time. A cancel performed -// by an earlier event in the same drained batch already suppresses it. +// staleableEvent is implemented by events whose context becoming cancelled +// makes them irrelevant, such as a timeout from an old WG watcher. type staleableEvent interface { isStale() bool } @@ -45,15 +42,6 @@ type evRemoteCandidate struct { haRoutes route.HAMap } -type evICEReady struct { - priority worker.ConnPriority - info worker.ICEConnInfo -} - -type evICEDown struct { - sessionChanged bool -} - type evRelayReady struct { info worker.RelayConnInfo } diff --git a/client/internal/peer/mailbox_test.go b/client/internal/peer/mailbox_test.go index 20e509886..d5f8bc734 100644 --- a/client/internal/peer/mailbox_test.go +++ b/client/internal/peer/mailbox_test.go @@ -4,6 +4,7 @@ import ( "testing" "github.com/netbirdio/netbird/client/internal/peer/signaling" + "github.com/netbirdio/netbird/client/internal/peer/worker" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -70,7 +71,7 @@ func TestMailbox_DrainOrder(t *testing.T) { require.True(t, mb.post(evRemoteAnswer{answer: signaling.OfferAnswer{}})) require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{}})) require.True(t, mb.post(evRelayDown{})) - require.True(t, mb.post(evICEDown{sessionChanged: true})) + require.True(t, mb.post(worker.ICEStateChanged{})) require.True(t, mb.post(evClose{})) evs := mb.drain() @@ -80,7 +81,7 @@ func TestMailbox_DrainOrder(t *testing.T) { assert.True(t, ok, "lifecycle events must come first") _, ok = evs[1].(evRelayDown) assert.True(t, ok, "transport events must keep FIFO order") - _, ok = evs[2].(evICEDown) + _, ok = evs[2].(worker.ICEStateChanged) assert.True(t, ok, "transport events must keep FIFO order") _, ok = evs[3].(evRemoteOffer) assert.True(t, ok, "offer must come after transport events") diff --git a/client/internal/peer/worker/event.go b/client/internal/peer/worker/event.go new file mode 100644 index 000000000..51509fca4 --- /dev/null +++ b/client/internal/peer/worker/event.go @@ -0,0 +1,36 @@ +package worker + +import ( + "net" + + "github.com/pion/ice/v4" + + icemaker "github.com/netbirdio/netbird/client/internal/peer/ice" + "github.com/netbirdio/netbird/client/internal/peer/signaling" +) + +// ICEStateChanged carries a Pion notification to the Conn event loop. +type ICEStateChanged struct { + Agent *icemaker.ThreadSafeAgent + State ice.ConnectionState +} + +// ICECandidate carries a gathered candidate, or nil when gathering finishes. +type ICECandidate struct { + Candidate ice.Candidate +} + +// ICESelectedPair carries Pion's selected candidate pair notification. +type ICESelectedPair struct { + Agent *icemaker.ThreadSafeAgent + Local, Remote ice.Candidate +} + +// ICEDialDone transfers the dial result to the event loop. The producer closes +// Conn if posting fails; otherwise the consumer owns it, including on errors. +type ICEDialDone struct { + Agent *icemaker.ThreadSafeAgent + Conn net.Conn + Offer signaling.OfferAnswer + Err error +} diff --git a/client/internal/peer/worker/worker_ice.go b/client/internal/peer/worker/worker_ice.go index 39e73cbbd..324bc735d 100644 --- a/client/internal/peer/worker/worker_ice.go +++ b/client/internal/peer/worker/worker_ice.go @@ -5,7 +5,7 @@ import ( "fmt" "net" "strconv" - "sync" + "sync/atomic" "time" "github.com/pion/ice/v4" @@ -40,13 +40,17 @@ type ICEDependencies struct { PortForwardManager *portforward.Manager } +type iceDialFunc func(context.Context, *icemaker.ThreadSafeAgent, *signaling.OfferAnswer) (net.Conn, error) + +// ICE is owned by the Conn event loop. Pion callbacks and the dial goroutine +// only post events. Credentials and InProgress expose atomic snapshots to +// signaling and the reconnection guard. type ICE struct { log *log.Entry key string iceConfig icemaker.Config isController bool - onConnReady func(priority ConnPriority, iceConnInfo ICEConnInfo) - onStatusDisconnect func(sessionChanged bool) + postEvent func(any) bool signaler *signaling.Signaler iFaceDiscover stdnet.ExternalIFaceDiscover statusRecorder *status.Recorder @@ -55,66 +59,53 @@ type ICE struct { agent *icemaker.ThreadSafeAgent agentDialerCancel context.CancelFunc - agentConnecting bool // while it is true, drop all incoming offers - lastSuccess time.Time // with this avoid the too frequent ICE agent recreation - // connectedAgent is the agent whose connection was last reported ready; guarded by muxAgent + agentConnecting atomic.Bool + // connectedAgent is the agent whose connection was last reported ready. connectedAgent *icemaker.ThreadSafeAgent // remoteSessionID represents the peer's session identifier from the latest remote offer. - remoteSessionID icemaker.SessionID - // sessionID is used to track the current session ID of the ICE agent - // increase by one when disconnecting the agent - // with it the remote peer can discard the already deprecated offer/answer - // Without it the remote peer may recreate a workable ICE connection - sessionID icemaker.SessionID + remoteSessionID icemaker.SessionID remoteSessionChanged bool - muxAgent sync.Mutex - - localUfrag string - localPwd string + credentials atomic.Pointer[signaling.Credentials] // 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 *signaling.OfferAnswer) (net.Conn, error) + // Captured before starting connect; only tests replace the dial operation. + dialFunc iceDialFunc } -func NewICE(log *log.Entry, key string, iceConfig icemaker.Config, isController bool, onConnReady func(ConnPriority, ICEConnInfo), onStatusDisconnect func(bool), services ICEDependencies, hasRelayOnLocally bool) (*ICE, error) { +// NewICE creates an event-loop-owned worker publishing to the current Open's mailbox. +func NewICE(log *log.Entry, key string, iceConfig icemaker.Config, isController bool, postEvent func(any) bool, services ICEDependencies, hasRelayOnLocally bool) (*ICE, error) { sessionID, err := icemaker.NewSessionID() if err != nil { return nil, err } + localUfrag, localPwd, err := icemaker.GenerateICECredentials() + if err != nil { + return nil, err + } w := &ICE{ log: log, key: key, iceConfig: iceConfig, isController: isController, - onConnReady: onConnReady, - onStatusDisconnect: onStatusDisconnect, + postEvent: postEvent, signaler: services.Signaler, iFaceDiscover: services.IFaceDiscover, statusRecorder: services.StatusRecorder, portForwardManager: services.PortForwardManager, hasRelayOnLocally: hasRelayOnLocally, - sessionID: sessionID, } - - localUfrag, localPwd, err := icemaker.GenerateICECredentials() - if err != nil { - return nil, err - } - w.localUfrag = localUfrag - w.localPwd = localPwd + w.credentials.Store(&signaling.Credentials{UFrag: localUfrag, Pwd: localPwd, SessionID: sessionID}) return w, nil } +// OnNewOffer starts a negotiation on the event loop. func (w *ICE) OnNewOffer(ctx context.Context, remoteOfferAnswer *signaling.OfferAnswer) { w.log.Debugf("OnNewOffer for ICE, serial: %s", remoteOfferAnswer.SessionIDString()) - w.muxAgent.Lock() - defer w.muxAgent.Unlock() - if w.agent != nil || w.agentConnecting { + if w.agent != nil || w.agentConnecting.Load() { // backward compatibility with old clients that do not send session ID if remoteOfferAnswer.SessionID == nil { w.log.Debugf("agent already exists, skipping the offer") @@ -137,7 +128,9 @@ func (w *ICE) OnNewOffer(ctx context.Context, remoteOfferAnswer *signaling.Offer if err != nil { w.log.Errorf("failed to create new session ID: %s", err) } - w.sessionID = sessionID + creds := w.Credentials() + creds.SessionID = sessionID + w.credentials.Store(&creds) w.abandonNegotiation() } @@ -149,32 +142,33 @@ func (w *ICE) OnNewOffer(ctx context.Context, remoteOfferAnswer *signaling.Offer } if remoteOfferAnswer.SessionID != nil { - w.log.Debugf("recreate ICE agent: %s / %s", w.sessionID, *remoteOfferAnswer.SessionID) + w.log.Debugf("recreate ICE agent: %s / %s", w.Credentials().SessionID, *remoteOfferAnswer.SessionID) } dialerCtx, dialerCancel := context.WithCancel(ctx) - agent, err := w.reCreateAgent(ctx, dialerCancel, preferredCandidateTypes) + agent, err := w.reCreateAgent(ctx, preferredCandidateTypes) if err != nil { + dialerCancel() w.log.Errorf("failed to recreate ICE Agent: %s", err) return } w.agent = agent w.agentDialerCancel = dialerCancel - w.agentConnecting = true + w.agentConnecting.Store(true) if remoteOfferAnswer.SessionID != nil { w.remoteSessionID = *remoteOfferAnswer.SessionID } else { w.remoteSessionID = "" } - // 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) + dial := w.dialFunc + if dial == nil { + dial = w.agentDial + } + go w.connect(dialerCtx, agent, *remoteOfferAnswer, dial) } // OnRemoteCandidate Handles ICE connection Candidate provided by the remote peer. func (w *ICE) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HAMap) { - w.muxAgent.Lock() - defer w.muxAgent.Unlock() w.log.Debugf("OnRemoteCandidate from peer %s -> %s", w.key, candidate.String()) if w.agent == nil { w.log.Warnf("ICE Agent is not initialized yet") @@ -202,136 +196,113 @@ func (w *ICE) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HAMap) { } } +// Credentials returns a consistent snapshot for asynchronous signaling. func (w *ICE) Credentials() signaling.Credentials { - w.muxAgent.Lock() - defer w.muxAgent.Unlock() - return signaling.Credentials{ - UFrag: w.localUfrag, - Pwd: w.localPwd, - SessionID: w.sessionID, - } + return *w.credentials.Load() } +// InProgress returns the negotiation state published by the event loop. func (w *ICE) InProgress() bool { - w.muxAgent.Lock() - defer w.muxAgent.Unlock() - - return w.agentConnecting + return w.agentConnecting.Load() } +// Close releases the current agent on the event loop. Repeated calls are harmless. func (w *ICE) Close() { - w.muxAgent.Lock() - defer w.muxAgent.Unlock() - if w.agent != nil { w.agentDialerCancel() if err := w.agent.Close(); err != nil { w.log.Warnf("failed to close ICE agent: %s", err) } } - // 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. + // A later dial result no longer owns the agent, so Close must clear the + // connecting state before the reconnection guard reads it. w.abandonNegotiation() } -func (w *ICE) reCreateAgent(ctx context.Context, dialerCancel context.CancelFunc, candidates []ice.CandidateType) (*icemaker.ThreadSafeAgent, error) { +func (w *ICE) reCreateAgent(ctx context.Context, candidates []ice.CandidateType) (*icemaker.ThreadSafeAgent, error) { w.portForwardAttempted = false - - agent, err := icemaker.NewAgent(ctx, w.iFaceDiscover, w.iceConfig, candidates, w.localUfrag, w.localPwd) + creds := w.Credentials() + agent, err := icemaker.NewAgent(ctx, w.iFaceDiscover, w.iceConfig, candidates, creds.UFrag, creds.Pwd) if err != nil { return nil, fmt.Errorf("create agent: %w", err) } + configured := false + defer func() { + if !configured { + if err := agent.Close(); err != nil { + w.log.Warnf("failed to close unconfigured ICE agent: %s", err) + } + } + }() - if err := agent.OnCandidate(w.onICECandidate); err != nil { - return nil, err - } - - if err := agent.OnConnectionStateChange(w.onConnectionStateChange(agent, dialerCancel)); err != nil { - return nil, err - } - - if err := agent.OnSelectedCandidatePairChange(func(c1, c2 ice.Candidate) { - w.onICESelectedCandidatePair(agent, c1, c2) + post := w.postEvent + if err := agent.OnCandidate(func(candidate ice.Candidate) { + post(ICECandidate{Candidate: candidate}) }); err != nil { return nil, err } - + if err := agent.OnConnectionStateChange(func(state ice.ConnectionState) { + post(ICEStateChanged{Agent: agent, State: state}) + }); err != nil { + return nil, err + } + if err := agent.OnSelectedCandidatePairChange(func(local, remote ice.Candidate) { + post(ICESelectedPair{Agent: agent, Local: local, Remote: remote}) + }); err != nil { + return nil, err + } + configured = true return agent, nil } -func (w *ICE) getSessionID() icemaker.SessionID { - w.muxAgent.Lock() - defer w.muxAgent.Unlock() - - return w.sessionID +// connect performs blocking ICE I/O without reading or changing negotiation state. +func (w *ICE) connect(ctx context.Context, agent *icemaker.ThreadSafeAgent, offer signaling.OfferAnswer, dial iceDialFunc) { + result := ICEDialDone{Agent: agent, Offer: offer} + result.Err = agent.GatherCandidates() + if result.Err == nil { + result.Conn, result.Err = dial(ctx, agent, &offer) + } + if !w.postEvent(result) { + w.closeUnusedConn(result.Conn) + } } -// 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 *ICE) connect(ctx context.Context, dialerCancel context.CancelFunc, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *signaling.OfferAnswer) { - w.log.Debugf("gather candidates") - if err := agent.GatherCandidates(); err != nil { - w.log.Warnf("failed to gather candidates: %s", err) - w.closeAgent(agent, dialerCancel) - return +// OnDialDone consumes the dial result on the event loop. Rejected results are +// released here; accepted results transfer their connection to the caller. +func (w *ICE) OnDialDone(e ICEDialDone) (ConnPriority, ICEConnInfo, bool) { + if e.Err != nil { + w.log.Debugf("ICE dial did not establish a connection: %v", e.Err) + w.closeUnusedConn(e.Conn) + w.closeAgent(e.Agent) + return None, ICEConnInfo{}, false } - w.log.Debugf("agent dial") - dial := func(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *signaling.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, dialerCancel) - return - } - w.log.Debugf("agent dial succeeded") - // A newer negotiation may have replaced this agent during the dial. - // Discard its connection before querying candidates or punching ports. - 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) - } + if w.agent != e.Agent { + w.closeUnusedConn(e.Conn) w.log.Warnf("discarding connection from a stale ICE negotiation") - return + return None, ICEConnInfo{}, false } - pair, err := agent.GetSelectedCandidatePair() - if err != nil { - w.closeAgent(agent, dialerCancel) - return - } - if pair == nil { - w.log.Warnf("selected candidate pair is nil, cannot proceed") - w.closeAgent(agent, dialerCancel) - return + pair, err := e.Agent.GetSelectedCandidatePair() + if err != nil || pair == nil { + w.log.Debugf("ICE dial has no selected candidate pair: %v", err) + w.closeUnusedConn(e.Conn) + w.closeAgent(e.Agent) + return None, ICEConnInfo{}, false } if !isRelayCandidate(pair.Local) { - // dynamically set remote WireGuard port if other side specified a different one from the default one remoteWgPort := iface.DefaultWgPort - if remoteOfferAnswer.WgListenPort != 0 { - remoteWgPort = remoteOfferAnswer.WgListenPort + if e.Offer.WgListenPort != 0 { + remoteWgPort = e.Offer.WgListenPort } - - // To support old version's with direct mode we attempt to punch an additional role with the remote WireGuard port go w.punchRemoteWGPort(pair, remoteWgPort) } - - ci := ICEConnInfo{ - RemoteConn: remoteConn, - RosenpassPubKey: remoteOfferAnswer.RosenpassPubKey, - RosenpassAddr: remoteOfferAnswer.RosenpassAddr, + info := ICEConnInfo{ + RemoteConn: e.Conn, + RosenpassPubKey: e.Offer.RosenpassPubKey, + RosenpassAddr: e.Offer.RosenpassAddr, LocalIceCandidateType: pair.Local.Type().String(), RemoteIceCandidateType: pair.Remote.Type().String(), LocalIceCandidateEndpoint: net.JoinHostPort(pair.Local.Address(), strconv.Itoa(pair.Local.Port())), @@ -339,58 +310,52 @@ func (w *ICE) connect(ctx context.Context, dialerCancel context.CancelFunc, agen Relayed: isRelayed(pair), RelayedOnLocal: isRelayCandidate(pair.Local), } - w.log.Debugf("on ICE conn is ready to use") - - w.muxAgent.Lock() - // Keep the ownership check atomic with the state update so a stale dial - // cannot overwrite a newer negotiation. - 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.connectedAgent = agent - w.muxAgent.Unlock() - - w.log.Infof("connection succeeded with offer session: %s", remoteOfferAnswer.SessionIDString()) - w.onConnReady(selectedPriority(pair), ci) + w.agentConnecting.Store(false) + w.connectedAgent = e.Agent + w.log.Infof("connection succeeded with offer session: %s", e.Offer.SessionIDString()) + return selectedPriority(pair), info, true } -func (w *ICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.CancelFunc) bool { - cancel() +func (w *ICE) closeUnusedConn(conn net.Conn) { + if conn != nil { + if err := conn.Close(); err != nil { + w.log.Debugf("close unused ICE connection: %v", err) + } + } +} + +func (w *ICE) closeAgent(agent *icemaker.ThreadSafeAgent) bool { + // Superseded agents had their dial context cancelled when replaced. + if w.agent == agent { + w.agentDialerCancel() + } if err := agent.Close(); err != nil { w.log.Warnf("failed to close ICE agent: %s", err) } - w.muxAgent.Lock() - defer w.muxAgent.Unlock() - 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. + // Only the owner of the current session may reset its state. if w.agent == agent { sessionID, err := icemaker.NewSessionID() if err != nil { w.log.Errorf("failed to create new session ID: %s", err) } - w.sessionID = sessionID + creds := w.Credentials() + creds.SessionID = sessionID + w.credentials.Store(&creds) w.abandonNegotiation() } return sessionChanged } // Clearing the agent and connecting flag together keeps retries from stalling. -// Callers must dispose of the agent first and hold muxAgent. +// Callers run on the event loop and must dispose of the agent first. func (w *ICE) abandonNegotiation() { w.agent = nil - w.agentConnecting = false + w.agentDialerCancel = nil + w.agentConnecting.Store(false) w.remoteSessionID = "" } @@ -414,9 +379,9 @@ func (w *ICE) punchRemoteWGPort(pair *ice.CandidatePair, remoteWgPort int) { } } -// onICECandidate is a callback attached to an ICE Agent to receive new local connection candidates -// and then signals them to the remote peer -func (w *ICE) onICECandidate(candidate ice.Candidate) { +// OnLocalCandidate signals a gathered candidate from the event loop. +func (w *ICE) OnLocalCandidate(e ICECandidate) { + candidate := e.Candidate // nil means candidate gathering has been ended if candidate == nil { return @@ -459,13 +424,10 @@ func (w *ICE) injectPortForwardedCandidate(srflxCandidate ice.Candidate) { return } - w.muxAgent.Lock() if w.portForwardAttempted { - w.muxAgent.Unlock() return } w.portForwardAttempted = true - w.muxAgent.Unlock() forwardedCandidate, err := w.createForwardedCandidate(srflxCandidate, mapping) if err != nil { @@ -530,7 +492,9 @@ func (w *ICE) createForwardedCandidate(srflxCandidate ice.Candidate, mapping *po return candidate, nil } -func (w *ICE) onICESelectedCandidatePair(agent *icemaker.ThreadSafeAgent, c1, c2 ice.Candidate) { +// OnSelectedCandidatePair records the selected pair on the event loop. +func (w *ICE) OnSelectedCandidatePair(e ICESelectedPair) { + agent, c1, c2 := e.Agent, e.Local, e.Remote w.log.Debugf("selected candidate pair [local <-> remote] -> [%s <-> %s], peer %s", c1.String(), c2.String(), w.key) @@ -548,7 +512,7 @@ func (w *ICE) onICESelectedCandidatePair(agent *icemaker.ThreadSafeAgent, c1, c2 } func (w *ICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) { - sessionID := w.getSessionID() + sessionID := w.Credentials().SessionID stats := agent.GetCandidatePairsStats() localCandidates, _ := agent.GetLocalCandidates() remoteCandidates, _ := agent.GetRemoteCandidates() @@ -578,48 +542,34 @@ func (w *ICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) { } } -func (w *ICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dialerCancel context.CancelFunc) func(ice.ConnectionState) { - // per-agent state; pion delivers callbacks of one agent sequentially - var connected bool - return func(state ice.ConnectionState) { - w.log.Debugf("ICE ConnectionState has changed to %s", state.String()) - switch state { - case ice.ConnectionStateConnected: - connected = true - w.logSuccessfulPaths(agent) - case ice.ConnectionStateFailed, ice.ConnectionStateDisconnected, ice.ConnectionStateClosed: - // ice.ConnectionStateClosed happens when we recreate the agent. The P2P to relay switch requires - // notifying conn.onICEStateDisconnected so it can update the currently used priority. - - sessionChanged := w.closeAgent(agent, dialerCancel) - - if !connected { - return - } - connected = false - - w.muxAgent.Lock() - stale := w.connectedAgent != agent - if !stale { - w.connectedAgent = nil - } - w.muxAgent.Unlock() - - if stale { - w.log.Debugf("suppress disconnected event of replaced ICE agent") - return - } - w.onStatusDisconnect(sessionChanged) +// OnConnectionStateChange processes Pion state on the event loop and reports +// whether the last ready connection disconnected and the remote session changed. +func (w *ICE) OnConnectionStateChange(e ICEStateChanged) (disconnected, sessionChanged bool) { + w.log.Debugf("ICE ConnectionState has changed to %s", e.State.String()) + switch e.State { + case ice.ConnectionStateConnected: + w.logSuccessfulPaths(e.Agent) + case ice.ConnectionStateFailed, ice.ConnectionStateDisconnected, ice.ConnectionStateClosed: + sessionChanged = w.closeAgent(e.Agent) + if w.connectedAgent != e.Agent { + return false, sessionChanged } + w.connectedAgent = nil + return true, sessionChanged } + return false, false } -func (w *ICE) agentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *signaling.OfferAnswer) (*ice.Conn, error) { +func (w *ICE) agentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *signaling.OfferAnswer) (net.Conn, error) { + dial := agent.Accept if w.isController { - return agent.Dial(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd) - } else { - return agent.Accept(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd) + dial = agent.Dial } + conn, err := dial(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd) + if err != nil { + return nil, err + } + return conn, nil } func shouldAddExtraCandidate(candidate ice.Candidate) bool { diff --git a/client/internal/peer/worker/worker_ice_close_test.go b/client/internal/peer/worker/worker_ice_close_test.go index 3175e22b9..85b7b4703 100644 --- a/client/internal/peer/worker/worker_ice_close_test.go +++ b/client/internal/peer/worker/worker_ice_close_test.go @@ -7,6 +7,7 @@ import ( "testing" "time" + "github.com/pion/ice/v4" log "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -34,137 +35,135 @@ func (stubSignalClient) SendToStream(*sProto.EncryptedMessage) error 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) *ICE { +// newTestWorkerICE collects raw events without applying them. Tests drive the +// worker from their own goroutine, just as the Conn event loop does. +func newTestWorkerICE(t *testing.T) (*ICE, <-chan any) { t.Helper() - - config := icemaker.Config{} stunTurn := &icemaker.StunTurn{} stunTurn.Store(nil) - config.StunTurn = stunTurn - - w, err := NewICE(log.WithField("test", t.Name()), "test-peer", config, true, nil, nil, + events := make(chan any, 256) + post := func(e any) bool { + select { + case events <- e: + return true + default: + t.Errorf("unexpected ICE event queue overflow") + return false + } + } + w, err := NewICE(log.WithField("test", t.Name()), "test-peer", icemaker.Config{StunTurn: stunTurn}, true, post, ICEDependencies{Signaler: signaling.NewSignaler(stubSignalClient{}, wgtypes.Key{})}, false) - require.NoError(t, err, "worker setup must succeed") - return w + require.NoError(t, err) + t.Cleanup(func() { w.Close() }) + return w, events +} + +func testICEOffer(id icemaker.SessionID) *signaling.OfferAnswer { + return &signaling.OfferAnswer{ + IceCredentials: signaling.IceCredentials{UFrag: "testufrag", Pwd: "testpwdtestpwdtestpwd12"}, + SessionID: &id, + } +} + +func waitICEDialDone(t *testing.T, events <-chan any, agent *icemaker.ThreadSafeAgent) ICEDialDone { + t.Helper() + timer := time.NewTimer(10 * time.Second) + defer timer.Stop() + for { + select { + case ev := <-events: + if e, ok := ev.(ICEDialDone); ok { + if e.Agent == agent { + return e + } + if e.Conn != nil { + _ = e.Conn.Close() + } + } + case <-timer.C: + t.Fatal("dial must publish a result") + return ICEDialDone{} + } + } } -// 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 := icemaker.SessionID("test-session-id") - w.OnNewOffer(t.Context(), &signaling.OfferAnswer{ - IceCredentials: signaling.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, events := newTestWorkerICE(t) + w.OnNewOffer(t.Context(), testICEOffer("a")) + agent := w.agent + require.True(t, w.InProgress(), "the negotiation must be in progress") 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") + // Close clears the state immediately; late dial completion is only data + // until consumed by the loop, and must not restore the closed agent. + assert.False(t, w.InProgress(), "Close must clear the connecting snapshot") + e := waitICEDialDone(t, events, agent) + _, _, ready := w.OnDialDone(e) + assert.False(t, ready, "a closed agent's result must be rejected") + assert.Nil(t, w.agent, "the agent must remain released") + assert.Empty(t, w.remoteSessionID, "the remote session must remain cleared") } -// 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, _ := newTestWorkerICE(t) + w.agentConnecting.Store(true) 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) + w.Close() + assert.False(t, w.InProgress(), "repeated Close must leave the worker idle") + assert.Nil(t, w.agent, "no agent may survive Close") } -// 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. +// The existing ownership guard must still protect a newer negotiation when +// cleanup for an older agent is dispatched on the event loop. func TestWorkerICE_StaleCloseAgentKeepsCurrentSession(t *testing.T) { - w := newTestWorkerICE(t) - t.Cleanup(w.Close) - - sidA := icemaker.SessionID("session-a") - w.OnNewOffer(t.Context(), &signaling.OfferAnswer{ - IceCredentials: signaling.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, _ := newTestWorkerICE(t) + w.OnNewOffer(t.Context(), testICEOffer("a")) + agentA := w.agent w.Close() + w.OnNewOffer(t.Context(), testICEOffer("b")) + agentB := w.agent + credentialsB := w.Credentials() - sidB := icemaker.SessionID("session-b") - w.OnNewOffer(t.Context(), &signaling.OfferAnswer{ - IceCredentials: signaling.IceCredentials{UFrag: "ufragbbbb", Pwd: "pwdpwdpwdpwdpwdpwdpwdp2"}, - SessionID: &sidB, - }) - require.True(t, w.InProgress(), "the second negotiation must be in flight") + w.closeAgent(agentA) + assert.Same(t, agentB, w.agent, "B must remain the current agent") + assert.True(t, w.InProgress(), "B must remain in flight") + assert.Equal(t, icemaker.SessionID("b"), w.remoteSessionID, "B's remote session must be preserved") + assert.Equal(t, credentialsB, w.Credentials(), "old cleanup must not rotate B's credentials") +} - w.muxAgent.Lock() - newAgent := w.agent - w.muxAgent.Unlock() +func TestWorkerICE_PionStateChangesWaitForEventLoop(t *testing.T) { + w, events := newTestWorkerICE(t) + w.OnNewOffer(t.Context(), testICEOffer("a")) + agent := w.agent + require.NoError(t, agent.Close()) - // 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") + timer := time.NewTimer(10 * time.Second) + defer timer.Stop() + for { + select { + case ev := <-events: + switch e := ev.(type) { + case ICEStateChanged: + if e.State != ice.ConnectionStateClosed { + continue + } + assert.True(t, w.InProgress(), "Pion's callback must not change worker state") + assert.Same(t, agent, w.agent, "the event loop still owns the agent") + w.OnConnectionStateChange(e) + assert.False(t, w.InProgress(), "consuming Closed must clear the connecting state") + assert.Nil(t, w.agent, "consuming Closed must release the agent") + return + case ICEDialDone: + if e.Conn != nil { + _ = e.Conn.Close() + } + } + case <-timer.C: + t.Fatal("Pion must post Closed to the event queue") + } + } } -// closeTrackConn records Close calls so a test can assert that a discarded -// connection was actually released. type closeTrackConn struct { net.Conn closed atomic.Bool @@ -175,84 +174,57 @@ func (c *closeTrackConn) Close() error { 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) - + w, events := newTestWorkerICE(t) dialStarted := make(chan struct{}) releaseDial := make(chan struct{}) - staleConn := &closeTrackConn{} - - var calls atomic.Int32 + client, server := net.Pipe() + t.Cleanup(func() { _ = client.Close(); _ = server.Close() }) + staleConn := &closeTrackConn{Conn: client} w.dialFunc = func(ctx context.Context, _ *icemaker.ThreadSafeAgent, _ *signaling.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 + close(dialStarted) + select { + case <-releaseDial: return staleConn, nil + case <-t.Context().Done(): + return nil, ctx.Err() } - // A newer negotiation parks on its dialer context, cancelled by the - // t.Cleanup Close at test end. + } + w.OnNewOffer(t.Context(), testICEOffer("a")) + agentA := w.agent + select { + case <-dialStarted: + case <-time.After(10 * time.Second): + t.Fatal("A's dial must start") + } + + // OnNewOffer captures the dial function before spawning the goroutine. + w.dialFunc = func(ctx context.Context, _ *icemaker.ThreadSafeAgent, _ *signaling.OfferAnswer) (net.Conn, error) { <-ctx.Done() return nil, ctx.Err() } - - sidA := icemaker.SessionID("session-a") - w.OnNewOffer(t.Context(), &signaling.OfferAnswer{ - IceCredentials: signaling.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 := icemaker.SessionID("session-b") - w.OnNewOffer(t.Context(), &signaling.OfferAnswer{ - IceCredentials: signaling.IceCredentials{UFrag: "ufragbbbb", Pwd: "pwdpwdpwdpwdpwdpwdpwdp2"}, - SessionID: &sidB, - }) - - w.muxAgent.Lock() + w.OnNewOffer(t.Context(), testICEOffer("b")) 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") + e := waitICEDialDone(t, events, agentA) + assert.False(t, staleConn.closed.Load(), "a successfully posted result is owned by the event loop") + _, _, ready := w.OnDialDone(e) + assert.False(t, ready, "A's late result must not be accepted") + assert.True(t, staleConn.closed.Load(), "the consumer must release A's connection") + assert.Same(t, agentB, w.agent, "A's result must leave B installed") + assert.True(t, w.InProgress(), "B must remain in flight") +} - 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. +func TestWorkerICE_RejectedDialEventClosesConnection(t *testing.T) { + w, _ := newTestWorkerICE(t) + w.postEvent = func(any) bool { return false } + client, server := net.Pipe() + t.Cleanup(func() { _ = client.Close(); _ = server.Close() }) + remote := &closeTrackConn{Conn: client} + w.dialFunc = func(context.Context, *icemaker.ThreadSafeAgent, *signaling.OfferAnswer) (net.Conn, error) { + return remote, nil + } + w.OnNewOffer(t.Context(), testICEOffer("a")) + require.Eventually(t, remote.closed.Load, 10*time.Second, time.Millisecond, + "when the mailbox rejects the result the producer must close its connection") }