From 5f98524e02046789e22f6d23f9d0344addbbbf9b Mon Sep 17 00:00:00 2001 From: Zoltan Papp Date: Sat, 11 Jul 2026 13:30:06 +0200 Subject: [PATCH] [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. --- client/internal/peer/conn.go | 565 ++++++++++++------ client/internal/peer/conn_test.go | 105 +--- client/internal/peer/dispatcher/dispatcher.go | 52 -- client/internal/peer/event.go | 68 +++ client/internal/peer/guard/ice_monitor.go | 3 - client/internal/peer/handshaker.go | 149 ++--- client/internal/peer/handshaker_listener.go | 62 -- .../internal/peer/handshaker_listener_test.go | 39 -- client/internal/peer/mailbox.go | 116 ++++ client/internal/peer/mailbox_test.go | 127 ++++ client/internal/peer/wg_watcher_test.go | 25 +- client/internal/peer/worker_relay.go | 4 +- 12 files changed, 778 insertions(+), 537 deletions(-) delete mode 100644 client/internal/peer/dispatcher/dispatcher.go create mode 100644 client/internal/peer/event.go delete mode 100644 client/internal/peer/handshaker_listener.go delete mode 100644 client/internal/peer/handshaker_listener_test.go create mode 100644 client/internal/peer/mailbox.go create mode 100644 client/internal/peer/mailbox_test.go diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index f0625c853..9b9dd322b 100644 --- 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 diff --git a/client/internal/peer/conn_test.go b/client/internal/peer/conn_test.go index f2312a66a..b36bd164a 100644 --- 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") } diff --git a/client/internal/peer/dispatcher/dispatcher.go b/client/internal/peer/dispatcher/dispatcher.go deleted file mode 100644 index 06124bc35..000000000 --- a/client/internal/peer/dispatcher/dispatcher.go +++ /dev/null @@ -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) - } -} diff --git a/client/internal/peer/event.go b/client/internal/peer/event.go new file mode 100644 index 000000000..ed5507326 --- /dev/null +++ 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{} diff --git a/client/internal/peer/guard/ice_monitor.go b/client/internal/peer/guard/ice_monitor.go index a201dd095..98fd8808a 100644 --- 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, diff --git a/client/internal/peer/handshaker.go b/client/internal/peer/handshaker.go index 56e82e6e3..73b8445ca 100644 --- 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) diff --git a/client/internal/peer/handshaker_listener.go b/client/internal/peer/handshaker_listener.go deleted file mode 100644 index 772e2777f..000000000 --- a/client/internal/peer/handshaker_listener.go +++ /dev/null @@ -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) -} diff --git a/client/internal/peer/handshaker_listener_test.go b/client/internal/peer/handshaker_listener_test.go deleted file mode 100644 index 1a7156d10..000000000 --- a/client/internal/peer/handshaker_listener_test.go +++ /dev/null @@ -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") - } -} diff --git a/client/internal/peer/mailbox.go b/client/internal/peer/mailbox.go new file mode 100644 index 000000000..53895819b --- /dev/null +++ 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 +} diff --git a/client/internal/peer/mailbox_test.go b/client/internal/peer/mailbox_test.go new file mode 100644 index 000000000..287647bc4 --- /dev/null +++ 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") +} diff --git a/client/internal/peer/wg_watcher_test.go b/client/internal/peer/wg_watcher_test.go index 80f34f1a1..f9dfe71e0 100644 --- 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) { diff --git a/client/internal/peer/worker_relay.go b/client/internal/peer/worker_relay.go index 0402992c9..273d16120 100644 --- 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() }