sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-07-16 19:49:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

[client] Move ICE and relay workers into the worker package

Browse Source 
Relocate WorkerICE (renamed worker.ICE) and WorkerRelay plus ConnPriority
into client/internal/peer/worker. To break the peer<->worker cycle the
workers no longer take *Conn or ConnConfig: callbacks are passed as plain
functions (Conn's unexported methods as method values), and each worker
receives only the fields it needs (key, ICE config, isController) plus a
small services struct. Context is passed to OnNewOffer instead of stored.

Move the worker connection-status helper the other way, out of the worker
package into peer as worker_status.go (WorkerStatus / AtomicWorkerStatus),
since only Conn uses it.
This commit is contained in:
Zoltan Papp
2026-07-11 20:06:46 +02:00
parent 917e85f648
commit 791e8b33ae
7 changed files with 154 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
client/internal/peer/conn.go
View File

@@ -97,6 +97,10 @@ type ConnConfig struct {
ICEConfig icemaker.Config
}
func (c ConnConfig) IsController() bool {
return c.LocalKey > c.Key
}
// Conn represents a connection to a remote peer. All mutable connection state
// is owned by a single event loop goroutine started in Open; external callers
// and the transport workers communicate with the loop by posting events into
@@ -120,9 +124,9 @@ type Conn struct {
onDisconnected func(remotePeer string)
rosenpassInitializedPresharedKeyValidator func(peerKey string) bool
statusRelay *worker.AtomicWorkerStatus
statusICE *worker.AtomicWorkerStatus
currentConnPriority ConnPriority
statusRelay *AtomicWorkerStatus
statusICE *AtomicWorkerStatus
currentConnPriority worker.ConnPriority
opened bool // this flag is used to prevent close in case of not opened connection
// mailbox delivers events to the event loop of the current open
@@ -130,8 +134,8 @@ type Conn struct {
mailbox atomic.Pointer[mailbox]
loopDone chan struct{}
workerICE *WorkerICE
workerRelay *WorkerRelay
workerICE *worker.ICE
workerRelay *worker.WorkerRelay
// relayDialInFlight and pendingRelayOffer serialize the blocking relay
// dials spawned by the event loop, keeping only the newest offer while a
@@ -189,10 +193,10 @@ func NewConn(config ConnConfig, services ServiceDependencies) (*Conn, error) {
relayManager: services.RelayManager,
srWatcher: services.SrWatcher,
portForwardManager: services.PortForwardManager,
statusRelay: worker.NewAtomicStatus(),
statusICE: worker.NewAtomicStatus(),
statusRelay: NewAtomicStatus(),
statusICE: NewAtomicStatus(),
dumpState: dumpState,
endpointUpdater: NewEndpointUpdater(connLog, config.WgConfig, isController(config)),
endpointUpdater: NewEndpointUpdater(connLog, config.WgConfig, config.IsController()),
wgWatcher: wg_watcher.NewWGWatcher(connLog, config.WgConfig.WgInterface, config.Key, dumpState),
metricsRecorder: services.MetricsRecorder,
}
@@ -226,11 +230,16 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error {
conn.ctx, conn.ctxCancel = context.WithCancel(engineCtx)
conn.workerRelay = NewWorkerRelay(conn.ctx, conn.Log, isController(conn.config), conn.config, conn, conn.relayManager)
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 := NewWorkerICE(conn.ctx, conn.Log, conn.config, conn, conn.signaler, conn.iFaceDiscover, conn.statusRecorder, conn.portForwardManager, relayIsSupportedLocally)
workerICE, err := worker.NewICE(conn.Log, conn.config.Key, conn.config.ICEConfig, conn.config.IsController(), conn.onICEConnectionIsReady, conn.onICEStateDisconnected, worker.ICEDependencies{
Signaler: conn.signaler,
IFaceDiscover: conn.iFaceDiscover,
StatusRecorder: conn.statusRecorder,
PortForwardManager: conn.portForwardManager,
}, relayIsSupportedLocally)
if err != nil {
return err
}
@@ -507,7 +516,7 @@ func (conn *Conn) releaseEvents(evs []event) {
for _, ev := range evs {
switch e := ev.(type) {
case evRelayReady:
if err := e.info.relayedConn.Close(); err != nil {
if err := e.info.RelayedConn.Close(); err != nil {
conn.Log.Warnf("failed to close unnecessary relayed connection: %v", err)
}
case evClose:
@@ -558,7 +567,7 @@ func (conn *Conn) dispatchOfferToICE(offer *signaling.OfferAnswer) {
if conn.workerICE == nil || !conn.handshaker.RemoteICESupported() {
return
}
conn.workerICE.OnNewOffer(offer)
conn.workerICE.OnNewOffer(conn.ctx, offer)
}
// dispatchOfferToRelay runs the blocking relay dial on a helper goroutine. A
@@ -572,7 +581,7 @@ func (conn *Conn) dispatchOfferToRelay(offer *signaling.OfferAnswer) {
conn.relayDialInFlight = true
go func() {
conn.workerRelay.OnNewOffer(offer)
conn.workerRelay.OnNewOffer(conn.ctx, offer)
conn.post(evRelayDialDone{})
}()
}
@@ -598,7 +607,7 @@ func (conn *Conn) handleGuardTick() {
// handleICEReady starts proxying traffic from/to local WireGuard and sets the
// connection status to StatusConnected.
func (conn *Conn) handleICEReady(priority ConnPriority, iceConnInfo ICEConnInfo) {
func (conn *Conn) handleICEReady(priority worker.ConnPriority, iceConnInfo worker.ICEConnInfo) {
if conn.ctx.Err() != nil {
return
}
@@ -707,16 +716,16 @@ func (conn *Conn) handleICEDisconnected(sessionChanged bool) {
conn.Log.Errorf("failed to switch to relay conn: %v", err)
}
conn.currentConnPriority = Relay
conn.currentConnPriority = worker.Relay
} else {
conn.Log.Infof("ICE disconnected, do not switch to Relay. Reset priority to: %s", None.String())
conn.currentConnPriority = None
conn.Log.Infof("ICE disconnected, do not switch to Relay. Reset priority to: %s", worker.None.String())
conn.currentConnPriority = worker.None
if err := conn.config.WgConfig.WgInterface.RemoveEndpointAddress(conn.config.WgConfig.RemoteKey); err != nil {
conn.Log.Errorf("failed to remove wg endpoint: %v", err)
}
}
changed := conn.statusICE.Get() != worker.StatusDisconnected
changed := conn.statusICE.Get() != WorkerStatusDisconnected
if changed {
conn.guard.SetICEConnDisconnected()
}
@@ -724,7 +733,7 @@ func (conn *Conn) handleICEDisconnected(sessionChanged bool) {
conn.disableWgWatcherIfNeeded()
if conn.currentConnPriority == None {
if conn.currentConnPriority == worker.None {
conn.metricsStages.Disconnected()
}
@@ -741,9 +750,9 @@ func (conn *Conn) handleICEDisconnected(sessionChanged bool) {
// handleRelayReady sets up the WireGuard proxy for a freshly opened relayed
// connection and activates it unless ICE has priority.
func (conn *Conn) handleRelayReady(rci RelayConnInfo) {
func (conn *Conn) handleRelayReady(rci worker.RelayConnInfo) {
if conn.ctx.Err() != nil {
if err := rci.relayedConn.Close(); err != nil {
if err := rci.RelayedConn.Close(); err != nil {
conn.Log.Warnf("failed to close unnecessary relayed connection: %v", err)
}
return
@@ -752,7 +761,7 @@ func (conn *Conn) handleRelayReady(rci RelayConnInfo) {
conn.dumpState.RelayConnected()
conn.Log.Debugf("Relay connection has been established, setup the WireGuard")
wgProxy, err := conn.newProxy(rci.relayedConn)
wgProxy, err := conn.newProxy(rci.RelayedConn)
if err != nil {
conn.Log.Errorf("failed to add relayed net.Conn to local proxy: %v", err)
return
@@ -767,18 +776,18 @@ func (conn *Conn) handleRelayReady(rci RelayConnInfo) {
conn.Log.Debugf("do not switch to relay because current priority is: %s", conn.currentConnPriority.String())
conn.setRelayedProxy(wgProxy)
conn.statusRelay.SetConnected()
conn.updateRelayStatus(rci.relayedConn.RemoteAddr().String(), rci.rosenpassPubKey, time.Now())
conn.updateRelayStatus(rci.RelayedConn.RemoteAddr().String(), rci.RosenpassPubKey, time.Now())
return
}
controller := isController(conn.config)
controller := conn.config.IsController()
if controller {
wgProxy.Work()
}
updateTime := time.Now()
conn.enableWgWatcherIfNeeded(updateTime)
if err := conn.endpointUpdater.ConfigureWGEndpoint(wgProxy.EndpointAddr(), conn.presharedKey(rci.rosenpassPubKey)); err != nil {
if err := conn.endpointUpdater.ConfigureWGEndpoint(wgProxy.EndpointAddr(), conn.presharedKey(rci.RosenpassPubKey)); err != nil {
if err := wgProxy.CloseConn(); err != nil {
conn.Log.Warnf("Failed to close relay connection: %v", err)
}
@@ -793,13 +802,13 @@ func (conn *Conn) handleRelayReady(rci RelayConnInfo) {
conn.injectPendingFirstPacket(wgProxy, nil)
conn.rosenpassRemoteKey = rci.rosenpassPubKey
conn.currentConnPriority = Relay
conn.rosenpassRemoteKey = rci.RosenpassPubKey
conn.currentConnPriority = worker.Relay
conn.statusRelay.SetConnected()
conn.setRelayedProxy(wgProxy)
conn.updateRelayStatus(rci.relayedConn.RemoteAddr().String(), rci.rosenpassPubKey, updateTime)
conn.updateRelayStatus(rci.RelayedConn.RemoteAddr().String(), rci.RosenpassPubKey, updateTime)
conn.Log.Infof("start to communicate with peer via relay")
conn.doOnConnected(rci.rosenpassPubKey, rci.rosenpassAddr, updateTime)
conn.doOnConnected(rci.RosenpassPubKey, rci.RosenpassAddr, updateTime)
}
// handleRelayDisconnected cleans up the relayed transport and the WireGuard
@@ -811,9 +820,9 @@ func (conn *Conn) handleRelayDisconnected() {
conn.Log.Debugf("relay connection is disconnected")
if conn.currentConnPriority == Relay {
if conn.currentConnPriority == worker.Relay {
conn.Log.Debugf("clean up WireGuard config")
conn.currentConnPriority = None
conn.currentConnPriority = worker.None
if err := conn.config.WgConfig.WgInterface.RemoveEndpointAddress(conn.config.WgConfig.RemoteKey); err != nil {
conn.Log.Errorf("failed to remove wg endpoint: %v", err)
}
@@ -824,7 +833,7 @@ func (conn *Conn) handleRelayDisconnected() {
conn.wgProxyRelay = nil
}
changed := conn.statusRelay.Get() != worker.StatusDisconnected
changed := conn.statusRelay.Get() != WorkerStatusDisconnected
if changed {
conn.guard.SetRelayedConnDisconnected()
}
@@ -832,7 +841,7 @@ func (conn *Conn) handleRelayDisconnected() {
conn.disableWgWatcherIfNeeded()
if conn.currentConnPriority == None {
if conn.currentConnPriority == worker.None {
conn.metricsStages.Disconnected()
}
@@ -858,10 +867,10 @@ func (conn *Conn) handleWGTimeout() {
// Close the active connection based on current priority
switch conn.currentConnPriority {
case Relay:
case worker.Relay:
conn.workerRelay.CloseConn()
conn.handleRelayDisconnected()
case ICEP2P, ICETurn:
case worker.ICEP2P, worker.ICETurn:
conn.workerICE.Close()
default:
conn.Log.Debugf("No active connection to close on WG timeout")
@@ -900,7 +909,7 @@ func (conn *Conn) handleWGCheckSuccess() {
conn.wgTimeouts = 0
}
func (conn *Conn) onICEConnectionIsReady(priority ConnPriority, iceConnInfo ICEConnInfo) {
func (conn *Conn) onICEConnectionIsReady(priority worker.ConnPriority, iceConnInfo worker.ICEConnInfo) {
conn.post(evICEReady{priority: priority, info: iceConnInfo})
}
@@ -910,11 +919,11 @@ func (conn *Conn) onICEStateDisconnected(sessionChanged bool) {
// onRelayConnectionIsReady closes the relayed connection when the event loop
// is gone and nobody will take ownership of it.
func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) {
func (conn *Conn) onRelayConnectionIsReady(rci worker.RelayConnInfo) {
if conn.post(evRelayReady{info: rci}) {
return
}
if err := rci.relayedConn.Close(); err != nil {
if err := rci.RelayedConn.Close(); err != nil {
conn.Log.Warnf("failed to close unnecessary relayed connection: %v", err)
}
}
@@ -984,7 +993,7 @@ func (conn *Conn) updateRelayStatus(relayServerAddr string, rosenpassPubKey []by
}
}
func (conn *Conn) updateIceState(iceConnInfo ICEConnInfo, updateTime time.Time) {
func (conn *Conn) updateIceState(iceConnInfo worker.ICEConnInfo, updateTime time.Time) {
peerState := status.State{
PubKey: conn.config.Key,
ConnStatusUpdate: updateTime,
@@ -1006,7 +1015,7 @@ func (conn *Conn) updateIceState(iceConnInfo ICEConnInfo, updateTime time.Time)
func (conn *Conn) setStatusToDisconnected() {
conn.statusRelay.SetDisconnected()
conn.statusICE.SetDisconnected()
conn.currentConnPriority = None
conn.currentConnPriority = worker.None
peerState := status.State{
PubKey: conn.config.Key,
@@ -1039,7 +1048,7 @@ func (conn *Conn) doOnConnected(remoteRosenpassPubKey []byte, remoteRosenpassAdd
func (conn *Conn) isRelayed() bool {
switch conn.currentConnPriority {
case Relay, ICETurn:
case worker.Relay, worker.ICETurn:
return true
default:
return false
@@ -1047,7 +1056,7 @@ func (conn *Conn) isRelayed() bool {
}
func (conn *Conn) evalStatus() status.ConnStatus {
if conn.statusRelay.Get() == worker.StatusConnected || conn.statusICE.Get() == worker.StatusConnected {
if conn.statusRelay.Get() == WorkerStatusConnected || conn.statusICE.Get() == WorkerStatusConnected {
return status.StatusConnected
}
@@ -1077,10 +1086,10 @@ func (conn *Conn) isConnectedOnAllWay() (status guard.ConnStatus) {
return evalConnStatus(connStatusInputs{
forceRelay: IsForceRelayed(),
peerUsesRelay: conn.workerRelay.IsRelayConnectionSupportedWithPeer(),
relayConnected: conn.statusRelay.Get() == worker.StatusConnected,
relayConnected: conn.statusRelay.Get() == WorkerStatusConnected,
remoteSupportsICE: conn.handshaker.RemoteICESupported(),
iceWorkerCreated: iceWorkerCreated,
iceStatusConnecting: conn.statusICE.Get() != worker.StatusDisconnected,
iceStatusConnecting: conn.statusICE.Get() != WorkerStatusDisconnected,
iceInProgress: iceInProgress,
})
}
@@ -1100,7 +1109,7 @@ func (conn *Conn) enableWgWatcherIfNeeded(enabledTime time.Time) {
}
func (conn *Conn) disableWgWatcherIfNeeded() {
if conn.currentConnPriority == None && conn.wgWatcherCancel != nil {
if conn.currentConnPriority == worker.None && conn.wgWatcherCancel != nil {
conn.wgWatcherCancel()
conn.wgWatcherCancel = nil
}
@@ -1122,7 +1131,7 @@ func (conn *Conn) newProxy(remoteConn net.Conn) (wgproxy.Proxy, error) {
}
func (conn *Conn) resetEndpoint() {
if !isController(conn.config) {
if !conn.config.IsController() {
return
}
conn.Log.Infof("reset wg endpoint")
@@ -1133,11 +1142,11 @@ func (conn *Conn) resetEndpoint() {
}
func (conn *Conn) isReadyToUpgrade() bool {
return conn.wgProxyRelay != nil && conn.currentConnPriority != Relay
return conn.wgProxyRelay != nil && conn.currentConnPriority != worker.Relay
}
func (conn *Conn) isICEActive() bool {
return (conn.currentConnPriority == ICEP2P || conn.currentConnPriority == ICETurn) && conn.statusICE.Get() == worker.StatusConnected
return (conn.currentConnPriority == worker.ICEP2P || conn.currentConnPriority == worker.ICETurn) && conn.statusICE.Get() == WorkerStatusConnected
}
func (conn *Conn) handleConfigurationFailure(err error, wgProxy wgproxy.Proxy) {
@@ -1177,7 +1186,7 @@ func (conn *Conn) recordConnectionMetrics() {
var connType metrics.ConnectionType
switch conn.currentConnPriority {
case Relay:
case worker.Relay:
connType = metrics.ConnectionTypeRelay
default:
connType = metrics.ConnectionTypeICE
@@ -1225,10 +1234,6 @@ func (conn *Conn) presharedKey(remoteRosenpassKey []byte) *wgtypes.Key {
return determKey
}
func isController(config ConnConfig) bool {
return config.LocalKey > config.Key
}
func isRosenpassEnabled(remoteRosenpassPubKey []byte) bool {
return remoteRosenpassPubKey != nil
}

2
client/internal/peer/conn_status.go
View File

@@ -8,7 +8,7 @@ type connStatusInputs struct {
peerUsesRelay bool // remote peer advertises relay support AND local has relay
relayConnected bool // statusRelay reports Connected (independent of whether peer uses relay)
remoteSupportsICE bool // remote peer sent ICE credentials
iceWorkerCreated bool // local WorkerICE exists (false in force-relay mode)
iceWorkerCreated bool // local ICE worker exists (false in force-relay mode)
iceStatusConnecting bool // statusICE is anything other than Disconnected
iceInProgress bool // a negotiation is currently in flight
}

7
client/internal/peer/event.go
View File

@@ -6,6 +6,7 @@ import (
"github.com/pion/ice/v4"
"github.com/netbirdio/netbird/client/internal/peer/signaling"
"github.com/netbirdio/netbird/client/internal/peer/worker"
"github.com/netbirdio/netbird/route"
)
@@ -35,8 +36,8 @@ type evRemoteCandidate struct {
}
type evICEReady struct {
priority ConnPriority
info ICEConnInfo
priority worker.ConnPriority
info worker.ICEConnInfo
}
type evICEDown struct {
@@ -44,7 +45,7 @@ type evICEDown struct {
}
type evRelayReady struct {
info RelayConnInfo
info worker.RelayConnInfo
}
type evRelayDown struct{}

2
client/internal/peer/priority.go → client/internal/peer/worker/priority.go
View File

@@ -1,4 +1,4 @@
package peer
package worker
import (
"fmt"

104
client/internal/peer/worker_ice.go → client/internal/peer/worker/worker_ice.go
View File

@@ -1,4 +1,4 @@
package peer
package worker
import (
"context"
@@ -21,11 +21,6 @@ import (
"github.com/netbirdio/netbird/route"
)
type iceCallbacks interface {
onICEConnectionIsReady(priority ConnPriority, iceConnInfo ICEConnInfo)
onICEStateDisconnected(sessionChanged bool)
}
type ICEConnInfo struct {
RemoteConn net.Conn
RosenpassPubKey []byte
@@ -38,11 +33,20 @@ type ICEConnInfo struct {
RelayedOnLocal bool
}
type WorkerICE struct {
ctx context.Context
type ICEDependencies struct {
Signaler *signaling.Signaler
IFaceDiscover stdnet.ExternalIFaceDiscover
StatusRecorder *status.Recorder
PortForwardManager *portforward.Manager
}
type ICE struct {
log *log.Entry
config ConnConfig
conn iceCallbacks
key string
iceConfig icemaker.Config
isController bool
onConnReady func(priority ConnPriority, iceConnInfo ICEConnInfo)
onStatusDisconnect func(sessionChanged bool)
signaler *signaling.Signaler
iFaceDiscover stdnet.ExternalIFaceDiscover
statusRecorder *status.Recorder
@@ -72,21 +76,23 @@ type WorkerICE struct {
portForwardAttempted bool
}
func NewWorkerICE(ctx context.Context, log *log.Entry, config ConnConfig, conn iceCallbacks, signaler *signaling.Signaler, ifaceDiscover stdnet.ExternalIFaceDiscover, statusRecorder *status.Recorder, portForwardManager *portforward.Manager, hasRelayOnLocally bool) (*WorkerICE, error) {
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) {
sessionID, err := icemaker.NewSessionID()
if err != nil {
return nil, err
}
w := &WorkerICE{
ctx: ctx,
w := &ICE{
log: log,
config: config,
conn: conn,
signaler: signaler,
iFaceDiscover: ifaceDiscover,
statusRecorder: statusRecorder,
portForwardManager: portForwardManager,
key: key,
iceConfig: iceConfig,
isController: isController,
onConnReady: onConnReady,
onStatusDisconnect: onStatusDisconnect,
signaler: services.Signaler,
iFaceDiscover: services.IFaceDiscover,
statusRecorder: services.StatusRecorder,
portForwardManager: services.PortForwardManager,
hasRelayOnLocally: hasRelayOnLocally,
sessionID: sessionID,
}
@@ -100,7 +106,7 @@ func NewWorkerICE(ctx context.Context, log *log.Entry, config ConnConfig, conn i
return w, nil
}
func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *signaling.OfferAnswer) {
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()
@@ -142,8 +148,8 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *signaling.OfferAnswer) {
if remoteOfferAnswer.SessionID != nil {
w.log.Debugf("recreate ICE agent: %s / %s", w.sessionID, *remoteOfferAnswer.SessionID)
}
dialerCtx, dialerCancel := context.WithCancel(w.ctx)
agent, err := w.reCreateAgent(dialerCancel, preferredCandidateTypes)
dialerCtx, dialerCancel := context.WithCancel(ctx)
agent, err := w.reCreateAgent(ctx, dialerCancel, preferredCandidateTypes)
if err != nil {
w.log.Errorf("failed to recreate ICE Agent: %s", err)
return
@@ -161,10 +167,10 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *signaling.OfferAnswer) {
}
// OnRemoteCandidate Handles ICE connection Candidate provided by the remote peer.
func (w *WorkerICE) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HAMap) {
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.config.Key, candidate.String())
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")
return
@@ -191,7 +197,7 @@ func (w *WorkerICE) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HA
}
}
func (w *WorkerICE) Credentials() signaling.Credentials {
func (w *ICE) Credentials() signaling.Credentials {
w.muxAgent.Lock()
defer w.muxAgent.Unlock()
return signaling.Credentials{
@@ -201,14 +207,14 @@ func (w *WorkerICE) Credentials() signaling.Credentials {
}
}
func (w *WorkerICE) InProgress() bool {
func (w *ICE) InProgress() bool {
w.muxAgent.Lock()
defer w.muxAgent.Unlock()
return w.agentConnecting
}
func (w *WorkerICE) Close() {
func (w *ICE) Close() {
w.muxAgent.Lock()
defer w.muxAgent.Unlock()
@@ -224,10 +230,10 @@ func (w *WorkerICE) Close() {
w.agent = nil
}
func (w *WorkerICE) reCreateAgent(dialerCancel context.CancelFunc, candidates []ice.CandidateType) (*icemaker.ThreadSafeAgent, error) {
func (w *ICE) reCreateAgent(ctx context.Context, dialerCancel context.CancelFunc, candidates []ice.CandidateType) (*icemaker.ThreadSafeAgent, error) {
w.portForwardAttempted = false
agent, err := icemaker.NewAgent(w.ctx, w.iFaceDiscover, w.config.ICEConfig, candidates, w.localUfrag, w.localPwd)
agent, err := icemaker.NewAgent(ctx, w.iFaceDiscover, w.iceConfig, candidates, w.localUfrag, w.localPwd)
if err != nil {
return nil, fmt.Errorf("create agent: %w", err)
}
@@ -249,7 +255,7 @@ func (w *WorkerICE) reCreateAgent(dialerCancel context.CancelFunc, candidates []
return agent, nil
}
func (w *WorkerICE) getSessionID() icemaker.SessionID {
func (w *ICE) getSessionID() icemaker.SessionID {
w.muxAgent.Lock()
defer w.muxAgent.Unlock()
@@ -259,7 +265,7 @@ func (w *WorkerICE) getSessionID() icemaker.SessionID {
// 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, dialerCancel context.CancelFunc, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *signaling.OfferAnswer) {
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)
@@ -323,10 +329,10 @@ func (w *WorkerICE) connect(ctx context.Context, dialerCancel context.CancelFunc
w.muxAgent.Unlock()
w.log.Infof("connection succeeded with offer session: %s", remoteOfferAnswer.SessionIDString())
w.conn.onICEConnectionIsReady(selectedPriority(pair), ci)
w.onConnReady(selectedPriority(pair), ci)
}
func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.CancelFunc) bool {
func (w *ICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.CancelFunc) bool {
cancel()
if err := agent.Close(); err != nil {
w.log.Warnf("failed to close ICE agent: %s", err)
@@ -352,7 +358,7 @@ func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.C
return sessionChanged
}
func (w *WorkerICE) punchRemoteWGPort(pair *ice.CandidatePair, remoteWgPort int) {
func (w *ICE) punchRemoteWGPort(pair *ice.CandidatePair, remoteWgPort int) {
// wait local endpoint configuration
time.Sleep(time.Second)
addr, err := net.ResolveUDPAddr("udp", net.JoinHostPort(pair.Remote.Address(), strconv.Itoa(remoteWgPort)))
@@ -361,7 +367,7 @@ func (w *WorkerICE) punchRemoteWGPort(pair *ice.CandidatePair, remoteWgPort int)
return
}
mux, ok := w.config.ICEConfig.UDPMuxSrflx.(*udpmux.UniversalUDPMuxDefault)
mux, ok := w.iceConfig.UDPMuxSrflx.(*udpmux.UniversalUDPMuxDefault)
if !ok {
w.log.Warn("invalid udp mux conversion")
return
@@ -374,7 +380,7 @@ func (w *WorkerICE) 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 *WorkerICE) onICECandidate(candidate ice.Candidate) {
func (w *ICE) onICECandidate(candidate ice.Candidate) {
// nil means candidate gathering has been ended
if candidate == nil {
return
@@ -383,9 +389,9 @@ func (w *WorkerICE) onICECandidate(candidate ice.Candidate) {
// TODO: reported port is incorrect for CandidateTypeHost, makes understanding ICE use via logs confusing as port is ignored
w.log.Debugf("discovered local candidate %s", candidate.String())
go func() {
err := w.signaler.SignalICECandidate(candidate, w.config.Key)
err := w.signaler.SignalICECandidate(candidate, w.key)
if err != nil {
w.log.Errorf("failed signaling candidate to the remote peer %s %s", w.config.Key, err)
w.log.Errorf("failed signaling candidate to the remote peer %s %s", w.key, err)
}
}()
@@ -395,7 +401,7 @@ func (w *WorkerICE) onICECandidate(candidate ice.Candidate) {
}
// injectPortForwardedCandidate signals an additional candidate using the pre-created port mapping.
func (w *WorkerICE) injectPortForwardedCandidate(srflxCandidate ice.Candidate) {
func (w *ICE) injectPortForwardedCandidate(srflxCandidate ice.Candidate) {
pfManager := w.portForwardManager
if pfManager == nil {
return
@@ -424,7 +430,7 @@ func (w *WorkerICE) injectPortForwardedCandidate(srflxCandidate ice.Candidate) {
forwardedCandidate.String(), mapping.InternalPort, mapping.ExternalPort, mapping.NATType, forwardedCandidate.Priority())
go func() {
if err := w.signaler.SignalICECandidate(forwardedCandidate, w.config.Key); err != nil {
if err := w.signaler.SignalICECandidate(forwardedCandidate, w.key); err != nil {
w.log.Errorf("signal port-forwarded candidate: %v", err)
}
}()
@@ -432,7 +438,7 @@ func (w *WorkerICE) injectPortForwardedCandidate(srflxCandidate ice.Candidate) {
// createForwardedCandidate creates a new server reflexive candidate with the forwarded port.
// It uses the NAT gateway's external IP with the forwarded port.
func (w *WorkerICE) createForwardedCandidate(srflxCandidate ice.Candidate, mapping *portforward.Mapping) (ice.Candidate, error) {
func (w *ICE) createForwardedCandidate(srflxCandidate ice.Candidate, mapping *portforward.Mapping) (ice.Candidate, error) {
var externalIP string
if mapping.ExternalIP != nil && !mapping.ExternalIP.IsUnspecified() {
externalIP = mapping.ExternalIP.String()
@@ -477,9 +483,9 @@ func (w *WorkerICE) createForwardedCandidate(srflxCandidate ice.Candidate, mappi
return candidate, nil
}
func (w *WorkerICE) onICESelectedCandidatePair(agent *icemaker.ThreadSafeAgent, c1, c2 ice.Candidate) {
func (w *ICE) onICESelectedCandidatePair(agent *icemaker.ThreadSafeAgent, c1, c2 ice.Candidate) {
w.log.Debugf("selected candidate pair [local <-> remote] -> [%s <-> %s], peer %s", c1.String(), c2.String(),
w.config.Key)
w.key)
pairStat, ok := agent.GetSelectedCandidatePairStats()
if !ok {
@@ -488,13 +494,13 @@ func (w *WorkerICE) onICESelectedCandidatePair(agent *icemaker.ThreadSafeAgent,
}
duration := time.Duration(pairStat.CurrentRoundTripTime * float64(time.Second))
if err := w.statusRecorder.UpdateLatency(w.config.Key, duration); err != nil {
if err := w.statusRecorder.UpdateLatency(w.key, duration); err != nil {
w.log.Debugf("failed to update latency for peer: %s", err)
return
}
}
func (w *WorkerICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) {
func (w *ICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) {
sessionID := w.getSessionID()
stats := agent.GetCandidatePairsStats()
localCandidates, _ := agent.GetLocalCandidates()
@@ -525,7 +531,7 @@ func (w *WorkerICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) {
}
}
func (w *WorkerICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dialerCancel context.CancelFunc) func(ice.ConnectionState) {
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) {
@@ -556,13 +562,13 @@ func (w *WorkerICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dia
w.log.Debugf("suppress disconnected event of replaced ICE agent")
return
}
w.conn.onICEStateDisconnected(sessionChanged)
w.onStatusDisconnect(sessionChanged)
}
}
}
func (w *WorkerICE) turnAgentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *signaling.OfferAnswer) (*ice.Conn, error) {
if isController(w.config) {
func (w *ICE) turnAgentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *signaling.OfferAnswer) (*ice.Conn, error) {
if w.isController {
return agent.Dial(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd)
} else {
return agent.Accept(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd)

53
client/internal/peer/worker_relay.go → client/internal/peer/worker/worker_relay.go
View File

@@ -1,4 +1,4 @@
package peer
package worker
import (
"context"
@@ -15,23 +15,18 @@ import (
)
type RelayConnInfo struct {
relayedConn net.Conn
rosenpassPubKey []byte
rosenpassAddr string
}
type relayCallbacks interface {
onRelayConnectionIsReady(rci RelayConnInfo)
onRelayDisconnected()
RelayedConn net.Conn
RosenpassPubKey []byte
RosenpassAddr string
}
type WorkerRelay struct {
peerCtx context.Context
log *log.Entry
isController bool
config ConnConfig
conn relayCallbacks
relayManager *relayClient.Manager
log *log.Entry
key string
isController bool
onConnReady func(RelayConnInfo)
onDisconnected func()
relayManager *relayClient.Manager
relayedConn net.Conn
relayLock sync.Mutex
@@ -39,19 +34,19 @@ type WorkerRelay struct {
relaySupportedOnRemotePeer atomic.Bool
}
func NewWorkerRelay(ctx context.Context, log *log.Entry, ctrl bool, config ConnConfig, conn relayCallbacks, relayManager *relayClient.Manager) *WorkerRelay {
func NewWorkerRelay(log *log.Entry, key string, isController bool, onConnReady func(RelayConnInfo), onDisconnected func(), relayManager *relayClient.Manager) *WorkerRelay {
r := &WorkerRelay{
peerCtx: ctx,
log: log,
isController: ctrl,
config: config,
conn: conn,
relayManager: relayManager,
log: log,
key: key,
isController: isController,
onConnReady: onConnReady,
onDisconnected: onDisconnected,
relayManager: relayManager,
}
return r
}
func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *signaling.OfferAnswer) {
func (w *WorkerRelay) OnNewOffer(ctx context.Context, remoteOfferAnswer *signaling.OfferAnswer) {
if !w.isRelaySupported(remoteOfferAnswer) {
w.log.Infof("Relay is not supported by remote peer")
w.relaySupportedOnRemotePeer.Store(false)
@@ -72,7 +67,7 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *signaling.OfferAnswer) {
serverIP = remoteOfferAnswer.RelaySrvIP
}
relayedConn, err := w.relayManager.OpenConn(w.peerCtx, srv, w.config.Key, serverIP)
relayedConn, err := w.relayManager.OpenConn(ctx, srv, w.key, serverIP)
if err != nil {
if errors.Is(err, relayClient.ErrConnAlreadyExists) {
w.log.Debugf("handled offer by reusing existing relay connection")
@@ -94,10 +89,10 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *signaling.OfferAnswer) {
}
w.log.Debugf("peer conn opened via Relay: %s", srv)
w.conn.onRelayConnectionIsReady(RelayConnInfo{
relayedConn: relayedConn,
rosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
rosenpassAddr: remoteOfferAnswer.RosenpassAddr,
w.onConnReady(RelayConnInfo{
RelayedConn: relayedConn,
RosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
RosenpassAddr: remoteOfferAnswer.RosenpassAddr,
})
}
@@ -140,5 +135,5 @@ func (w *WorkerRelay) preferredRelayServer(myRelayAddress, remoteRelayAddress st
}
func (w *WorkerRelay) onRelayClientDisconnected() {
w.conn.onRelayDisconnected()
w.onDisconnected()
}

22
client/internal/peer/worker/state.go → client/internal/peer/worker_status.go
View File

@@ -1,4 +1,4 @@
package worker
package peer
import (
"sync/atomic"
@@ -7,17 +7,17 @@ import (
)
const (
StatusDisconnected Status = iota
StatusConnected
WorkerStatusDisconnected WorkerStatus = iota
WorkerStatusConnected
)
type Status int32
type WorkerStatus int32
func (s Status) String() string {
func (s WorkerStatus) String() string {
switch s {
case StatusDisconnected:
case WorkerStatusDisconnected:
return "Disconnected"
case StatusConnected:
case WorkerStatusConnected:
return "Connected"
default:
log.Errorf("unknown status: %d", s)
@@ -37,16 +37,16 @@ func NewAtomicStatus() *AtomicWorkerStatus {
}
// Get returns the current connection status
func (acs *AtomicWorkerStatus) Get() Status {
return Status(acs.status.Load())
func (acs *AtomicWorkerStatus) Get() WorkerStatus {
return WorkerStatus(acs.status.Load())
}
func (acs *AtomicWorkerStatus) SetConnected() {
acs.status.Store(int32(StatusConnected))
acs.status.Store(int32(WorkerStatusConnected))
}
func (acs *AtomicWorkerStatus) SetDisconnected() {
acs.status.Store(int32(StatusDisconnected))
acs.status.Store(int32(WorkerStatusDisconnected))
}
// String returns the string representation of the current status