[client] Move metrics saver into its own metricsstages package

Relocate MetricsStages (per-connection metric stage timestamps) out of
the peer package into client/internal/peer/metricsstages.

client/internal/engine.go

@@ -48,6 +48,7 @@ import (
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/client/internal/peer/guard"
 	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
+	"github.com/netbirdio/netbird/client/internal/peer/signaling"
 	"github.com/netbirdio/netbird/client/internal/peerstore"
 	"github.com/netbirdio/netbird/client/internal/portforward"
 	"github.com/netbirdio/netbird/client/internal/profilemanager"
@@ -182,7 +183,7 @@ type EngineServices struct {
 type Engine struct {
 	// signal is a Signal Service client
 	signal   signal.Client
-	signaler *peer.Signaler
+	signaler *signaling.Signaler
 	// mgmClient is a Management Service client
 	mgmClient mgm.Client
 	// peerConns is a map that holds all the peers that are known to this peer
@@ -318,7 +319,7 @@ func NewEngine(
 		ctx:                ctx,
 		cancel:             cancel,
 		signal:             services.SignalClient,
-		signaler:           peer.NewSignaler(services.SignalClient, config.WgPrivateKey),
+		signaler:           signaling.NewSignaler(services.SignalClient, config.WgPrivateKey),
 		mgmClient:          services.MgmClient,
 		relayManager:       services.RelayManager,
 		peerStore:          peerstore.NewConnStore(),
@@ -2747,7 +2748,7 @@ func createFile(path string) error {
 	return file.Close()
 }
 
-func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) {
+func convertToOfferAnswer(msg *sProto.Message) (*signaling.OfferAnswer, error) {
 	remoteCred, err := signal.UnMarshalCredential(msg)
 	if err != nil {
 		return nil, err
@@ -2763,9 +2764,9 @@ func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) {
 	}
 
 	// Handle optional SessionID
-	var sessionID *peer.ICESessionID
+	var sessionID *icemaker.SessionID
 	if sessionBytes := msg.GetBody().GetSessionId(); sessionBytes != nil {
-		if id, err := peer.ICESessionIDFromBytes(sessionBytes); err != nil {
+		if id, err := icemaker.SessionIDFromBytes(sessionBytes); err != nil {
 			log.Warnf("Invalid session ID in message: %v", err)
 			sessionID = nil // Set to nil if conversion fails
 		} else {
@@ -2775,8 +2776,8 @@ func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) {
 
 	relayIP := decodeRelayIP(msg.GetBody().GetRelayServerIP())
 
-	offerAnswer := peer.OfferAnswer{
-		IceCredentials: peer.IceCredentials{
+	offerAnswer := signaling.OfferAnswer{
+		IceCredentials: signaling.IceCredentials{
 			UFrag: remoteCred.UFrag,
 			Pwd:   remoteCred.Pwd,
 		},

client/internal/peer/conn_status.go

@@ -1,18 +1,5 @@
 package peer
 
-import (
-	log "github.com/sirupsen/logrus"
-)
-
-const (
-	// StatusIdle indicate the peer is in disconnected state
-	StatusIdle ConnStatus = iota
-	// StatusConnecting indicate the peer is in connecting state
-	StatusConnecting
-	// StatusConnected indicate the peer is in connected state
-	StatusConnected
-)
-
 // connStatusInputs is the primitive-valued snapshot of the state that drives the
 // tri-state connection classification. Extracted so the decision logic can be unit-tested
 // without constructing full Worker/Handshaker objects.
@@ -21,24 +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
 }
-
-// ConnStatus describe the status of a peer's connection
-type ConnStatus int32
-
-func (s ConnStatus) String() string {
-	switch s {
-	case StatusConnecting:
-		return "Connecting"
-	case StatusConnected:
-		return "Connected"
-	case StatusIdle:
-		return "Idle"
-	default:
-		log.Errorf("unknown status: %d", s)
-		return "INVALID_PEER_CONNECTION_STATUS"
-	}
-}

client/internal/peer/conn_test.go

@@ -3,28 +3,33 @@ package peer
 import (
 	"context"
 	"fmt"
+	"net/netip"
 	"os"
 	"testing"
 	"time"
 
 	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/netbirdio/netbird/client/iface"
-	"github.com/netbirdio/netbird/client/internal/peer/dispatcher"
 	"github.com/netbirdio/netbird/client/internal/peer/guard"
 	"github.com/netbirdio/netbird/client/internal/peer/ice"
+	"github.com/netbirdio/netbird/client/internal/peer/metricsstages"
+	"github.com/netbirdio/netbird/client/internal/peer/signaling"
+	"github.com/netbirdio/netbird/client/internal/peer/status"
 	"github.com/netbirdio/netbird/client/internal/stdnet"
 	"github.com/netbirdio/netbird/util"
 )
 
-var testDispatcher = dispatcher.NewConnectionDispatcher()
-
 var connConf = ConnConfig{
 	Key:         "LLHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=",
 	LocalKey:    "RRHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=",
 	Timeout:     time.Second,
 	LocalWgPort: 51820,
+	WgConfig: WgConfig{
+		AllowedIps: []netip.Prefix{netip.MustParsePrefix("100.64.0.1/32")},
+	},
 	ICEConfig: ice.Config{
 		InterfaceBlackList: nil,
 	},
@@ -52,92 +57,37 @@ func TestConn_GetKey(t *testing.T) {
 	swWatcher := guard.NewSRWatcher(nil, nil, nil, connConf.ICEConfig)
 
 	sd := ServiceDependencies{
-		SrWatcher:          swWatcher,
-		PeerConnDispatcher: testDispatcher,
+		SrWatcher: swWatcher,
 	}
 	conn, err := NewConn(connConf, sd)
-	if err != nil {
-		return
-	}
+	require.NoError(t, err)
 
 	got := conn.GetKey()
 
 	assert.Equal(t, got, connConf.Key, "they should be equal")
 }
 
-func TestConn_OnRemoteOffer(t *testing.T) {
+// TestConn_DiscardMessagesWhenNotOpened: signal messages posted to a not yet
+// opened connection must be discarded without blocking or panicking.
+func TestConn_DiscardMessagesWhenNotOpened(t *testing.T) {
 	swWatcher := guard.NewSRWatcher(nil, nil, nil, connConf.ICEConfig)
 	sd := ServiceDependencies{
-		StatusRecorder:     NewRecorder("https://mgm"),
-		SrWatcher:          swWatcher,
-		PeerConnDispatcher: testDispatcher,
+		StatusRecorder: status.NewRecorder("https://mgm"),
+		SrWatcher:      swWatcher,
 	}
 	conn, err := NewConn(connConf, sd)
-	if err != nil {
-		return
-	}
+	require.NoError(t, err)
 
-	onNewOfferChan := make(chan struct{})
-
-	conn.handshaker.AddRelayListener(func(remoteOfferAnswer *OfferAnswer) {
-		onNewOfferChan <- struct{}{}
-	})
-
-	conn.OnRemoteOffer(OfferAnswer{
-		IceCredentials: IceCredentials{
+	offerAnswer := signaling.OfferAnswer{
+		IceCredentials: signaling.IceCredentials{
 			UFrag: "test",
 			Pwd:   "test",
 		},
-		WgListenPort: 0,
-		Version:      "",
-	})
-
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	select {
-	case <-onNewOfferChan:
-		// success
-	case <-ctx.Done():
-		t.Error("expected to receive a new offer notification, but timed out")
-	}
-}
-
-func TestConn_OnRemoteAnswer(t *testing.T) {
-	swWatcher := guard.NewSRWatcher(nil, nil, nil, connConf.ICEConfig)
-	sd := ServiceDependencies{
-		StatusRecorder:     NewRecorder("https://mgm"),
-		SrWatcher:          swWatcher,
-		PeerConnDispatcher: testDispatcher,
-	}
-	conn, err := NewConn(connConf, sd)
-	if err != nil {
-		return
-	}
-
-	onNewOfferChan := make(chan struct{})
-
-	conn.handshaker.AddRelayListener(func(remoteOfferAnswer *OfferAnswer) {
-		onNewOfferChan <- struct{}{}
-	})
-
-	conn.OnRemoteAnswer(OfferAnswer{
-		IceCredentials: IceCredentials{
-			UFrag: "test",
-			Pwd:   "test",
-		},
-		WgListenPort: 0,
-		Version:      "",
-	})
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	select {
-	case <-onNewOfferChan:
-		// success
-	case <-ctx.Done():
-		t.Error("expected to receive a new offer notification, but timed out")
 	}
+	conn.OnRemoteOffer(offerAnswer)
+	conn.OnRemoteAnswer(offerAnswer)
+	conn.OnRemoteCandidate(nil, nil)
+	conn.Close(false)
 }
 
 func TestConn_presharedKey(t *testing.T) {
@@ -320,7 +270,7 @@ func newWGTimeoutTestConn(rosenpassEnabled bool, disconnected *[]string) *Conn {
 		ctx:           context.Background(),
 		config:        cfg,
 		Log:           log.WithField("peer", cfg.Key),
-		metricsStages: &MetricsStages{},
+		metricsStages: &metricsstages.MetricsStages{},
 	}
 	conn.SetOnDisconnected(func(remotePeer string) {
 		*disconnected = append(*disconnected, remotePeer)
@@ -339,20 +289,20 @@ func TestConn_onWGDisconnected_EscalatesToRosenpassReset(t *testing.T) {
 	conn := newWGTimeoutTestConn(true, &disconnected)
 
 	for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
-		conn.onWGDisconnected()
+		conn.handleWGTimeout()
 	}
 	assert.Empty(t, disconnected, "escalation must not fire below the threshold")
 
-	conn.onWGDisconnected()
+	conn.handleWGTimeout()
 	assert.Equal(t, []string{conn.config.WgConfig.RemoteKey}, disconnected,
 		"reaching the threshold must report the peer disconnected once")
 
 	for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
-		conn.onWGDisconnected()
+		conn.handleWGTimeout()
 	}
 	assert.Len(t, disconnected, 1, "escalation must restart counting after firing")
 
-	conn.onWGDisconnected()
+	conn.handleWGTimeout()
 	assert.Len(t, disconnected, 2, "continued timeouts must escalate again")
 }
 
@@ -364,12 +314,12 @@ func TestConn_onWGDisconnected_CheckSuccessResetsEscalation(t *testing.T) {
 	conn := newWGTimeoutTestConn(true, &disconnected)
 
 	for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
-		conn.onWGDisconnected()
+		conn.handleWGTimeout()
 	}
-	conn.onWGCheckSuccess()
+	conn.handleWGCheckSuccess()
 
 	for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
-		conn.onWGDisconnected()
+		conn.handleWGTimeout()
 	}
 	assert.Empty(t, disconnected, "handshake success must reset the timeout count")
 }
@@ -382,7 +332,7 @@ func TestConn_onWGDisconnected_NoEscalationWithoutRosenpass(t *testing.T) {
 	conn := newWGTimeoutTestConn(false, &disconnected)
 
 	for i := 0; i < wgTimeoutEscalationThreshold*3; i++ {
-		conn.onWGDisconnected()
+		conn.handleWGTimeout()
 	}
 	assert.Empty(t, disconnected, "escalation must be limited to rosenpass connections")
 }

client/internal/peer/dispatcher/dispatcher.go

@@ -1,52 +0,0 @@
-package dispatcher
-
-import (
-	"sync"
-
-	"github.com/netbirdio/netbird/client/internal/peer/id"
-)
-
-type ConnectionListener struct {
-	OnConnected    func(peerID id.ConnID)
-	OnDisconnected func(peerID id.ConnID)
-}
-
-type ConnectionDispatcher struct {
-	listeners map[*ConnectionListener]struct{}
-	mu        sync.Mutex
-}
-
-func NewConnectionDispatcher() *ConnectionDispatcher {
-	return &ConnectionDispatcher{
-		listeners: make(map[*ConnectionListener]struct{}),
-	}
-}
-
-func (e *ConnectionDispatcher) AddListener(listener *ConnectionListener) {
-	e.mu.Lock()
-	defer e.mu.Unlock()
-	e.listeners[listener] = struct{}{}
-}
-
-func (e *ConnectionDispatcher) RemoveListener(listener *ConnectionListener) {
-	e.mu.Lock()
-	defer e.mu.Unlock()
-
-	delete(e.listeners, listener)
-}
-
-func (e *ConnectionDispatcher) NotifyConnected(peerConnID id.ConnID) {
-	e.mu.Lock()
-	defer e.mu.Unlock()
-	for listener := range e.listeners {
-		listener.OnConnected(peerConnID)
-	}
-}
-
-func (e *ConnectionDispatcher) NotifyDisconnected(peerConnID id.ConnID) {
-	e.mu.Lock()
-	defer e.mu.Unlock()
-	for listener := range e.listeners {
-		listener.OnDisconnected(peerConnID)
-	}
-}

client/internal/peer/event.go

@@ -0,0 +1,69 @@
+package peer
+
+import (
+	"time"
+
+	"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"
+)
+
+// 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
+
+// evClose asks the event loop to tear down the connection. done is closed
+// once the teardown finished.
+type evClose struct {
+	signalToRemote bool
+	done           chan struct{}
+}
+
+type evRemoteOffer struct {
+	offer signaling.OfferAnswer
+}
+
+type evRemoteAnswer struct {
+	answer signaling.OfferAnswer
+}
+
+type evRemoteCandidate struct {
+	candidate ice.Candidate
+	haRoutes  route.HAMap
+}
+
+type evICEReady struct {
+	priority worker.ConnPriority
+	info     worker.ICEConnInfo
+}
+
+type evICEDown struct {
+	sessionChanged bool
+}
+
+type evRelayReady struct {
+	info worker.RelayConnInfo
+}
+
+type evRelayDown struct{}
+
+// evRelayDialDone reports that the relay dial helper goroutine finished,
+// successfully or not, so the loop may dispatch a pending offer.
+type evRelayDialDone struct{}
+
+type evWGTimeout struct{}
+
+// evWGHandshake reports the first WireGuard handshake of the current watcher run.
+type evWGHandshake struct {
+	when time.Time
+}
+
+// evWGCheckOK reports a watcher check that observed a fresh handshake,
+// including handshakes of connections that were already up.
+type evWGCheckOK struct{}
+
+// evGuardTick asks the loop to send a new offer to restore connectivity.
+type evGuardTick struct{}

client/internal/peer/guard/ice_monitor.go

@@ -21,8 +21,6 @@ const (
 )
 
 type ICEMonitor struct {
-	ReconnectCh chan struct{}
-
 	iFaceDiscover stdnet.ExternalIFaceDiscover
 	iceConfig     icemaker.Config
 	tickerPeriod  time.Duration
@@ -34,7 +32,6 @@ type ICEMonitor struct {
 func NewICEMonitor(iFaceDiscover stdnet.ExternalIFaceDiscover, config icemaker.Config, period time.Duration) *ICEMonitor {
 	log.Debugf("prepare ICE monitor with period: %s", period)
 	cm := &ICEMonitor{
-		ReconnectCh:   make(chan struct{}, 1),
 		iFaceDiscover: iFaceDiscover,
 		iceConfig:     config,
 		tickerPeriod:  period,

client/internal/peer/handshaker.go

@@ -1,246 +0,0 @@
-package peer
-
-import (
-	"context"
-	"errors"
-	"net/netip"
-	"sync"
-	"sync/atomic"
-
-	log "github.com/sirupsen/logrus"
-
-	"github.com/netbirdio/netbird/version"
-)
-
-var (
-	ErrSignalIsNotReady = errors.New("signal is not ready")
-)
-
-// IceCredentials ICE protocol credentials struct
-type IceCredentials struct {
-	UFrag string
-	Pwd   string
-}
-
-// OfferAnswer represents a session establishment offer or answer
-type OfferAnswer struct {
-	IceCredentials IceCredentials
-	// WgListenPort is a remote WireGuard listen port.
-	// This field is used when establishing a direct WireGuard connection without any proxy.
-	// We can set the remote peer's endpoint with this port.
-	WgListenPort int
-
-	// Version of NetBird Agent
-	Version string
-	// RosenpassPubKey is the Rosenpass public key of the remote peer when receiving this message
-	// This value is the local Rosenpass server public key when sending the message
-	RosenpassPubKey []byte
-	// RosenpassAddr is the Rosenpass server address (IP:port) of the remote peer when receiving this message
-	// This value is the local Rosenpass server address when sending the message
-	RosenpassAddr string
-
-	// relay server address
-	RelaySrvAddress string
-	// RelaySrvIP is the IP the remote peer is connected to on its
-	// relay server. Used as a dial target if DNS for RelaySrvAddress
-	// fails. Zero value if the peer did not advertise an IP.
-	RelaySrvIP netip.Addr
-	// SessionID is the unique identifier of the session, used to discard old messages
-	SessionID *ICESessionID
-}
-
-func (o *OfferAnswer) hasICECredentials() bool {
-	return o.IceCredentials.UFrag != "" && o.IceCredentials.Pwd != ""
-}
-
-type Handshaker struct {
-	mu            sync.Mutex
-	log           *log.Entry
-	config        ConnConfig
-	signaler      *Signaler
-	ice           *WorkerICE
-	relay         *WorkerRelay
-	metricsStages *MetricsStages
-	// relayListener is not blocking because the listener is using a goroutine to process the messages
-	// and it will only keep the latest message if multiple offers are received in a short time
-	// this is to avoid blocking the handshaker if the listener is doing some heavy processing
-	// and also to avoid processing old offers if multiple offers are received in a short time
-	// the listener will always process the latest offer
-	relayListener *AsyncOfferListener
-	iceListener   func(remoteOfferAnswer *OfferAnswer)
-
-	// remoteICESupported tracks whether the remote peer includes ICE credentials in its offers/answers.
-	// When false, the local side skips ICE listener dispatch and suppresses ICE credentials in responses.
-	remoteICESupported atomic.Bool
-
-	// remoteOffersCh is a channel used to wait for remote credentials to proceed with the connection
-	remoteOffersCh chan OfferAnswer
-	// remoteAnswerCh is a channel used to wait for remote credentials answer (confirmation of our offer) to proceed with the connection
-	remoteAnswerCh chan OfferAnswer
-}
-
-func NewHandshaker(log *log.Entry, config ConnConfig, signaler *Signaler, ice *WorkerICE, relay *WorkerRelay, metricsStages *MetricsStages) *Handshaker {
-	h := &Handshaker{
-		log:            log,
-		config:         config,
-		signaler:       signaler,
-		ice:            ice,
-		relay:          relay,
-		metricsStages:  metricsStages,
-		remoteOffersCh: make(chan OfferAnswer),
-		remoteAnswerCh: make(chan OfferAnswer),
-	}
-	// assume remote supports ICE until we learn otherwise from received offers
-	h.remoteICESupported.Store(ice != nil)
-	return h
-}
-
-func (h *Handshaker) RemoteICESupported() bool {
-	return h.remoteICESupported.Load()
-}
-
-func (h *Handshaker) AddRelayListener(offer func(remoteOfferAnswer *OfferAnswer)) {
-	h.relayListener = NewAsyncOfferListener(offer)
-}
-
-func (h *Handshaker) AddICEListener(offer func(remoteOfferAnswer *OfferAnswer)) {
-	h.iceListener = offer
-}
-
-func (h *Handshaker) Listen(ctx context.Context) {
-	for {
-		select {
-		case remoteOfferAnswer := <-h.remoteOffersCh:
-			h.log.Infof("received offer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", remoteOfferAnswer.Version, remoteOfferAnswer.WgListenPort, remoteOfferAnswer.SessionIDString(), remoteOfferAnswer.hasICECredentials())
-
-			// Record signaling received for reconnection attempts
-			if h.metricsStages != nil {
-				h.metricsStages.RecordSignalingReceived()
-			}
-
-			h.updateRemoteICEState(&remoteOfferAnswer)
-
-			if h.relayListener != nil {
-				h.relayListener.Notify(&remoteOfferAnswer)
-			}
-
-			if h.iceListener != nil && h.RemoteICESupported() {
-				h.iceListener(&remoteOfferAnswer)
-			}
-
-			if err := h.sendAnswer(); err != nil {
-				h.log.Errorf("failed to send remote offer confirmation: %s", err)
-				continue
-			}
-		case remoteOfferAnswer := <-h.remoteAnswerCh:
-			h.log.Infof("received answer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", remoteOfferAnswer.Version, remoteOfferAnswer.WgListenPort, remoteOfferAnswer.SessionIDString(), remoteOfferAnswer.hasICECredentials())
-
-			// Record signaling received for reconnection attempts
-			if h.metricsStages != nil {
-				h.metricsStages.RecordSignalingReceived()
-			}
-
-			h.updateRemoteICEState(&remoteOfferAnswer)
-
-			if h.relayListener != nil {
-				h.relayListener.Notify(&remoteOfferAnswer)
-			}
-
-			if h.iceListener != nil && h.RemoteICESupported() {
-				h.iceListener(&remoteOfferAnswer)
-			}
-		case <-ctx.Done():
-			h.log.Infof("stop listening for remote offers and answers")
-			return
-		}
-	}
-}
-
-func (h *Handshaker) SendOffer() error {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	return h.sendOffer()
-}
-
-// OnRemoteOffer handles an offer from the remote peer and returns true if the message was accepted, false otherwise
-// doesn't block, discards the message if connection wasn't ready
-func (h *Handshaker) OnRemoteOffer(offer OfferAnswer) {
-	select {
-	case h.remoteOffersCh <- offer:
-		return
-	default:
-		h.log.Warnf("skipping remote offer message because receiver not ready")
-		// connection might not be ready yet to receive so we ignore the message
-		return
-	}
-}
-
-// OnRemoteAnswer handles an offer from the remote peer and returns true if the message was accepted, false otherwise
-// doesn't block, discards the message if connection wasn't ready
-func (h *Handshaker) OnRemoteAnswer(answer OfferAnswer) {
-	select {
-	case h.remoteAnswerCh <- answer:
-		return
-	default:
-		// connection might not be ready yet to receive so we ignore the message
-		h.log.Warnf("skipping remote answer message because receiver not ready")
-		return
-	}
-}
-
-// sendOffer prepares local user credentials and signals them to the remote peer
-func (h *Handshaker) sendOffer() error {
-	if !h.signaler.Ready() {
-		return ErrSignalIsNotReady
-	}
-
-	offer := h.buildOfferAnswer()
-	h.log.Debugf("sending offer with serial: %s", offer.SessionIDString())
-
-	return h.signaler.SignalOffer(offer, h.config.Key)
-}
-
-func (h *Handshaker) sendAnswer() error {
-	answer := h.buildOfferAnswer()
-	h.log.Debugf("sending answer with serial: %s", answer.SessionIDString())
-
-	return h.signaler.SignalAnswer(answer, h.config.Key)
-}
-
-func (h *Handshaker) buildOfferAnswer() OfferAnswer {
-	answer := OfferAnswer{
-		WgListenPort:    h.config.LocalWgPort,
-		Version:         version.NetbirdVersion(),
-		RosenpassPubKey: h.config.RosenpassConfig.PubKey,
-		RosenpassAddr:   h.config.RosenpassConfig.Addr,
-	}
-
-	if h.ice != nil && h.RemoteICESupported() {
-		uFrag, pwd := h.ice.GetLocalUserCredentials()
-		sid := h.ice.SessionID()
-		answer.IceCredentials = IceCredentials{uFrag, pwd}
-		answer.SessionID = &sid
-	}
-
-	if addr, ip, err := h.relay.RelayInstanceAddress(); err == nil {
-		answer.RelaySrvAddress = addr
-		answer.RelaySrvIP = ip
-	}
-
-	return answer
-}
-
-func (h *Handshaker) updateRemoteICEState(offer *OfferAnswer) {
-	hasICE := offer.hasICECredentials()
-	prev := h.remoteICESupported.Swap(hasICE)
-	if prev != hasICE {
-		if hasICE {
-			h.log.Infof("remote peer started sending ICE credentials")
-		} else {
-			h.log.Infof("remote peer stopped sending ICE credentials")
-			if h.ice != nil {
-				h.ice.Close()
-			}
-		}
-	}
-}

client/internal/peer/handshaker_listener.go

@@ -1,62 +0,0 @@
-package peer
-
-import (
-	"sync"
-)
-
-type callbackFunc func(remoteOfferAnswer *OfferAnswer)
-
-func (oa *OfferAnswer) SessionIDString() string {
-	if oa.SessionID == nil {
-		return "unknown"
-	}
-	return oa.SessionID.String()
-}
-
-type AsyncOfferListener struct {
-	fn      callbackFunc
-	running bool
-	latest  *OfferAnswer
-	mu      sync.Mutex
-}
-
-func NewAsyncOfferListener(fn callbackFunc) *AsyncOfferListener {
-	return &AsyncOfferListener{
-		fn: fn,
-	}
-}
-
-func (o *AsyncOfferListener) Notify(remoteOfferAnswer *OfferAnswer) {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-
-	// Store the latest offer
-	o.latest = remoteOfferAnswer
-
-	// If already running, the running goroutine will pick up this latest value
-	if o.running {
-		return
-	}
-
-	// Start processing
-	o.running = true
-
-	// Process in a goroutine to avoid blocking the caller
-	go func(remoteOfferAnswer *OfferAnswer) {
-		for {
-			o.fn(remoteOfferAnswer)
-
-			o.mu.Lock()
-			if o.latest == nil {
-				// No more work to do
-				o.running = false
-				o.mu.Unlock()
-				return
-			}
-			remoteOfferAnswer = o.latest
-			// Clear the latest to mark it as being processed
-			o.latest = nil
-			o.mu.Unlock()
-		}
-	}(remoteOfferAnswer)
-}

client/internal/peer/handshaker_listener_test.go

@@ -1,39 +0,0 @@
-package peer
-
-import (
-	"testing"
-	"time"
-)
-
-func Test_newOfferListener(t *testing.T) {
-	dummyOfferAnswer := &OfferAnswer{}
-	runChan := make(chan struct{}, 10)
-
-	longRunningFn := func(remoteOfferAnswer *OfferAnswer) {
-		time.Sleep(1 * time.Second)
-		runChan <- struct{}{}
-	}
-
-	hl := NewAsyncOfferListener(longRunningFn)
-
-	hl.Notify(dummyOfferAnswer)
-	hl.Notify(dummyOfferAnswer)
-	hl.Notify(dummyOfferAnswer)
-
-	// Wait for exactly 2 callbacks
-	for i := 0; i < 2; i++ {
-		select {
-		case <-runChan:
-		case <-time.After(3 * time.Second):
-			t.Fatal("Timeout waiting for callback")
-		}
-	}
-
-	// Verify no additional callbacks happen
-	select {
-	case <-runChan:
-		t.Fatal("Unexpected additional callback")
-	case <-time.After(100 * time.Millisecond):
-		t.Log("Correctly received exactly 2 callbacks")
-	}
-}

client/internal/peer/ice/session_id.go

@@ -1,4 +1,4 @@
-package peer
+package ice
 
 import (
 	"crypto/rand"
@@ -9,26 +9,26 @@ import (
 
 const sessionIDSize = 5
 
-type ICESessionID string
+type SessionID string
 
-// NewICESessionID generates a new session ID for distinguishing sessions
-func NewICESessionID() (ICESessionID, error) {
+// NewSessionID generates a new session ID for distinguishing sessions
+func NewSessionID() (SessionID, error) {
 	b := make([]byte, sessionIDSize)
 	if _, err := io.ReadFull(rand.Reader, b); err != nil {
 		return "", fmt.Errorf("failed to generate session ID: %w", err)
 	}
-	return ICESessionID(hex.EncodeToString(b)), nil
+	return SessionID(hex.EncodeToString(b)), nil
 }
 
-func ICESessionIDFromBytes(b []byte) (ICESessionID, error) {
+func SessionIDFromBytes(b []byte) (SessionID, error) {
 	if len(b) != sessionIDSize {
 		return "", fmt.Errorf("invalid session ID length: %d", len(b))
 	}
-	return ICESessionID(hex.EncodeToString(b)), nil
+	return SessionID(hex.EncodeToString(b)), nil
 }
 
 // Bytes returns the raw bytes of the session ID for protobuf serialization
-func (id ICESessionID) Bytes() ([]byte, error) {
+func (id SessionID) Bytes() ([]byte, error) {
 	if len(id) == 0 {
 		return nil, fmt.Errorf("ICE session ID is empty")
 	}
@@ -42,6 +42,6 @@ func (id ICESessionID) Bytes() ([]byte, error) {
 	return b, nil
 }
 
-func (id ICESessionID) String() string {
+func (id SessionID) String() string {
 	return string(id)
 }

client/internal/peer/iface.go

@@ -1,22 +0,0 @@
-package peer
-
-import (
-	"net"
-	"net/netip"
-	"time"
-
-	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
-
-	"github.com/netbirdio/netbird/client/iface/configurer"
-	"github.com/netbirdio/netbird/client/iface/wgaddr"
-	"github.com/netbirdio/netbird/client/iface/wgproxy"
-)
-
-type WGIface interface {
-	UpdatePeer(peerKey string, allowedIps []netip.Prefix, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error
-	RemovePeer(peerKey string) error
-	GetStats() (map[string]configurer.WGStats, error)
-	GetProxy() wgproxy.Proxy
-	Address() wgaddr.Address
-	RemoveEndpointAddress(key string) error
-}

client/internal/peer/listener.go

@@ -1,11 +0,0 @@
-package peer
-
-// Listener is a callback type about the NetBird network connection state
-type Listener interface {
-	OnConnected()
-	OnDisconnected()
-	OnConnecting()
-	OnDisconnecting()
-	OnAddressChanged(string, string)
-	OnPeersListChanged(int)
-}

client/internal/peer/mailbox.go

@@ -0,0 +1,116 @@
+package peer
+
+import (
+	"sync"
+)
+
+// maxQueuedCandidates bounds the remote candidate queue; on overflow the
+// oldest candidate is dropped. Lost candidates are recovered by the next
+// offer exchange triggered by the guard.
+const maxQueuedCandidates = 128
+
+// mailbox is the coalescing inbox of the Conn event loop. Posting never
+// blocks. Per message kind either the latest value wins (offer, answer,
+// guard tick), the values queue in bounded FIFO order (candidates) or in
+// unbounded FIFO order (lifecycle and transport state changes, which are
+// low-volume and must not be lost). A new offer flushes the queued
+// candidates because they belong to the superseded session.
+type mailbox struct {
+	mu     sync.Mutex
+	closed bool
+
+	lifecycle  []event
+	transport  []event
+	offer      *evRemoteOffer
+	answer     *evRemoteAnswer
+	candidates []evRemoteCandidate
+	guardTick  bool
+
+	wake chan struct{}
+}
+
+func newMailbox() *mailbox {
+	return &mailbox{
+		wake: make(chan struct{}, 1),
+	}
+}
+
+// post stores the event and wakes the loop. It reports false if the mailbox
+// is already closed and the event was not accepted.
+func (m *mailbox) post(ev event) bool {
+	m.mu.Lock()
+	if m.closed {
+		m.mu.Unlock()
+		return false
+	}
+
+	switch e := ev.(type) {
+	case evClose:
+		m.lifecycle = append(m.lifecycle, e)
+	case evRemoteOffer:
+		m.offer = &e
+		m.candidates = nil
+	case evRemoteAnswer:
+		m.answer = &e
+	case evRemoteCandidate:
+		if len(m.candidates) >= maxQueuedCandidates {
+			m.candidates = m.candidates[1:]
+		}
+		m.candidates = append(m.candidates, e)
+	case evGuardTick:
+		m.guardTick = true
+	default:
+		m.transport = append(m.transport, ev)
+	}
+	m.mu.Unlock()
+
+	select {
+	case m.wake <- struct{}{}:
+	default:
+	}
+	return true
+}
+
+// drain returns the pending events in processing order: lifecycle first,
+// then transport state changes, the coalesced offer and answer, the queued
+// candidates and finally the guard tick.
+func (m *mailbox) drain() []event {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.drainLocked()
+}
+
+// closeAndDrain marks the mailbox closed so further posts are rejected and
+// returns the events that were still pending.
+func (m *mailbox) closeAndDrain() []event {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.closed = true
+	return m.drainLocked()
+}
+
+func (m *mailbox) drainLocked() []event {
+	evs := make([]event, 0, len(m.lifecycle)+len(m.transport)+len(m.candidates)+3)
+	evs = append(evs, m.lifecycle...)
+	evs = append(evs, m.transport...)
+	if m.offer != nil {
+		evs = append(evs, *m.offer)
+	}
+	if m.answer != nil {
+		evs = append(evs, *m.answer)
+	}
+	for _, c := range m.candidates {
+		evs = append(evs, c)
+	}
+	if m.guardTick {
+		evs = append(evs, evGuardTick{})
+	}
+
+	m.lifecycle = nil
+	m.transport = nil
+	m.offer = nil
+	m.answer = nil
+	m.candidates = nil
+	m.guardTick = false
+	return evs
+}

client/internal/peer/mailbox_test.go

@@ -0,0 +1,128 @@
+package peer
+
+import (
+	"testing"
+
+	"github.com/netbirdio/netbird/client/internal/peer/signaling"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMailbox_OfferCoalescing(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 1}}))
+	require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 2}}))
+	require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 3}}))
+
+	evs := mb.drain()
+	require.Len(t, evs, 1, "consecutive offers must coalesce to a single event")
+	offer, ok := evs[0].(evRemoteOffer)
+	require.True(t, ok, "coalesced event must be an offer")
+	assert.Equal(t, 3, offer.offer.WgListenPort, "the newest offer must win")
+}
+
+func TestMailbox_OfferFlushesCandidates(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evRemoteCandidate{}))
+	require.True(t, mb.post(evRemoteCandidate{}))
+	require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{}}))
+
+	evs := mb.drain()
+	require.Len(t, evs, 1, "candidates of the superseded session must be flushed")
+	_, ok := evs[0].(evRemoteOffer)
+	assert.True(t, ok, "only the offer must remain after the flush")
+}
+
+func TestMailbox_CandidatesKeepOrderAfterOffer(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{}}))
+	require.True(t, mb.post(evRemoteCandidate{haRoutes: nil}))
+	require.True(t, mb.post(evRemoteCandidate{haRoutes: nil}))
+
+	evs := mb.drain()
+	require.Len(t, evs, 3)
+	_, ok := evs[0].(evRemoteOffer)
+	assert.True(t, ok, "offer must be processed before the candidates")
+	for _, ev := range evs[1:] {
+		_, ok := ev.(evRemoteCandidate)
+		assert.True(t, ok, "candidates posted after the offer must survive")
+	}
+}
+
+func TestMailbox_CandidateQueueBounded(t *testing.T) {
+	mb := newMailbox()
+
+	for i := 0; i < maxQueuedCandidates+10; i++ {
+		require.True(t, mb.post(evRemoteCandidate{}))
+	}
+
+	evs := mb.drain()
+	assert.Len(t, evs, maxQueuedCandidates, "candidate queue must stay bounded")
+}
+
+func TestMailbox_DrainOrder(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evGuardTick{}))
+	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(evClose{}))
+
+	evs := mb.drain()
+	require.Len(t, evs, 6)
+
+	_, ok := evs[0].(evClose)
+	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)
+	assert.True(t, ok, "transport events must keep FIFO order")
+	_, ok = evs[3].(evRemoteOffer)
+	assert.True(t, ok, "offer must come after transport events")
+	_, ok = evs[4].(evRemoteAnswer)
+	assert.True(t, ok, "answer must come after the offer")
+	_, ok = evs[5].(evGuardTick)
+	assert.True(t, ok, "guard tick must come last")
+}
+
+func TestMailbox_GuardTickCoalesced(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evGuardTick{}))
+	require.True(t, mb.post(evGuardTick{}))
+	require.True(t, mb.post(evGuardTick{}))
+
+	evs := mb.drain()
+	assert.Len(t, evs, 1, "guard ticks must coalesce to a single event")
+}
+
+func TestMailbox_PostAfterCloseRejected(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evRelayDown{}))
+	leftovers := mb.closeAndDrain()
+	assert.Len(t, leftovers, 1, "pending events must be returned on close")
+
+	assert.False(t, mb.post(evRelayDown{}), "posts must be rejected after close")
+	assert.Empty(t, mb.drain(), "no events must remain after close")
+}
+
+func TestMailbox_WakeSignal(t *testing.T) {
+	mb := newMailbox()
+
+	require.True(t, mb.post(evRelayDown{}))
+	require.True(t, mb.post(evGuardTick{}))
+
+	select {
+	case <-mb.wake:
+	default:
+		t.Fatal("wake signal must be pending after posts")
+	}
+
+	assert.Len(t, mb.drain(), 2, "a single wake must deliver all pending events")
+}

client/internal/peer/metricsstages/metrics_saver.go

@@ -1,4 +1,4 @@
-package peer
+package metricsstages
 
 import (
 	"sync"

client/internal/peer/metricsstages/metrics_saver_test.go

@@ -1,4 +1,4 @@
-package peer
+package metricsstages
 
 import (
 	"testing"

client/internal/peer/signaling/handshaker.go

@@ -0,0 +1,189 @@
+package signaling
+
+import (
+	"errors"
+	"net/netip"
+	"sync"
+	"sync/atomic"
+
+	log "github.com/sirupsen/logrus"
+
+	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
+	relayClient "github.com/netbirdio/netbird/shared/relay/client"
+	"github.com/netbirdio/netbird/version"
+)
+
+var (
+	ErrSignalIsNotReady = errors.New("signal is not ready")
+)
+
+// IceCredentials ICE protocol credentials struct
+type IceCredentials struct {
+	UFrag string
+	Pwd   string
+}
+
+// OfferAnswer represents a session establishment offer or answer
+type OfferAnswer struct {
+	IceCredentials IceCredentials
+	// WgListenPort is a remote WireGuard listen port.
+	// This field is used when establishing a direct WireGuard connection without any proxy.
+	// We can set the remote peer's endpoint with this port.
+	WgListenPort int
+
+	// Version of NetBird Agent
+	Version string
+	// RosenpassPubKey is the Rosenpass public key of the remote peer when receiving this message
+	// This value is the local Rosenpass server public key when sending the message
+	RosenpassPubKey []byte
+	// RosenpassAddr is the Rosenpass server address (IP:port) of the remote peer when receiving this message
+	// This value is the local Rosenpass server address when sending the message
+	RosenpassAddr string
+
+	// relay server address
+	RelaySrvAddress string
+	// RelaySrvIP is the IP the remote peer is connected to on its
+	// relay server. Used as a dial target if DNS for RelaySrvAddress
+	// fails. Zero value if the peer did not advertise an IP.
+	RelaySrvIP netip.Addr
+	// SessionID is the unique identifier of the session, used to discard old messages
+	SessionID *icemaker.SessionID
+}
+
+func (o *OfferAnswer) HasICECredentials() bool {
+	return o.IceCredentials.UFrag != "" && o.IceCredentials.Pwd != ""
+}
+
+func (o *OfferAnswer) SessionIDString() string {
+	if o.SessionID == nil {
+		return "unknown"
+	}
+	return o.SessionID.String()
+}
+
+// Config carries the peer-specific values the Handshaker embeds into offers
+// and answers.
+type Config struct {
+	Key             string
+	LocalWgPort     int
+	RosenpassPubKey []byte
+	RosenpassAddr   string
+}
+
+// Credentials are the local ICE credentials and session id the Handshaker embeds in offers.
+type Credentials struct {
+	UFrag     string
+	Pwd       string
+	SessionID icemaker.SessionID
+}
+
+// ICEWorker is the subset of the ICE worker the Handshaker needs to build offers.
+type ICEWorker interface {
+	Credentials() Credentials
+	Close()
+}
+
+// Handshaker keeps the signaling protocol logic: building and sending offers
+// and answers and tracking whether the remote peer supports ICE. Incoming
+// message processing is driven by the Conn event loop.
+type Handshaker struct {
+	mu           sync.Mutex
+	log          *log.Entry
+	config       Config
+	signaler     *Signaler
+	ice          ICEWorker
+	relayManager *relayClient.Manager
+
+	// remoteICESupported tracks whether the remote peer includes ICE credentials in its offers/answers.
+	// When false, the local side skips ICE dispatch and suppresses ICE credentials in responses.
+	remoteICESupported atomic.Bool
+}
+
+func NewHandshaker(log *log.Entry, config Config, signaler *Signaler, ice ICEWorker, relayManager *relayClient.Manager) *Handshaker {
+	h := &Handshaker{
+		log:          log,
+		config:       config,
+		signaler:     signaler,
+		ice:          ice,
+		relayManager: relayManager,
+	}
+	// assume remote supports ICE until we learn otherwise from received offers
+	h.remoteICESupported.Store(ice != nil)
+	return h
+}
+
+func (h *Handshaker) RemoteICESupported() bool {
+	return h.remoteICESupported.Load()
+}
+
+func (h *Handshaker) SendOffer() error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.sendOffer()
+}
+
+func (h *Handshaker) SendAnswer() error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.sendAnswer()
+}
+
+// sendOffer prepares local user credentials and signals them to the remote peer
+func (h *Handshaker) sendOffer() error {
+	if !h.signaler.Ready() {
+		return ErrSignalIsNotReady
+	}
+
+	offer := h.buildOfferAnswer()
+	h.log.Debugf("sending offer with serial: %s", offer.SessionIDString())
+
+	return h.signaler.SignalOffer(offer, h.config.Key)
+}
+
+func (h *Handshaker) sendAnswer() error {
+	answer := h.buildOfferAnswer()
+	h.log.Debugf("sending answer with serial: %s", answer.SessionIDString())
+
+	return h.signaler.SignalAnswer(answer, h.config.Key)
+}
+
+func (h *Handshaker) buildOfferAnswer() OfferAnswer {
+	answer := OfferAnswer{
+		WgListenPort:    h.config.LocalWgPort,
+		Version:         version.NetbirdVersion(),
+		RosenpassPubKey: h.config.RosenpassPubKey,
+		RosenpassAddr:   h.config.RosenpassAddr,
+	}
+
+	if h.ice != nil && h.RemoteICESupported() {
+		creds := h.ice.Credentials()
+		answer.IceCredentials = IceCredentials{creds.UFrag, creds.Pwd}
+		sid := creds.SessionID
+		answer.SessionID = &sid
+	}
+
+	if addr, ip, err := h.relayManager.RelayInstanceAddress(); err == nil {
+		answer.RelaySrvAddress = addr
+		answer.RelaySrvIP = ip
+	}
+
+	return answer
+}
+
+// UpdateRemoteICEState refreshes the remote ICE support flag from a received
+// offer or answer and closes the ICE worker when the remote peer stopped
+// sending ICE credentials. Runs on the Conn event loop.
+func (h *Handshaker) UpdateRemoteICEState(offer *OfferAnswer) {
+	hasICE := offer.HasICECredentials()
+	prev := h.remoteICESupported.Swap(hasICE)
+	if prev != hasICE {
+		if hasICE {
+			h.log.Infof("remote peer started sending ICE credentials")
+		} else {
+			h.log.Infof("remote peer stopped sending ICE credentials")
+			if h.ice != nil {
+				h.ice.Close()
+			}
+		}
+	}
+}

client/internal/peer/signaling/signaler.go

@@ -1,4 +1,4 @@
-package peer
+package signaling
 
 import (
 	"github.com/pion/ice/v4"

client/internal/peer/state_dump/state_dump.go

@@ -1,4 +1,4 @@
-package peer
+package state_dump
 
 import (
 	"context"
@@ -6,11 +6,13 @@ import (
 	"time"
 
 	log "github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/client/internal/peer/status"
 )
 
-type stateDump struct {
+type StateDump struct {
 	log    *log.Entry
-	status *Status
+	status *status.Recorder
 	key    string
 
 	sentOffer       int
@@ -26,15 +28,15 @@ type stateDump struct {
 	mu sync.Mutex
 }
 
-func newStateDump(key string, log *log.Entry, statusRecorder *Status) *stateDump {
-	return &stateDump{
+func NewStateDump(key string, log *log.Entry, statusRecorder *status.Recorder) *StateDump {
+	return &StateDump{
 		log:    log,
 		status: statusRecorder,
 		key:    key,
 	}
 }
 
-func (s *stateDump) Start(ctx context.Context) {
+func (s *StateDump) Start(ctx context.Context) {
 	ticker := time.NewTicker(10 * time.Minute)
 	defer ticker.Stop()
 
@@ -48,25 +50,25 @@ func (s *stateDump) Start(ctx context.Context) {
 	}
 }
 
-func (s *stateDump) RemoteOffer() {
+func (s *StateDump) RemoteOffer() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.remoteOffer++
 }
 
-func (s *stateDump) RemoteCandidate() {
+func (s *StateDump) RemoteCandidate() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.remoteCandidate++
 }
 
-func (s *stateDump) SendOffer() {
+func (s *StateDump) SendOffer() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.sentOffer++
 }
 
-func (s *stateDump) dumpState() {
+func (s *StateDump) dumpState() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -80,41 +82,41 @@ func (s *stateDump) dumpState() {
 		status, s.sentOffer, s.remoteOffer, s.remoteAnswer, s.remoteCandidate, s.p2pConnected, s.switchToRelay, s.wgCheckSuccess, s.relayConnected, s.localProxies)
 }
 
-func (s *stateDump) RemoteAnswer() {
+func (s *StateDump) RemoteAnswer() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.remoteAnswer++
 }
 
-func (s *stateDump) P2PConnected() {
+func (s *StateDump) P2PConnected() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.p2pConnected++
 }
 
-func (s *stateDump) SwitchToRelay() {
+func (s *StateDump) SwitchToRelay() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.switchToRelay++
 }
 
-func (s *stateDump) WGcheckSuccess() {
+func (s *StateDump) WGcheckSuccess() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.wgCheckSuccess++
 }
 
-func (s *stateDump) RelayConnected() {
+func (s *StateDump) RelayConnected() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.relayConnected++
 }
 
-func (s *stateDump) NewLocalProxy() {
+func (s *StateDump) NewLocalProxy() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 

client/internal/peer/status/conn_status.go

@@ -0,0 +1,31 @@
+package status
+
+import (
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	// StatusIdle indicate the peer is in disconnected state
+	StatusIdle ConnStatus = iota
+	// StatusConnecting indicate the peer is in connecting state
+	StatusConnecting
+	// StatusConnected indicate the peer is in connected state
+	StatusConnected
+)
+
+// ConnStatus describe the status of a peer's connection
+type ConnStatus int32
+
+func (s ConnStatus) String() string {
+	switch s {
+	case StatusConnecting:
+		return "Connecting"
+	case StatusConnected:
+		return "Connected"
+	case StatusIdle:
+		return "Idle"
+	default:
+		log.Errorf("unknown status: %d", s)
+		return "INVALID_PEER_CONNECTION_STATUS"
+	}
+}

client/internal/peer/status/conn_status_test.go

@@ -1,4 +1,4 @@
-package peer
+package status
 
 import (
 	"testing"

client/internal/peer/status/events.go

@@ -0,0 +1,48 @@
+package status
+
+import (
+	"slices"
+	"sync"
+
+	"github.com/netbirdio/netbird/client/proto"
+)
+
+type EventQueue struct {
+	maxSize int
+	events  []*proto.SystemEvent
+	mutex   sync.RWMutex
+}
+
+func NewEventQueue(size int) *EventQueue {
+	return &EventQueue{
+		maxSize: size,
+		events:  make([]*proto.SystemEvent, 0, size),
+	}
+}
+
+func (q *EventQueue) Add(event *proto.SystemEvent) {
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+
+	q.events = append(q.events, event)
+
+	if len(q.events) > q.maxSize {
+		q.events = q.events[len(q.events)-q.maxSize:]
+	}
+}
+
+func (q *EventQueue) GetAll() []*proto.SystemEvent {
+	q.mutex.RLock()
+	defer q.mutex.RUnlock()
+
+	return slices.Clone(q.events)
+}
+
+type EventSubscription struct {
+	id     string
+	events chan *proto.SystemEvent
+}
+
+func (s *EventSubscription) Events() <-chan *proto.SystemEvent {
+	return s.events
+}

client/internal/peer/status/full_status.go

@@ -0,0 +1,122 @@
+package status
+
+import (
+	"golang.org/x/exp/maps"
+	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/netbirdio/netbird/client/internal/relay"
+	"github.com/netbirdio/netbird/client/proto"
+)
+
+// FullStatus contains the full state held by the Recorder instance
+type FullStatus struct {
+	Peers                 []State
+	ManagementState       ManagementState
+	SignalState           SignalState
+	LocalPeerState        LocalPeerState
+	RosenpassState        RosenpassState
+	Relays                []relay.ProbeResult
+	NSGroupStates         []NSGroupState
+	NumOfForwardingRules  int
+	LazyConnectionEnabled bool
+	Events                []*proto.SystemEvent
+}
+
+// ToProto converts FullStatus to proto.FullStatus.
+func (fs FullStatus) ToProto() *proto.FullStatus {
+	pbFullStatus := proto.FullStatus{
+		ManagementState: &proto.ManagementState{},
+		SignalState:     &proto.SignalState{},
+		LocalPeerState:  &proto.LocalPeerState{},
+		Peers:           []*proto.PeerState{},
+	}
+
+	pbFullStatus.ManagementState.URL = fs.ManagementState.URL
+	pbFullStatus.ManagementState.Connected = fs.ManagementState.Connected
+	if err := fs.ManagementState.Error; err != nil {
+		pbFullStatus.ManagementState.Error = err.Error()
+	}
+
+	pbFullStatus.SignalState.URL = fs.SignalState.URL
+	pbFullStatus.SignalState.Connected = fs.SignalState.Connected
+	if err := fs.SignalState.Error; err != nil {
+		pbFullStatus.SignalState.Error = err.Error()
+	}
+
+	pbFullStatus.LocalPeerState.IP = fs.LocalPeerState.IP
+	pbFullStatus.LocalPeerState.Ipv6 = fs.LocalPeerState.IPv6
+	pbFullStatus.LocalPeerState.PubKey = fs.LocalPeerState.PubKey
+	pbFullStatus.LocalPeerState.KernelInterface = fs.LocalPeerState.KernelInterface
+	pbFullStatus.LocalPeerState.Fqdn = fs.LocalPeerState.FQDN
+	pbFullStatus.LocalPeerState.WgPort = int32(fs.LocalPeerState.WgPort)
+	pbFullStatus.LocalPeerState.RosenpassPermissive = fs.RosenpassState.Permissive
+	pbFullStatus.LocalPeerState.RosenpassEnabled = fs.RosenpassState.Enabled
+	pbFullStatus.NumberOfForwardingRules = int32(fs.NumOfForwardingRules)
+	pbFullStatus.LazyConnectionEnabled = fs.LazyConnectionEnabled
+
+	pbFullStatus.LocalPeerState.Networks = maps.Keys(fs.LocalPeerState.Routes)
+
+	for _, peerState := range fs.Peers {
+		networks := maps.Keys(peerState.GetRoutes())
+
+		pbPeerState := &proto.PeerState{
+			IP:                         peerState.IP,
+			Ipv6:                       peerState.IPv6,
+			PubKey:                     peerState.PubKey,
+			ConnStatus:                 peerState.ConnStatus.String(),
+			ConnStatusUpdate:           timestamppb.New(peerState.ConnStatusUpdate),
+			Relayed:                    peerState.Relayed,
+			LocalIceCandidateType:      peerState.LocalIceCandidateType,
+			RemoteIceCandidateType:     peerState.RemoteIceCandidateType,
+			LocalIceCandidateEndpoint:  peerState.LocalIceCandidateEndpoint,
+			RemoteIceCandidateEndpoint: peerState.RemoteIceCandidateEndpoint,
+			RelayAddress:               peerState.RelayServerAddress,
+			Fqdn:                       peerState.FQDN,
+			LastWireguardHandshake:     timestamppb.New(peerState.LastWireguardHandshake),
+			BytesRx:                    peerState.BytesRx,
+			BytesTx:                    peerState.BytesTx,
+			RosenpassEnabled:           peerState.RosenpassEnabled,
+			Networks:                   networks,
+			Latency:                    durationpb.New(peerState.Latency),
+			SshHostKey:                 peerState.SSHHostKey,
+		}
+		pbFullStatus.Peers = append(pbFullStatus.Peers, pbPeerState)
+	}
+
+	for _, relayState := range fs.Relays {
+		pbRelayState := &proto.RelayState{
+			URI:       relayState.URI,
+			Available: relayState.Err == nil,
+			Transport: relayState.Transport,
+		}
+		if err := relayState.Err; err != nil {
+			pbRelayState.Error = err.Error()
+		}
+		pbFullStatus.Relays = append(pbFullStatus.Relays, pbRelayState)
+	}
+
+	for _, dnsState := range fs.NSGroupStates {
+		var err string
+		if dnsState.Error != nil {
+			err = dnsState.Error.Error()
+		}
+
+		var servers []string
+		for _, server := range dnsState.Servers {
+			servers = append(servers, server.String())
+		}
+
+		pbDnsState := &proto.NSGroupState{
+			Servers: servers,
+			Domains: dnsState.Domains,
+			Enabled: dnsState.Enabled,
+			Error:   err,
+		}
+		pbFullStatus.DnsServers = append(pbFullStatus.DnsServers, pbDnsState)
+	}
+
+	pbFullStatus.Events = fs.Events
+
+	return &pbFullStatus
+}

client/internal/peer/status/notifier.go

@@ -1,4 +1,4 @@
-package peer
+package status
 
 import (
 	"sync"
@@ -11,6 +11,16 @@ const (
 	stateDisconnecting
 )
 
+// Listener is a callback type about the NetBird network connection state
+type Listener interface {
+	OnConnected()
+	OnDisconnected()
+	OnConnecting()
+	OnDisconnecting()
+	OnAddressChanged(string, string)
+	OnPeersListChanged(int)
+}
+
 type notifier struct {
 	serverStateLock    sync.Mutex
 	listenersLock      sync.Mutex

client/internal/peer/status/notifier_test.go

@@ -1,4 +1,4 @@
-package peer
+package status
 
 import (
 	"sync"

client/internal/peer/status/peer_state.go

@@ -0,0 +1,63 @@
+package status
+
+import (
+	"sync"
+	"time"
+
+	"golang.org/x/exp/maps"
+)
+
+// State contains the latest state of a peer
+type State struct {
+	Mux                        *sync.RWMutex
+	IP                         string
+	IPv6                       string
+	PubKey                     string
+	FQDN                       string
+	ConnStatus                 ConnStatus
+	ConnStatusUpdate           time.Time
+	Relayed                    bool
+	LocalIceCandidateType      string
+	RemoteIceCandidateType     string
+	LocalIceCandidateEndpoint  string
+	RemoteIceCandidateEndpoint string
+	RelayServerAddress         string
+	LastWireguardHandshake     time.Time
+	BytesTx                    int64
+	BytesRx                    int64
+	Latency                    time.Duration
+	RosenpassEnabled           bool
+	SSHHostKey                 []byte
+	routes                     map[string]struct{}
+}
+
+// AddRoute add a single route to routes map
+func (s *State) AddRoute(network string) {
+	s.Mux.Lock()
+	defer s.Mux.Unlock()
+	if s.routes == nil {
+		s.routes = make(map[string]struct{})
+	}
+	s.routes[network] = struct{}{}
+}
+
+// SetRoutes set state routes
+func (s *State) SetRoutes(routes map[string]struct{}) {
+	s.Mux.Lock()
+	defer s.Mux.Unlock()
+	s.routes = routes
+}
+
+// DeleteRoute removes a route from the network amp
+func (s *State) DeleteRoute(network string) {
+	s.Mux.Lock()
+	defer s.Mux.Unlock()
+	delete(s.routes, network)
+}
+
+// GetRoutes return routes map
+func (s *State) GetRoutes() map[string]struct{} {
+	s.Mux.RLock()
+	defer s.Mux.RUnlock()
+	return maps.Clone(s.routes)
+}

client/internal/peer/status/recorder_test.go

@@ -1,4 +1,4 @@
-package peer
+package status
 
 import (
 	"context"

client/internal/peer/status/route.go

@@ -1,4 +1,4 @@
-package peer
+package status
 
 import (
 	"net/netip"

client/internal/peer/status_alias.go

@@ -0,0 +1,36 @@
+package peer
+
+import "github.com/netbirdio/netbird/client/internal/peer/status"
+
+// Transitional aliases re-exporting the peer status recorder from its own
+// package. Callers are being migrated to reference the status package
+// directly; these aliases will be removed once the migration completes.
+type (
+	Status                   = status.Recorder
+	State                    = status.State
+	ConnStatus               = status.ConnStatus
+	FullStatus               = status.FullStatus
+	RouterState              = status.RouterState
+	LocalPeerState           = status.LocalPeerState
+	SignalState              = status.SignalState
+	ManagementState          = status.ManagementState
+	RosenpassState           = status.RosenpassState
+	NSGroupState             = status.NSGroupState
+	ResolvedDomainInfo       = status.ResolvedDomainInfo
+	StatusChangeSubscription = status.StatusChangeSubscription
+	EventQueue               = status.EventQueue
+	EventSubscription        = status.EventSubscription
+	WGIfaceStatus            = status.WGIfaceStatus
+	Listener                 = status.Listener
+	EventListener            = status.EventListener
+)
+
+const (
+	StatusIdle       = status.StatusIdle
+	StatusConnecting = status.StatusConnecting
+	StatusConnected  = status.StatusConnected
+)
+
+var (
+	NewRecorder = status.NewRecorder
+)

client/internal/peer/wg_watcher/wg_watcher.go

@@ -1,4 +1,4 @@
-package peer
+package wg_watcher
 
 import (
 	"context"
@@ -9,6 +9,7 @@ import (
 	log "github.com/sirupsen/logrus"
 
 	"github.com/netbirdio/netbird/client/iface/configurer"
+	"github.com/netbirdio/netbird/client/internal/peer/state_dump"
 )
 
 const (
@@ -28,7 +29,7 @@ type WGWatcher struct {
 	log           *log.Entry
 	wgIfaceStater WGInterfaceStater
 	peerKey       string
-	stateDump     *stateDump
+	stateDump     *state_dump.StateDump
 
 	enabled   bool
 	muEnabled sync.Mutex
@@ -38,7 +39,7 @@ type WGWatcher struct {
 	resetCh chan struct{}
 }
 
-func NewWGWatcher(log *log.Entry, wgIfaceStater WGInterfaceStater, peerKey string, stateDump *stateDump) *WGWatcher {
+func NewWGWatcher(log *log.Entry, wgIfaceStater WGInterfaceStater, peerKey string, stateDump *state_dump.StateDump) *WGWatcher {
 	return &WGWatcher{
 		log:           log,
 		wgIfaceStater: wgIfaceStater,

client/internal/peer/wg_watcher/wg_watcher_test.go

@@ -1,4 +1,4 @@
-package peer
+package wg_watcher
 
 import (
 	"context"
@@ -10,6 +10,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/netbirdio/netbird/client/iface/configurer"
+	"github.com/netbirdio/netbird/client/internal/peer/state_dump"
+	"github.com/netbirdio/netbird/client/internal/peer/status"
 )
 
 type MocWgIface struct {
@@ -57,7 +59,7 @@ func TestWGWatcher_CheckSuccessCallback(t *testing.T) {
 	// platforms with coarse clock resolution (Windows), where two time.Now() calls
 	// microseconds apart can return the same instant and read as a timed-out handshake.
 	stats := &mockHandshakeStats{handshake: time.Now().Add(-time.Hour)}
-	watcher := NewWGWatcher(mlog, stats, "", newStateDump("peer", mlog, &Status{}))
+	watcher := NewWGWatcher(mlog, stats, "", state_dump.NewStateDump("peer", mlog, &status.Recorder{}))
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -66,14 +68,18 @@ func TestWGWatcher_CheckSuccessCallback(t *testing.T) {
 
 	firstHandshake := make(chan struct{}, 1)
 	checkSuccess := make(chan struct{}, 1)
-	go watcher.EnableWgWatcher(ctx, time.Now(), func() {}, func(when time.Time) {
-		firstHandshake <- struct{}{}
-	}, func() {
-		select {
-		case checkSuccess <- struct{}{}:
-		default:
-		}
-	})
+	watcherDone := make(chan struct{})
+	go func() {
+		defer close(watcherDone)
+		watcher.EnableWgWatcher(ctx, time.Now(), func() {}, func(when time.Time) {
+			firstHandshake <- struct{}{}
+		}, func() {
+			select {
+			case checkSuccess <- struct{}{}:
+			default:
+			}
+		})
+	}()
 
 	stats.advance()
 
@@ -88,6 +94,11 @@ func TestWGWatcher_CheckSuccessCallback(t *testing.T) {
 		t.Errorf("first-handshake callback must not fire for a non-zero baseline")
 	default:
 	}
+
+	// Wait for the watcher goroutine to exit so it cannot race with other
+	// tests mutating the package-level check timing variables.
+	cancel()
+	<-watcherDone
 }
 
 func TestWGWatcher_EnableWgWatcher(t *testing.T) {
@@ -96,7 +107,7 @@ func TestWGWatcher_EnableWgWatcher(t *testing.T) {
 
 	mlog := log.WithField("peer", "tet")
 	mocWgIface := &MocWgIface{}
-	watcher := NewWGWatcher(mlog, mocWgIface, "", newStateDump("peer", mlog, &Status{}))
+	watcher := NewWGWatcher(mlog, mocWgIface, "", state_dump.NewStateDump("peer", mlog, &status.Recorder{}))
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -129,7 +140,7 @@ func TestWGWatcher_ReEnable(t *testing.T) {
 
 	mlog := log.WithField("peer", "tet")
 	mocWgIface := &MocWgIface{}
-	watcher := NewWGWatcher(mlog, mocWgIface, "", newStateDump("peer", mlog, &Status{}))
+	watcher := NewWGWatcher(mlog, mocWgIface, "", state_dump.NewStateDump("peer", mlog, &status.Recorder{}))
 
 	ctx, cancel := context.WithCancel(context.Background())
 	ok := watcher.PrepareInitialHandshake()

client/internal/peer/worker/priority.go

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

client/internal/peer/worker/worker_ice.go

@@ -1,4 +1,4 @@
-package peer
+package worker
 
 import (
 	"context"
@@ -13,8 +13,9 @@ import (
 
 	"github.com/netbirdio/netbird/client/iface"
 	"github.com/netbirdio/netbird/client/iface/udpmux"
-	"github.com/netbirdio/netbird/client/internal/peer/conntype"
 	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
+	"github.com/netbirdio/netbird/client/internal/peer/signaling"
+	"github.com/netbirdio/netbird/client/internal/peer/status"
 	"github.com/netbirdio/netbird/client/internal/portforward"
 	"github.com/netbirdio/netbird/client/internal/stdnet"
 	"github.com/netbirdio/netbird/route"
@@ -32,57 +33,68 @@ type ICEConnInfo struct {
 	RelayedOnLocal             bool
 }
 
-type WorkerICE struct {
-	ctx               context.Context
-	log               *log.Entry
-	config            ConnConfig
-	conn              *Conn
-	signaler          *Signaler
-	iFaceDiscover     stdnet.ExternalIFaceDiscover
-	statusRecorder    *Status
-	hasRelayOnLocally bool
+type ICEDependencies struct {
+	Signaler           *signaling.Signaler
+	IFaceDiscover      stdnet.ExternalIFaceDiscover
+	StatusRecorder     *status.Recorder
+	PortForwardManager *portforward.Manager
+}
+
+type ICE struct {
+	log                *log.Entry
+	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
+	portForwardManager *portforward.Manager
+	hasRelayOnLocally  bool
 
 	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
+	connectedAgent *icemaker.ThreadSafeAgent
 	// remoteSessionID represents the peer's session identifier from the latest remote offer.
-	remoteSessionID ICESessionID
+	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            ICESessionID
+	sessionID            icemaker.SessionID
 	remoteSessionChanged bool
 	muxAgent             sync.Mutex
 
 	localUfrag string
 	localPwd   string
 
-	// we record the last known state of the ICE agent to avoid duplicate on disconnected events
-	lastKnownState ice.ConnectionState
-
 	// portForwardAttempted tracks if we've already tried port forwarding this session
 	portForwardAttempted bool
 }
 
-func NewWorkerICE(ctx context.Context, log *log.Entry, config ConnConfig, conn *Conn, signaler *Signaler, ifaceDiscover stdnet.ExternalIFaceDiscover, statusRecorder *Status, hasRelayOnLocally bool) (*WorkerICE, error) {
-	sessionID, err := NewICESessionID()
+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,
-		log:               log,
-		config:            config,
-		conn:              conn,
-		signaler:          signaler,
-		iFaceDiscover:     ifaceDiscover,
-		statusRecorder:    statusRecorder,
-		hasRelayOnLocally: hasRelayOnLocally,
-		lastKnownState:    ice.ConnectionStateDisconnected,
-		sessionID:         sessionID,
+	w := &ICE{
+		log:                log,
+		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,
 	}
 
 	localUfrag, localPwd, err := icemaker.GenerateICECredentials()
@@ -94,7 +106,7 @@ func NewWorkerICE(ctx context.Context, log *log.Entry, config ConnConfig, conn *
 	return w, nil
 }
 
-func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *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()
@@ -118,7 +130,7 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 			}
 		}
 
-		sessionID, err := NewICESessionID()
+		sessionID, err := icemaker.NewSessionID()
 		if err != nil {
 			w.log.Errorf("failed to create new session ID: %s", err)
 		}
@@ -136,8 +148,8 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *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
@@ -151,14 +163,14 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 		w.remoteSessionID = ""
 	}
 
-	go w.connect(dialerCtx, agent, remoteOfferAnswer)
+	go w.connect(dialerCtx, dialerCancel, agent, remoteOfferAnswer)
 }
 
 // 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
@@ -185,18 +197,24 @@ func (w *WorkerICE) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HA
 	}
 }
 
-func (w *WorkerICE) GetLocalUserCredentials() (frag string, pwd string) {
-	return w.localUfrag, w.localPwd
+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,
+	}
 }
 
-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()
 
@@ -212,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)
 	}
@@ -237,7 +255,7 @@ func (w *WorkerICE) reCreateAgent(dialerCancel context.CancelFunc, candidates []
 	return agent, nil
 }
 
-func (w *WorkerICE) SessionID() ICESessionID {
+func (w *ICE) getSessionID() icemaker.SessionID {
 	w.muxAgent.Lock()
 	defer w.muxAgent.Unlock()
 
@@ -247,11 +265,11 @@ 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 *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, w.agentDialerCancel)
+		w.closeAgent(agent, dialerCancel)
 		return
 	}
 
@@ -259,19 +277,19 @@ func (w *WorkerICE) connect(ctx context.Context, agent *icemaker.ThreadSafeAgent
 	remoteConn, err := w.turnAgentDial(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")
 
 	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
 	}
 
@@ -299,17 +317,22 @@ func (w *WorkerICE) connect(ctx context.Context, agent *icemaker.ThreadSafeAgent
 	}
 	w.log.Debugf("on ICE conn is ready to use")
 
-	w.log.Infof("connection succeeded with offer session: %s", remoteOfferAnswer.SessionIDString())
 	w.muxAgent.Lock()
+	if w.agent != agent {
+		w.muxAgent.Unlock()
+		w.log.Debugf("agent has been replaced during connect, dropping obsolete connection")
+		return
+	}
 	w.agentConnecting = false
 	w.lastSuccess = time.Now()
+	w.connectedAgent = agent
 	w.muxAgent.Unlock()
 
-	// todo: the potential problem is a race between the onConnectionStateChange
-	w.conn.onICEConnectionIsReady(selectedPriority(pair), ci)
+	w.log.Infof("connection succeeded with offer session: %s", remoteOfferAnswer.SessionIDString())
+	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)
@@ -323,7 +346,7 @@ func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.C
 
 	if w.agent == agent {
 		// consider to remove from here and move to the OnNewOffer
-		sessionID, err := NewICESessionID()
+		sessionID, err := icemaker.NewSessionID()
 		if err != nil {
 			w.log.Errorf("failed to create new session ID: %s", err)
 		}
@@ -335,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)))
@@ -344,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
@@ -357,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
@@ -366,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)
 		}
 	}()
 
@@ -378,8 +401,8 @@ 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) {
-	pfManager := w.conn.portForwardManager
+func (w *ICE) injectPortForwardedCandidate(srflxCandidate ice.Candidate) {
+	pfManager := w.portForwardManager
 	if pfManager == nil {
 		return
 	}
@@ -407,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)
 		}
 	}()
@@ -415,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()
@@ -460,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 {
@@ -471,14 +494,14 @@ 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) {
-	sessionID := w.SessionID()
+func (w *ICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) {
+	sessionID := w.getSessionID()
 	stats := agent.GetCandidatePairsStats()
 	localCandidates, _ := agent.GetLocalCandidates()
 	remoteCandidates, _ := agent.GetRemoteCandidates()
@@ -508,32 +531,44 @@ 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) {
 		w.log.Debugf("ICE ConnectionState has changed to %s", state.String())
 		switch state {
 		case ice.ConnectionStateConnected:
-			w.lastKnownState = ice.ConnectionStateConnected
+			connected = true
 			w.logSuccessfulPaths(agent)
-			return
 		case ice.ConnectionStateFailed, ice.ConnectionStateDisconnected, ice.ConnectionStateClosed:
 			// ice.ConnectionStateClosed happens when we recreate the agent. For the P2P to TURN switch important to
 			// notify the conn.onICEStateDisconnected changes to update the current used priority
 
 			sessionChanged := w.closeAgent(agent, dialerCancel)
 
-			if w.lastKnownState == ice.ConnectionStateConnected {
-				w.lastKnownState = ice.ConnectionStateDisconnected
-				w.conn.onICEStateDisconnected(sessionChanged)
+			if !connected {
+				return
 			}
-		default:
-			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)
 		}
 	}
 }
 
-func (w *WorkerICE) turnAgentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *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)
@@ -595,10 +630,10 @@ func isRelayed(pair *ice.CandidatePair) bool {
 	return false
 }
 
-func selectedPriority(pair *ice.CandidatePair) conntype.ConnPriority {
+func selectedPriority(pair *ice.CandidatePair) ConnPriority {
 	if isRelayed(pair) {
-		return conntype.ICETurn
+		return ICETurn
 	} else {
-		return conntype.ICEP2P
+		return ICEP2P
 	}
 }

client/internal/peer/worker/worker_relay.go

@@ -1,4 +1,4 @@
-package peer
+package worker
 
 import (
 	"context"
@@ -10,22 +10,23 @@ import (
 
 	log "github.com/sirupsen/logrus"
 
+	"github.com/netbirdio/netbird/client/internal/peer/signaling"
 	relayClient "github.com/netbirdio/netbird/shared/relay/client"
 )
 
 type RelayConnInfo struct {
-	relayedConn     net.Conn
-	rosenpassPubKey []byte
-	rosenpassAddr   string
+	RelayedConn     net.Conn
+	RosenpassPubKey []byte
+	RosenpassAddr   string
 }
 
 type WorkerRelay struct {
-	peerCtx      context.Context
-	log          *log.Entry
-	isController bool
-	config       ConnConfig
-	conn         *Conn
-	relayManager *relayClient.Manager
+	log            *log.Entry
+	key            string
+	isController   bool
+	onConnReady    func(RelayConnInfo)
+	onDisconnected func()
+	relayManager   *relayClient.Manager
 
 	relayedConn net.Conn
 	relayLock   sync.Mutex
@@ -33,19 +34,19 @@ type WorkerRelay struct {
 	relaySupportedOnRemotePeer atomic.Bool
 }
 
-func NewWorkerRelay(ctx context.Context, log *log.Entry, ctrl bool, config ConnConfig, conn *Conn, 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 *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)
@@ -66,7 +67,7 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *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")
@@ -88,10 +89,10 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 	}
 
 	w.log.Debugf("peer conn opened via Relay: %s", srv)
-	go w.conn.onRelayConnectionIsReady(RelayConnInfo{
-		relayedConn:     relayedConn,
-		rosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
-		rosenpassAddr:   remoteOfferAnswer.RosenpassAddr,
+	w.onConnReady(RelayConnInfo{
+		RelayedConn:     relayedConn,
+		RosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
+		RosenpassAddr:   remoteOfferAnswer.RosenpassAddr,
 	})
 }
 
@@ -119,7 +120,7 @@ func (w *WorkerRelay) CloseConn() {
 	}
 }
 
-func (w *WorkerRelay) isRelaySupported(answer *OfferAnswer) bool {
+func (w *WorkerRelay) isRelaySupported(answer *signaling.OfferAnswer) bool {
 	if !w.relayManager.HasRelayAddress() {
 		return false
 	}
@@ -134,5 +135,5 @@ func (w *WorkerRelay) preferredRelayServer(myRelayAddress, remoteRelayAddress st
 }
 
 func (w *WorkerRelay) onRelayClientDisconnected() {
-	go w.conn.onRelayDisconnected()
+	w.onDisconnected()
 }

client/internal/peer/worker_status.go

@@ -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