[client] Refactor peer Conn to a single-owner event loop

Replace the mutex-guarded callback model of peer.Conn with a per-peer event loop that exclusively owns all mutable connection state. External callers and transport workers post typed events into a non-blocking, coalescing mailbox instead of contending on conn.mu: - offers/answers coalesce to the newest message, a new offer flushes queued candidates of the superseded session - candidates are applied in arrival order from a bounded FIFO - transport state changes are never dropped - the blocking relay dial runs on a helper goroutine with a single dial in flight; signaling I/O (offer/answer sends) runs off the loop conn.mu now only guards the open/close lifecycle. Close posts a close event and waits for the loop teardown; the loop also tears down on engine context cancellation and releases resources of unprocessed events. Delete the Handshaker listener machinery (Listen loop, unbuffered drop-on-busy channels, AsyncOfferListener with its double-processing of the first offer), the never-wired dispatcher package and the unused ICEMonitor.ReconnectCh. Fix a goroutine leak in the WG watcher test that raced with tests mutating the package-level check timing vars.
2026-07-16 11:39:57 +00:00 · 2026-07-11 13:30:06 +02:00
parent 30d15ecc3d
commit 5f98524e02
12 changed files with 778 additions and 537 deletions
--- a/client/internal/peer/conn.go
+++ b/client/internal/peer/conn.go
@@ -8,6 +8,7 @@ import (
 	"runtime"
 	"slices"
 	"sync"
+	"sync/atomic"
 	"time"

 	"github.com/pion/ice/v4"
@@ -18,7 +19,6 @@ import (
 	"github.com/netbirdio/netbird/client/iface/wgproxy"
 	"github.com/netbirdio/netbird/client/internal/metrics"
 	"github.com/netbirdio/netbird/client/internal/peer/conntype"
-	"github.com/netbirdio/netbird/client/internal/peer/dispatcher"
 	"github.com/netbirdio/netbird/client/internal/peer/guard"
 	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
 	"github.com/netbirdio/netbird/client/internal/peer/id"
@@ -52,7 +52,6 @@ type ServiceDependencies struct {
 	IFaceDiscover      stdnet.ExternalIFaceDiscover
 	RelayManager       *relayClient.Manager
 	SrWatcher          *guard.SRWatcher
-	PeerConnDispatcher *dispatcher.ConnectionDispatcher
 	PortForwardManager *portforward.Manager
 	MetricsRecorder    MetricsRecorder
 }
@@ -95,8 +94,14 @@ type ConnConfig struct {
 	ICEConfig icemaker.Config
 }

+// 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
+// a non-blocking mailbox.
 type Conn struct {
-	Log                *log.Entry
+	Log *log.Entry
+	// mu guards the open/close lifecycle (opened, loopDone). Everything else
+	// is either immutable after construction or owned by the event loop.
 	mu                 sync.Mutex
 	ctx                context.Context
 	ctxCancel          context.CancelFunc
@@ -117,14 +122,25 @@ type Conn struct {
 	currentConnPriority conntype.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
+	// generation; a nil pointer or a closed mailbox rejects the post.
+	mailbox  atomic.Pointer[mailbox]
+	loopDone chan struct{}
+
 	workerICE   *WorkerICE
 	workerRelay *WorkerRelay

+	// relayDialInFlight and pendingRelayOffer serialize the blocking relay
+	// dials spawned by the event loop, keeping only the newest offer while a
+	// dial is running.
+	relayDialInFlight bool
+	pendingRelayOffer *OfferAnswer
+
 	wgWatcher       *WGWatcher
 	wgWatcherWg     sync.WaitGroup
 	wgWatcherCancel context.CancelFunc
 	// wgTimeouts counts consecutive WireGuard handshake timeouts without a
-	// successful handshake in between. Guarded by mu.
+	// successful handshake in between. Owned by the event loop.
 	wgTimeouts int

 	// used to store the remote Rosenpass key for Relayed connection in case of connection update from ice
@@ -151,35 +167,6 @@ type Conn struct {
 	pendingFirstPacket []byte
 }

-// injectPendingFirstPacket replays the captured handshake through the proxy if present, else
-// directly through the ICE conn. The packet is cleared only after a successful write, so a failed
-// or transport-less attempt leaves it available for a later reinjection. Caller must hold conn.mu.
-func (conn *Conn) injectPendingFirstPacket(proxy wgproxy.Proxy, directConn net.Conn) {
-	pkt := conn.pendingFirstPacket
-	if len(pkt) == 0 {
-		return
-	}
-
-	switch {
-	case proxy != nil:
-		if err := proxy.InjectPacket(pkt); err != nil {
-			conn.Log.Debugf("failed to reinject captured first packet via proxy: %v", err)
-			return
-		}
-	case directConn != nil:
-		if _, err := directConn.Write(pkt); err != nil {
-			conn.Log.Debugf("failed to reinject captured first packet via direct conn: %v", err)
-			return
-		}
-	default:
-		conn.Log.Debugf("no transport available to reinject captured first packet")
-		return
-	}
-
-	conn.pendingFirstPacket = nil
-	conn.Log.Debugf("reinjected captured first packet (%d bytes)", len(pkt))
-}
-
 // NewConn creates a new not opened Conn to the remote peer.
 // To establish a connection run Conn.Open
 func NewConn(config ConnConfig, services ServiceDependencies) (*Conn, error) {
@@ -238,8 +225,7 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error {

 	conn.workerRelay = NewWorkerRelay(conn.ctx, conn.Log, isController(conn.config), conn.config, conn, conn.relayManager)

-	forceRelay := IsForceRelayed()
-	if !forceRelay {
+	if !IsForceRelayed() {
 		relayIsSupportedLocally := conn.workerRelay.RelayIsSupportedLocally()
 		workerICE, err := NewWorkerICE(conn.ctx, conn.Log, conn.config, conn, conn.signaler, conn.iFaceDiscover, conn.statusRecorder, relayIsSupportedLocally)
 		if err != nil {
@@ -248,21 +234,15 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error {
 		conn.workerICE = workerICE
 	}

-	conn.handshaker = NewHandshaker(conn.Log, conn.config, conn.signaler, conn.workerICE, conn.workerRelay, conn.metricsStages)
-
-	conn.handshaker.AddRelayListener(conn.workerRelay.OnNewOffer)
-	if !forceRelay {
-		conn.handshaker.AddICEListener(conn.workerICE.OnNewOffer)
-	}
+	conn.handshaker = NewHandshaker(conn.Log, conn.config, conn.signaler, conn.workerICE, conn.workerRelay)

 	conn.guard = guard.NewGuard(conn.Log, conn.isConnectedOnAllWay, conn.config.Timeout, conn.srWatcher)

-	conn.wg.Add(1)
-	go func() {
-		defer conn.wg.Done()
-		conn.handshaker.Listen(conn.ctx)
-	}()
-	go conn.dumpState.Start(conn.ctx)
+	conn.relayDialInFlight = false
+	conn.pendingRelayOffer = nil
+	if len(firstPacket) > 0 {
+		conn.pendingFirstPacket = slices.Clone(firstPacket)
+	}

 	peerState := State{
 		PubKey:           conn.config.Key,
@@ -274,22 +254,27 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error {
 		conn.Log.Warnf("error while updating the state err: %v", err)
 	}

+	mb := newMailbox()
+	conn.loopDone = make(chan struct{})
+	conn.mailbox.Store(mb)
+
+	go conn.run(mb)
+	go conn.dumpState.Start(conn.ctx)
+
 	conn.wg.Add(1)
 	go func() {
 		defer conn.wg.Done()
 		conn.guard.Start(conn.ctx, conn.onGuardEvent)
 	}()
-	if len(firstPacket) > 0 {
-		conn.pendingFirstPacket = slices.Clone(firstPacket)
-	}
+
 	conn.opened = true
 	return nil
 }

-// Close closes this peer Conn issuing a close event to the Conn closeCh
+// Close closes this peer Conn. It posts a close event to the event loop and
+// blocks until the loop finished the teardown.
 func (conn *Conn) Close(signalToRemote bool) {
 	conn.mu.Lock()
-	defer conn.wgWatcherWg.Wait()
 	defer conn.mu.Unlock()

 	if !conn.opened {
@@ -297,66 +282,43 @@ func (conn *Conn) Close(signalToRemote bool) {
 		return
 	}

-	if signalToRemote {
-		if err := conn.signaler.SignalIdle(conn.config.Key); err != nil {
-			conn.Log.Errorf("failed to signal idle state to peer: %v", err)
-		}
+	done := make(chan struct{})
+	if conn.post(evClose{signalToRemote: signalToRemote, done: done}) {
+		<-done
+	} else {
+		<-conn.loopDone
 	}
-
-	conn.Log.Infof("close peer connection")
-	conn.ctxCancel()
-
-	if conn.wgWatcherCancel != nil {
-		conn.wgWatcherCancel()
-	}
-	conn.workerRelay.CloseConn()
-	if conn.workerICE != nil {
-		conn.workerICE.Close()
-	}
-
-	if conn.wgProxyRelay != nil {
-		err := conn.wgProxyRelay.CloseConn()
-		if err != nil {
-			conn.Log.Errorf("failed to close wg proxy for relay: %v", err)
-		}
-		conn.wgProxyRelay = nil
-	}
-
-	if conn.wgProxyICE != nil {
-		err := conn.wgProxyICE.CloseConn()
-		if err != nil {
-			conn.Log.Errorf("failed to close wg proxy for ice: %v", err)
-		}
-		conn.wgProxyICE = nil
-	}
-
-	if err := conn.endpointUpdater.RemoveWgPeer(); err != nil {
-		conn.Log.Errorf("failed to remove wg endpoint: %v", err)
-	}
-
-	if conn.evalStatus() == StatusConnected && conn.onDisconnected != nil {
-		conn.onDisconnected(conn.config.WgConfig.RemoteKey)
-	}
-
-	conn.setStatusToDisconnected()
 	conn.opened = false
-	conn.wg.Wait()
-	conn.Log.Infof("peer connection closed")
 }

-// 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
+// OnRemoteOffer handles an offer from the remote peer. It never blocks; the
+// offer is coalesced in the mailbox and processed by the event loop, older
+// unprocessed offers are replaced by the newest one.
+func (conn *Conn) OnRemoteOffer(offer OfferAnswer) {
+	conn.dumpState.RemoteOffer()
+	conn.Log.Infof("OnRemoteOffer, on status ICE: %s, status Relay: %s", conn.statusICE, conn.statusRelay)
+	if !conn.post(evRemoteOffer{offer: offer}) {
+		conn.Log.Debugf("connection is not open, discarding remote offer")
+	}
+}
+
+// OnRemoteAnswer handles an answer from the remote peer. It never blocks; the
+// answer is coalesced in the mailbox and processed by the event loop.
 func (conn *Conn) OnRemoteAnswer(answer OfferAnswer) {
 	conn.dumpState.RemoteAnswer()
-	conn.Log.Infof("OnRemoteAnswer, priority: %s, status ICE: %s, status relay: %s", conn.currentConnPriority, conn.statusICE, conn.statusRelay)
-	conn.handshaker.OnRemoteAnswer(answer)
+	conn.Log.Infof("OnRemoteAnswer, on status ICE: %s, status Relay: %s", conn.statusICE, conn.statusRelay)
+	if !conn.post(evRemoteAnswer{answer: answer}) {
+		conn.Log.Debugf("connection is not open, discarding remote answer")
+	}
 }

-// OnRemoteCandidate Handles ICE connection Candidate provided by the remote peer.
+// OnRemoteCandidate handles an ICE connection candidate provided by the
+// remote peer. Candidates are queued in arrival order and applied by the
+// event loop.
 func (conn *Conn) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HAMap) {
 	conn.dumpState.RemoteCandidate()
-	if conn.workerICE != nil {
-		conn.workerICE.OnRemoteCandidate(candidate, haRoutes)
+	if !conn.post(evRemoteCandidate{candidate: candidate, haRoutes: haRoutes}) {
+		conn.Log.Debugf("connection is not open, discarding remote candidate")
 	}
 }

@@ -377,12 +339,6 @@ func (conn *Conn) SetRosenpassInitializedPresharedKeyValidator(handler func(peer
 	conn.rosenpassInitializedPresharedKeyValidator = handler
 }

-func (conn *Conn) OnRemoteOffer(offer OfferAnswer) {
-	conn.dumpState.RemoteOffer()
-	conn.Log.Infof("OnRemoteOffer, on status ICE: %s, status Relay: %s", conn.statusICE, conn.statusRelay)
-	conn.handshaker.OnRemoteOffer(offer)
-}
-
 // WgConfig returns the WireGuard config
 func (conn *Conn) WgConfig() WgConfig {
 	return conn.config.WgConfig
@@ -390,9 +346,6 @@ func (conn *Conn) WgConfig() WgConfig {

 // IsConnected returns true if the peer is connected
 func (conn *Conn) IsConnected() bool {
-	conn.mu.Lock()
-	defer conn.mu.Unlock()
-
 	return conn.evalStatus() == StatusConnected
 }

@@ -404,11 +357,236 @@ func (conn *Conn) ConnID() id.ConnID {
 	return id.ConnID(conn)
 }

-// configureConnection starts proxying traffic from/to local Wireguard and sets connection status to StatusConnected
-func (conn *Conn) onICEConnectionIsReady(priority conntype.ConnPriority, iceConnInfo ICEConnInfo) {
-	conn.mu.Lock()
-	defer conn.mu.Unlock()
+// AllowedIP returns the allowed IP of the remote peer
+func (conn *Conn) AllowedIP() netip.Addr {
+	return conn.config.WgConfig.AllowedIps[0].Addr()
+}

+func (conn *Conn) AgentVersionString() string {
+	return conn.config.AgentVersion
+}
+
+// post delivers an event to the event loop of the current open generation.
+// It reports false if the connection is not open.
+func (conn *Conn) post(ev event) bool {
+	mb := conn.mailbox.Load()
+	if mb == nil {
+		return false
+	}
+	return mb.post(ev)
+}
+
+// run is the Conn event loop. From the moment open returns until the teardown
+// it exclusively owns all mutable Conn state. It exits on an evClose event or
+// when the engine context is cancelled.
+func (conn *Conn) run(mb *mailbox) {
+	defer close(conn.loopDone)
+
+	for {
+		select {
+		case <-mb.wake:
+			evs := mb.drain()
+			for i, ev := range evs {
+				if c, ok := ev.(evClose); ok {
+					conn.teardown(mb, evs[i+1:], c.signalToRemote, c.done)
+					return
+				}
+				conn.handleEvent(ev)
+			}
+		case <-conn.ctx.Done():
+			conn.teardown(mb, nil, false, nil)
+			return
+		}
+	}
+}
+
+func (conn *Conn) handleEvent(ev event) {
+	switch e := ev.(type) {
+	case evRemoteOffer:
+		conn.handleRemoteOffer(&e.offer)
+	case evRemoteAnswer:
+		conn.handleRemoteAnswer(&e.answer)
+	case evRemoteCandidate:
+		conn.handleRemoteCandidate(e)
+	case evICEReady:
+		conn.handleICEReady(e.priority, e.info)
+	case evICEDown:
+		conn.handleICEDisconnected(e.sessionChanged)
+	case evRelayReady:
+		conn.handleRelayReady(e.info)
+	case evRelayDown:
+		conn.handleRelayDisconnected()
+	case evRelayDialDone:
+		conn.handleRelayDialDone()
+	case evWGTimeout:
+		conn.handleWGTimeout()
+	case evWGHandshake:
+		conn.handleWGHandshakeSuccess(e.when)
+	case evWGCheckOK:
+		conn.handleWGCheckSuccess()
+	case evGuardTick:
+		conn.handleGuardTick()
+	default:
+		conn.Log.Errorf("unhandled conn event type %T", ev)
+	}
+}
+
+// teardown closes the transports and releases every resource of the current
+// open generation. It runs exclusively on the event loop, either for an
+// evClose event or after engine context cancellation. Leftover events drained
+// together with the close are cleaned up alongside the ones still sitting in
+// the mailbox.
+func (conn *Conn) teardown(mb *mailbox, leftover []event, signalToRemote bool, done chan struct{}) {
+	if signalToRemote {
+		if err := conn.signaler.SignalIdle(conn.config.Key); err != nil {
+			conn.Log.Errorf("failed to signal idle state to peer: %v", err)
+		}
+	}
+
+	conn.Log.Infof("close peer connection")
+	conn.ctxCancel()
+
+	if conn.wgWatcherCancel != nil {
+		conn.wgWatcherCancel()
+		conn.wgWatcherCancel = nil
+	}
+	conn.workerRelay.CloseConn()
+	if conn.workerICE != nil {
+		conn.workerICE.Close()
+	}
+
+	if conn.wgProxyRelay != nil {
+		if err := conn.wgProxyRelay.CloseConn(); err != nil {
+			conn.Log.Errorf("failed to close wg proxy for relay: %v", err)
+		}
+		conn.wgProxyRelay = nil
+	}
+
+	if conn.wgProxyICE != nil {
+		if err := conn.wgProxyICE.CloseConn(); err != nil {
+			conn.Log.Errorf("failed to close wg proxy for ice: %v", err)
+		}
+		conn.wgProxyICE = nil
+	}
+
+	if err := conn.endpointUpdater.RemoveWgPeer(); err != nil {
+		conn.Log.Errorf("failed to remove wg endpoint: %v", err)
+	}
+
+	if conn.evalStatus() == StatusConnected && conn.onDisconnected != nil {
+		conn.onDisconnected(conn.config.WgConfig.RemoteKey)
+	}
+
+	conn.setStatusToDisconnected()
+	conn.wgWatcherWg.Wait()
+	conn.wg.Wait()
+
+	conn.releaseEvents(leftover)
+	conn.releaseEvents(mb.closeAndDrain())
+	if done != nil {
+		close(done)
+	}
+	conn.Log.Infof("peer connection closed")
+}
+
+// releaseEvents cleans up events that will never be processed because the
+// event loop is shutting down.
+func (conn *Conn) releaseEvents(evs []event) {
+	for _, ev := range evs {
+		switch e := ev.(type) {
+		case evRelayReady:
+			if err := e.info.relayedConn.Close(); err != nil {
+				conn.Log.Warnf("failed to close unnecessary relayed connection: %v", err)
+			}
+		case evClose:
+			if e.done != nil {
+				close(e.done)
+			}
+		}
+	}
+}
+
+// handleRemoteOffer applies a remote offer on the event loop: refreshes the
+// remote ICE support state, dispatches the offer to the relay and ICE workers
+// and answers it.
+func (conn *Conn) handleRemoteOffer(offer *OfferAnswer) {
+	conn.Log.Infof("received offer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", offer.Version, offer.WgListenPort, offer.SessionIDString(), offer.hasICECredentials())
+
+	conn.metricsStages.RecordSignalingReceived()
+	conn.handshaker.updateRemoteICEState(offer)
+	conn.dispatchOfferToRelay(offer)
+	conn.dispatchOfferToICE(offer)
+
+	go func() {
+		if err := conn.handshaker.SendAnswer(); err != nil {
+			conn.Log.Errorf("failed to send remote offer confirmation: %s", err)
+		}
+	}()
+}
+
+// handleRemoteAnswer applies a remote answer on the event loop the same way
+// as an offer, without answering it.
+func (conn *Conn) handleRemoteAnswer(answer *OfferAnswer) {
+	conn.Log.Infof("received answer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", answer.Version, answer.WgListenPort, answer.SessionIDString(), answer.hasICECredentials())
+
+	conn.metricsStages.RecordSignalingReceived()
+	conn.handshaker.updateRemoteICEState(answer)
+	conn.dispatchOfferToRelay(answer)
+	conn.dispatchOfferToICE(answer)
+}
+
+func (conn *Conn) handleRemoteCandidate(e evRemoteCandidate) {
+	if conn.workerICE == nil {
+		return
+	}
+	conn.workerICE.OnRemoteCandidate(e.candidate, e.haRoutes)
+}
+
+func (conn *Conn) dispatchOfferToICE(offer *OfferAnswer) {
+	if conn.workerICE == nil || !conn.handshaker.RemoteICESupported() {
+		return
+	}
+	conn.workerICE.OnNewOffer(offer)
+}
+
+// dispatchOfferToRelay runs the blocking relay dial on a helper goroutine. A
+// single dial is kept in flight; newer offers replace the pending one and the
+// newest is dispatched once the running dial reports completion.
+func (conn *Conn) dispatchOfferToRelay(offer *OfferAnswer) {
+	if conn.relayDialInFlight {
+		conn.pendingRelayOffer = offer
+		return
+	}
+	conn.relayDialInFlight = true
+
+	go func() {
+		conn.workerRelay.OnNewOffer(offer)
+		conn.post(evRelayDialDone{})
+	}()
+}
+
+func (conn *Conn) handleRelayDialDone() {
+	conn.relayDialInFlight = false
+	if offer := conn.pendingRelayOffer; offer != nil {
+		conn.pendingRelayOffer = nil
+		conn.dispatchOfferToRelay(offer)
+	}
+}
+
+// handleGuardTick sends a new offer to restore connectivity; the signaling
+// I/O runs off the loop.
+func (conn *Conn) handleGuardTick() {
+	conn.dumpState.SendOffer()
+	go func() {
+		if err := conn.handshaker.SendOffer(); err != nil {
+			conn.Log.Errorf("failed to send offer: %v", err)
+		}
+	}()
+}
+
+// handleICEReady starts proxying traffic from/to local WireGuard and sets the
+// connection status to StatusConnected.
+func (conn *Conn) handleICEReady(priority conntype.ConnPriority, iceConnInfo ICEConnInfo) {
 	if conn.ctx.Err() != nil {
 		return
 	}
@@ -486,10 +664,9 @@ func (conn *Conn) onICEConnectionIsReady(priority conntype.ConnPriority, iceConn
 	conn.doOnConnected(iceConnInfo.RosenpassPubKey, iceConnInfo.RosenpassAddr, updateTime)
 }

-func (conn *Conn) onICEStateDisconnected(sessionChanged bool) {
-	conn.mu.Lock()
-	defer conn.mu.Unlock()
-
+// handleICEDisconnected switches back to the relay connection if available,
+// otherwise cleans up the WireGuard endpoint.
+func (conn *Conn) handleICEDisconnected(sessionChanged bool) {
 	if conn.ctx.Err() != nil {
 		return
 	}
@@ -550,10 +727,9 @@ func (conn *Conn) onICEStateDisconnected(sessionChanged bool) {
 	}
 }

-func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) {
-	conn.mu.Lock()
-	defer conn.mu.Unlock()
-
+// 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) {
 	if conn.ctx.Err() != nil {
 		if err := rci.relayedConn.Close(); err != nil {
 			conn.Log.Warnf("failed to close unnecessary relayed connection: %v", err)
@@ -614,14 +790,9 @@ func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) {
 	conn.doOnConnected(rci.rosenpassPubKey, rci.rosenpassAddr, updateTime)
 }

-func (conn *Conn) onRelayDisconnected() {
-	conn.mu.Lock()
-	defer conn.mu.Unlock()
-	conn.handleRelayDisconnectedLocked()
-}
-
-// handleRelayDisconnectedLocked handles relay disconnection. Caller must hold conn.mu.
-func (conn *Conn) handleRelayDisconnectedLocked() {
+// handleRelayDisconnected cleans up the relayed transport and the WireGuard
+// endpoint if the relay was the active connection.
+func (conn *Conn) handleRelayDisconnected() {
 	if conn.ctx.Err() != nil {
 		return
 	}
@@ -664,17 +835,9 @@ func (conn *Conn) handleRelayDisconnectedLocked() {
 	}
 }

-func (conn *Conn) onGuardEvent() {
-	conn.dumpState.SendOffer()
-	if err := conn.handshaker.SendOffer(); err != nil {
-		conn.Log.Errorf("failed to send offer: %v", err)
-	}
-}
-
-func (conn *Conn) onWGDisconnected() {
-	conn.mu.Lock()
-	defer conn.mu.Unlock()
-
+// handleWGTimeout closes the active connection after a WireGuard handshake
+// timeout so the guard can trigger a reconnection.
+func (conn *Conn) handleWGTimeout() {
 	if conn.ctx.Err() != nil {
 		return
 	}
@@ -685,23 +848,23 @@ func (conn *Conn) onWGDisconnected() {
 	switch conn.currentConnPriority {
 	case conntype.Relay:
 		conn.workerRelay.CloseConn()
-		conn.handleRelayDisconnectedLocked()
+		conn.handleRelayDisconnected()
 	case conntype.ICEP2P, conntype.ICETurn:
 		conn.workerICE.Close()
 	default:
 		conn.Log.Debugf("No active connection to close on WG timeout")
 	}

-	conn.escalateWGTimeoutLocked()
+	conn.escalateWGTimeout()
 }

-// escalateWGTimeoutLocked resets the peer's rosenpass state after repeated
+// escalateWGTimeout resets the peer's rosenpass state after repeated
 // handshake timeouts. With rosenpass enabled, persistent timeouts mean the
 // preshared keys have desynced; the renewal exchange runs over the dead
 // tunnel and cannot resync them. Reporting the peer disconnected drops its
 // rosenpass state, so the next connection configuration programs the
-// rendezvous key and the tunnel can bootstrap again. Callers must hold mu.
-func (conn *Conn) escalateWGTimeoutLocked() {
+// rendezvous key and the tunnel can bootstrap again. Runs on the event loop.
+func (conn *Conn) escalateWGTimeout() {
 	if conn.config.RosenpassConfig.PubKey == nil {
 		return
 	}
@@ -716,6 +879,83 @@ func (conn *Conn) escalateWGTimeoutLocked() {
 	conn.onDisconnected(conn.config.WgConfig.RemoteKey)
 }

+func (conn *Conn) handleWGHandshakeSuccess(when time.Time) {
+	conn.metricsStages.RecordWGHandshakeSuccess(when)
+	conn.recordConnectionMetrics()
+}
+
+func (conn *Conn) handleWGCheckSuccess() {
+	conn.wgTimeouts = 0
+}
+
+func (conn *Conn) onICEConnectionIsReady(priority conntype.ConnPriority, iceConnInfo ICEConnInfo) {
+	conn.post(evICEReady{priority: priority, info: iceConnInfo})
+}
+
+func (conn *Conn) onICEStateDisconnected(sessionChanged bool) {
+	conn.post(evICEDown{sessionChanged: sessionChanged})
+}
+
+// onRelayConnectionIsReady closes the relayed connection when the event loop
+// is gone and nobody will take ownership of it.
+func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) {
+	if conn.post(evRelayReady{info: rci}) {
+		return
+	}
+	if err := rci.relayedConn.Close(); err != nil {
+		conn.Log.Warnf("failed to close unnecessary relayed connection: %v", err)
+	}
+}
+
+func (conn *Conn) onRelayDisconnected() {
+	conn.post(evRelayDown{})
+}
+
+func (conn *Conn) onGuardEvent() {
+	conn.post(evGuardTick{})
+}
+
+func (conn *Conn) onWGDisconnected() {
+	conn.post(evWGTimeout{})
+}
+
+func (conn *Conn) onWGHandshakeSuccess(when time.Time) {
+	conn.post(evWGHandshake{when: when})
+}
+
+func (conn *Conn) onWGCheckSuccess() {
+	conn.post(evWGCheckOK{})
+}
+
+// injectPendingFirstPacket replays the captured handshake through the proxy if present, else
+// directly through the ICE conn. The packet is cleared only after a successful write, so a failed
+// or transport-less attempt leaves it available for a later reinjection. Runs on the event loop.
+func (conn *Conn) injectPendingFirstPacket(proxy wgproxy.Proxy, directConn net.Conn) {
+	pkt := conn.pendingFirstPacket
+	if len(pkt) == 0 {
+		return
+	}
+
+	switch {
+	case proxy != nil:
+		if err := proxy.InjectPacket(pkt); err != nil {
+			conn.Log.Debugf("failed to reinject captured first packet via proxy: %v", err)
+			return
+		}
+	case directConn != nil:
+		if _, err := directConn.Write(pkt); err != nil {
+			conn.Log.Debugf("failed to reinject captured first packet via direct conn: %v", err)
+			return
+		}
+	default:
+		conn.Log.Debugf("no transport available to reinject captured first packet")
+		return
+	}
+
+	conn.pendingFirstPacket = nil
+	conn.Log.Debugf("reinjected captured first packet (%d bytes)", len(pkt))
+}
+
 func (conn *Conn) updateRelayStatus(relayServerAddr string, rosenpassPubKey []byte, updateTime time.Time) {
 	peerState := State{
 		PubKey:             conn.config.Key,
@@ -917,34 +1157,14 @@ func (conn *Conn) setRelayedProxy(proxy wgproxy.Proxy) {
 	conn.wgProxyRelay = proxy
 }

-// onWGHandshakeSuccess is called when the first WireGuard handshake is detected
-func (conn *Conn) onWGHandshakeSuccess(when time.Time) {
-	conn.metricsStages.RecordWGHandshakeSuccess(when)
-	conn.recordConnectionMetrics()
-}
-
-// onWGCheckSuccess is called for every watcher check that observed a fresh
-// handshake, including handshakes of connections that were already up when
-// the watcher started.
-func (conn *Conn) onWGCheckSuccess() {
-	conn.mu.Lock()
-	conn.wgTimeouts = 0
-	conn.mu.Unlock()
-}
-
 // recordConnectionMetrics records connection stage timestamps as metrics
 func (conn *Conn) recordConnectionMetrics() {
 	if conn.metricsRecorder == nil {
 		return
 	}

-	// Determine connection type based on current priority
-	conn.mu.Lock()
-	priority := conn.currentConnPriority
-	conn.mu.Unlock()
-
 	var connType metrics.ConnectionType
-	switch priority {
+	switch conn.currentConnPriority {
 	case conntype.Relay:
 		connType = metrics.ConnectionTypeRelay
 	default:
@@ -961,15 +1181,6 @@ func (conn *Conn) recordConnectionMetrics() {
 	)
 }

-// AllowedIP returns the allowed IP of the remote peer
-func (conn *Conn) AllowedIP() netip.Addr {
-	return conn.config.WgConfig.AllowedIps[0].Addr()
-}
-
-func (conn *Conn) AgentVersionString() string {
-	return conn.config.AgentVersion
-}
-
 func (conn *Conn) presharedKey(remoteRosenpassKey []byte) *wgtypes.Key {
 	if conn.config.RosenpassConfig.PubKey == nil {
 		return conn.config.WgConfig.PreSharedKey
--- a/client/internal/peer/conn_test.go
+++ b/client/internal/peer/conn_test.go
@@ -3,28 +3,30 @@ 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/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 +54,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: 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{
+	offerAnswer := 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")
-	}
-}
-
-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) {
@@ -339,20 +286,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 +311,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 +329,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")
 }
--- a/client/internal/peer/dispatcher/dispatcher.go
+++ b/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)
-	}
-}
--- a/client/internal/peer/event.go
+++ b/client/internal/peer/event.go
@@ -0,0 +1,68 @@
+package peer
+
+import (
+	"time"
+
+	"github.com/pion/ice/v4"
+
+	"github.com/netbirdio/netbird/client/internal/peer/conntype"
+	"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 OfferAnswer
+}
+
+type evRemoteAnswer struct {
+	answer OfferAnswer
+}
+
+type evRemoteCandidate struct {
+	candidate ice.Candidate
+	haRoutes  route.HAMap
+}
+
+type evICEReady struct {
+	priority conntype.ConnPriority
+	info     ICEConnInfo
+}
+
+type evICEDown struct {
+	sessionChanged bool
+}
+
+type evRelayReady struct {
+	info 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{}
--- a/client/internal/peer/guard/ice_monitor.go
+++ b/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,
--- a/client/internal/peer/handshaker.go
+++ b/client/internal/peer/handshaker.go
@@ -1,7 +1,6 @@
 package peer

 import (
-	"context"
 	"errors"
 	"net/netip"
 	"sync"
@@ -53,42 +52,36 @@ 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 (o *OfferAnswer) SessionIDString() string {
+	if o.SessionID == nil {
+		return "unknown"
+	}
+	return o.SessionID.String()
 }

-func NewHandshaker(log *log.Entry, config ConnConfig, signaler *Signaler, ice *WorkerICE, relay *WorkerRelay, metricsStages *MetricsStages) *Handshaker {
+// 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   ConnConfig
+	signaler *Signaler
+	ice      *WorkerICE
+	relay    *WorkerRelay
+
+	// 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 ConnConfig, signaler *Signaler, ice *WorkerICE, relay *WorkerRelay) *Handshaker {
 	h := &Handshaker{
-		log:            log,
-		config:         config,
-		signaler:       signaler,
-		ice:            ice,
-		relay:          relay,
-		metricsStages:  metricsStages,
-		remoteOffersCh: make(chan OfferAnswer),
-		remoteAnswerCh: make(chan OfferAnswer),
+		log:      log,
+		config:   config,
+		signaler: signaler,
+		ice:      ice,
+		relay:    relay,
 	}
 	// assume remote supports ICE until we learn otherwise from received offers
 	h.remoteICESupported.Store(ice != nil)
@@ -99,93 +92,16 @@ 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
-	}
+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
@@ -230,6 +146,9 @@ func (h *Handshaker) buildOfferAnswer() OfferAnswer {
 	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)
--- a/client/internal/peer/handshaker_listener.go
+++ b/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)
-}
--- a/client/internal/peer/handshaker_listener_test.go
+++ b/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")
-	}
-}
--- a/client/internal/peer/mailbox.go
+++ b/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
+}
--- a/client/internal/peer/mailbox_test.go
+++ b/client/internal/peer/mailbox_test.go
@@ -0,0 +1,127 @@
+package peer
+
+import (
+	"testing"
+
+	"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: OfferAnswer{WgListenPort: 1}}))
+	require.True(t, mb.post(evRemoteOffer{offer: OfferAnswer{WgListenPort: 2}}))
+	require.True(t, mb.post(evRemoteOffer{offer: 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: 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: 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: OfferAnswer{}}))
+	require.True(t, mb.post(evRemoteOffer{offer: 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")
+}
--- a/client/internal/peer/wg_watcher_test.go
+++ b/client/internal/peer/wg_watcher_test.go
@@ -66,14 +66,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 +92,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) {
--- a/client/internal/peer/worker_relay.go
+++ b/client/internal/peer/worker_relay.go
@@ -88,7 +88,7 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 	}

 	w.log.Debugf("peer conn opened via Relay: %s", srv)
-	go w.conn.onRelayConnectionIsReady(RelayConnInfo{
+	w.conn.onRelayConnectionIsReady(RelayConnInfo{
 		relayedConn:     relayedConn,
 		rosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
 		rosenpassAddr:   remoteOfferAnswer.RosenpassAddr,
@@ -134,5 +134,5 @@ func (w *WorkerRelay) preferredRelayServer(myRelayAddress, remoteRelayAddress st
 }

 func (w *WorkerRelay) onRelayClientDisconnected() {
-	go w.conn.onRelayDisconnected()
+	w.conn.onRelayDisconnected()
 }