From be42b0b12d2bb565e9e6cbdbcc26d022dc2c435a Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Thu, 26 Sep 2024 15:55:34 +0200
Subject: [PATCH 01/14] Limit P2P attempts and restart on specific events

---
 client/internal/peer/conn.go           | 231 ++++++++++++++++++++++---
 client/internal/peer/stdnet.go         |   4 +-
 client/internal/peer/stdnet_android.go |   4 +-
 client/internal/peer/worker_ice.go     |  63 +++----
 4 files changed, 241 insertions(+), 61 deletions(-)

diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go
index 911ddd2281c..9d86b990f85 100644
--- a/client/internal/peer/conn.go
+++ b/client/internal/peer/conn.go
@@ -2,6 +2,7 @@ package peer
 
 import (
 	"context"
+	"fmt"
 	"math/rand"
 	"net"
 	"os"
@@ -31,6 +32,10 @@ const (
 	connPriorityRelay   ConnPriority = 1
 	connPriorityICETurn ConnPriority = 1
 	connPriorityICEP2P  ConnPriority = 2
+
+	reconnectMaxElapsedTime    = 30 * time.Minute
+	candidatesMonitorPeriod    = 5 * time.Minute
+	candidatedGatheringTimeout = 5 * time.Second
 )
 
 type WgConfig struct {
@@ -82,6 +87,7 @@ type Conn struct {
 	wgProxyICE     wgproxy.Proxy
 	wgProxyRelay   wgproxy.Proxy
 	signaler       *Signaler
+	iFaceDiscover  stdnet.ExternalIFaceDiscover
 	relayManager   *relayClient.Manager
 	allowedIPsIP   string
 	handshaker     *Handshaker
@@ -107,6 +113,10 @@ type Conn struct {
 	// for reconnection operations
 	iCEDisconnected   chan bool
 	relayDisconnected chan bool
+	reconnectCh       chan struct{}
+
+	currentCandidates []ice.Candidate
+	candidatesMu      sync.Mutex
 }
 
 // NewConn creates a new not opened Conn to the remote peer.
@@ -122,19 +132,22 @@ func NewConn(engineCtx context.Context, config ConnConfig, statusRecorder *Statu
 	connLog := log.WithField("peer", config.Key)
 
 	var conn = &Conn{
-		log:               connLog,
-		ctx:               ctx,
-		ctxCancel:         ctxCancel,
-		config:            config,
-		statusRecorder:    statusRecorder,
-		wgProxyFactory:    wgProxyFactory,
-		signaler:          signaler,
-		relayManager:      relayManager,
-		allowedIPsIP:      allowedIPsIP.String(),
-		statusRelay:       NewAtomicConnStatus(),
-		statusICE:         NewAtomicConnStatus(),
+		log:            connLog,
+		ctx:            ctx,
+		ctxCancel:      ctxCancel,
+		config:         config,
+		statusRecorder: statusRecorder,
+		wgProxyFactory: wgProxyFactory,
+		signaler:       signaler,
+		iFaceDiscover:  iFaceDiscover,
+		relayManager:   relayManager,
+		allowedIPsIP:   allowedIPsIP.String(),
+		statusRelay:    NewAtomicConnStatus(),
+		statusICE:      NewAtomicConnStatus(),
+
 		iCEDisconnected:   make(chan bool, 1),
 		relayDisconnected: make(chan bool, 1),
+		reconnectCh:       make(chan struct{}, 1),
 	}
 
 	rFns := WorkerRelayCallbacks{
@@ -305,21 +318,23 @@ func (conn *Conn) GetKey() string {
 
 func (conn *Conn) reconnectLoopWithRetry() {
 	// Give chance to the peer to establish the initial connection.
-	// With it, we can decrease to send necessary offer
 	select {
 	case <-conn.ctx.Done():
+		return
 	case <-time.After(3 * time.Second):
 	}
 
+	go conn.monitorReconnectEvents()
+
 	ticker := conn.prepareExponentTicker()
 	defer ticker.Stop()
-	time.Sleep(1 * time.Second)
+
 	for {
 		select {
 		case t := <-ticker.C:
 			if t.IsZero() {
 				// in case if the ticker has been canceled by context then avoid the temporary loop
-				return
+				continue
 			}
 
 			if conn.workerRelay.IsRelayConnectionSupportedWithPeer() {
@@ -341,20 +356,12 @@ func (conn *Conn) reconnectLoopWithRetry() {
 			if err != nil {
 				conn.log.Errorf("failed to do handshake: %v", err)
 			}
-		case changed := <-conn.relayDisconnected:
-			if !changed {
-				continue
-			}
-			conn.log.Debugf("Relay state changed, reset reconnect timer")
-			ticker.Stop()
-			ticker = conn.prepareExponentTicker()
-		case changed := <-conn.iCEDisconnected:
-			if !changed {
-				continue
-			}
-			conn.log.Debugf("ICE state changed, reset reconnect timer")
+
+		case <-conn.reconnectCh:
+			conn.log.Debugf("Reconnect event received, resetting reconnect timer")
 			ticker.Stop()
 			ticker = conn.prepareExponentTicker()
+
 		case <-conn.ctx.Done():
 			conn.log.Debugf("context is done, stop reconnect loop")
 			return
@@ -365,10 +372,10 @@ func (conn *Conn) reconnectLoopWithRetry() {
 func (conn *Conn) prepareExponentTicker() *backoff.Ticker {
 	bo := backoff.WithContext(&backoff.ExponentialBackOff{
 		InitialInterval:     800 * time.Millisecond,
-		RandomizationFactor: 0.01,
+		RandomizationFactor: 0.1,
 		Multiplier:          2,
 		MaxInterval:         conn.config.Timeout,
-		MaxElapsedTime:      0,
+		MaxElapsedTime:      reconnectMaxElapsedTime,
 		Stop:                backoff.Stop,
 		Clock:               backoff.SystemClock,
 	}, conn.ctx)
@@ -379,6 +386,174 @@ func (conn *Conn) prepareExponentTicker() *backoff.Ticker {
 	return ticker
 }
 
+func (conn *Conn) monitorReconnectEvents() {
+	signalerReady := make(chan struct{}, 1)
+	go conn.monitorSignalerReady(signalerReady)
+
+	localCandidatesChanged := make(chan struct{}, 1)
+	go conn.monitorLocalCandidatesChanged(localCandidatesChanged)
+
+	for {
+		select {
+		case changed := <-conn.relayDisconnected:
+			if !changed {
+				continue
+			}
+
+			conn.log.Debugf("Relay state changed, triggering reconnect")
+			conn.triggerReconnect()
+
+		case changed := <-conn.iCEDisconnected:
+			if !changed {
+				continue
+			}
+
+			conn.log.Debugf("ICE state changed, triggering reconnect")
+			conn.triggerReconnect()
+
+		case <-signalerReady:
+			conn.log.Debugf("Signaler became ready, triggering reconnect")
+			conn.triggerReconnect()
+
+		case <-localCandidatesChanged:
+			conn.log.Debugf("Local candidates changed, triggering reconnect")
+			conn.triggerReconnect()
+
+		case <-conn.ctx.Done():
+			return
+		}
+	}
+}
+
+// monitorSignalerReady monitors the signaler ready state and triggers reconnect when it transitions from not ready to ready
+func (conn *Conn) monitorSignalerReady(signalerReady chan<- struct{}) {
+	ticker := time.NewTicker(signalerMonitorPeriod)
+	defer ticker.Stop()
+
+	lastReady := true
+	for {
+		select {
+		case <-ticker.C:
+			currentReady := conn.signaler.Ready()
+			if !lastReady && currentReady {
+				select {
+				case signalerReady <- struct{}{}:
+				default:
+				}
+			}
+			lastReady = currentReady
+		case <-conn.ctx.Done():
+			return
+		}
+	}
+}
+
+// monitorLocalCandidatesChanged monitors the local candidates and triggers reconnect when they change
+func (conn *Conn) monitorLocalCandidatesChanged(localCandidatesChanged chan<- struct{}) {
+	// TODO: make this global and not per-conn
+
+	ufrag, pwd, err := generateICECredentials()
+	if err != nil {
+		conn.log.Warnf("Failed to generate ICE credentials: %v", err)
+		return
+	}
+
+	ticker := time.NewTicker(candidatesMonitorPeriod)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			if err := conn.handleCandidateTick(localCandidatesChanged, ufrag, pwd); err != nil {
+				conn.log.Warnf("Failed to handle candidate tick: %v", err)
+			}
+		case <-conn.ctx.Done():
+			return
+		}
+	}
+}
+
+func (conn *Conn) handleCandidateTick(localCandidatesChanged chan<- struct{}, ufrag string, pwd string) error {
+	conn.log.Debugf("Gathering ICE candidates")
+
+	transportNet, err := newStdNet(conn.iFaceDiscover, conn.config.ICEConfig.InterfaceBlackList)
+	if err != nil {
+		conn.log.Errorf("failed to create pion's stdnet: %s", err)
+	}
+
+	agent, err := newAgent(conn.config, transportNet, candidateTypes(), ufrag, pwd)
+	if err != nil {
+		return fmt.Errorf("create ICE agent: %w", err)
+	}
+	defer func() {
+		if err := agent.Close(); err != nil {
+			conn.log.Warnf("Failed to close ICE agent: %v", err)
+		}
+	}()
+
+	gatherDone := make(chan struct{})
+	agent.OnCandidate(func(c ice.Candidate) {
+		log.Debugf("Got candidate: %v", c)
+		if c == nil {
+			close(gatherDone)
+		}
+	})
+
+	if err := agent.GatherCandidates(); err != nil {
+		return fmt.Errorf("gather ICE candidates: %w", err)
+	}
+
+	ctx, cancel := context.WithTimeout(conn.ctx, candidatedGatheringTimeout)
+	defer cancel()
+
+	select {
+	case <-ctx.Done():
+		return fmt.Errorf("wait for gathering: %w", ctx.Err())
+	case <-gatherDone:
+	}
+
+	candidates, err := agent.GetLocalCandidates()
+	if err != nil {
+		return fmt.Errorf("get local candidates: %w", err)
+	}
+	log.Debugf("Got candidates: %v", candidates)
+
+	if changed := conn.updateCandidates(candidates); changed {
+		select {
+		case localCandidatesChanged <- struct{}{}:
+		default:
+		}
+	}
+
+	return nil
+}
+
+func (conn *Conn) updateCandidates(newCandidates []ice.Candidate) bool {
+	conn.candidatesMu.Lock()
+	defer conn.candidatesMu.Unlock()
+
+	if len(conn.currentCandidates) != len(newCandidates) {
+		conn.currentCandidates = newCandidates
+		return true
+	}
+
+	for i, candidate := range conn.currentCandidates {
+		if candidate.String() != newCandidates[i].String() {
+			conn.currentCandidates = newCandidates
+			return true
+		}
+	}
+
+	return false
+}
+
+func (conn *Conn) triggerReconnect() {
+	select {
+	case conn.reconnectCh <- struct{}{}:
+	default:
+	}
+}
+
 // reconnectLoopForOnDisconnectedEvent is used when the peer is not a controller and it should reconnect to the peer
 // when the connection is lost. It will try to establish a connection only once time if before the connection was established
 // It track separately the ice and relay connection status. Just because a lover priority connection reestablished it does not
diff --git a/client/internal/peer/stdnet.go b/client/internal/peer/stdnet.go
index ae31ebbf067..96d211dbc77 100644
--- a/client/internal/peer/stdnet.go
+++ b/client/internal/peer/stdnet.go
@@ -6,6 +6,6 @@ import (
 	"github.com/netbirdio/netbird/client/internal/stdnet"
 )
 
-func (w *WorkerICE) newStdNet() (*stdnet.Net, error) {
-	return stdnet.NewNet(w.config.ICEConfig.InterfaceBlackList)
+func newStdNet(_ stdnet.ExternalIFaceDiscover, ifaceBlacklist []string) (*stdnet.Net, error) {
+	return stdnet.NewNet(ifaceBlacklist)
 }
diff --git a/client/internal/peer/stdnet_android.go b/client/internal/peer/stdnet_android.go
index b411405bb95..a39a03b1c83 100644
--- a/client/internal/peer/stdnet_android.go
+++ b/client/internal/peer/stdnet_android.go
@@ -2,6 +2,6 @@ package peer
 
 import "github.com/netbirdio/netbird/client/internal/stdnet"
 
-func (w *WorkerICE) newStdNet() (*stdnet.Net, error) {
-	return stdnet.NewNetWithDiscover(w.iFaceDiscover, w.config.ICEConfig.InterfaceBlackList)
+func newStdNet(iFaceDiscover stdnet.ExternalIFaceDiscover, ifaceBlacklist []string) (*stdnet.Net, error) {
+	return stdnet.NewNetWithDiscover(iFaceDiscover, ifaceBlacklist)
 }
diff --git a/client/internal/peer/worker_ice.go b/client/internal/peer/worker_ice.go
index 8bf1b75684a..b5935557393 100644
--- a/client/internal/peer/worker_ice.go
+++ b/client/internal/peer/worker_ice.go
@@ -233,41 +233,16 @@ func (w *WorkerICE) Close() {
 }
 
 func (w *WorkerICE) reCreateAgent(agentCancel context.CancelFunc, relaySupport []ice.CandidateType) (*ice.Agent, error) {
-	transportNet, err := w.newStdNet()
+	transportNet, err := newStdNet(w.iFaceDiscover, w.config.ICEConfig.InterfaceBlackList)
 	if err != nil {
 		w.log.Errorf("failed to create pion's stdnet: %s", err)
 	}
 
-	iceKeepAlive := iceKeepAlive()
-	iceDisconnectedTimeout := iceDisconnectedTimeout()
-	iceRelayAcceptanceMinWait := iceRelayAcceptanceMinWait()
-
-	agentConfig := &ice.AgentConfig{
-		MulticastDNSMode:       ice.MulticastDNSModeDisabled,
-		NetworkTypes:           []ice.NetworkType{ice.NetworkTypeUDP4, ice.NetworkTypeUDP6},
-		Urls:                   w.config.ICEConfig.StunTurn.Load().([]*stun.URI),
-		CandidateTypes:         relaySupport,
-		InterfaceFilter:        stdnet.InterfaceFilter(w.config.ICEConfig.InterfaceBlackList),
-		UDPMux:                 w.config.ICEConfig.UDPMux,
-		UDPMuxSrflx:            w.config.ICEConfig.UDPMuxSrflx,
-		NAT1To1IPs:             w.config.ICEConfig.NATExternalIPs,
-		Net:                    transportNet,
-		FailedTimeout:          &failedTimeout,
-		DisconnectedTimeout:    &iceDisconnectedTimeout,
-		KeepaliveInterval:      &iceKeepAlive,
-		RelayAcceptanceMinWait: &iceRelayAcceptanceMinWait,
-		LocalUfrag:             w.localUfrag,
-		LocalPwd:               w.localPwd,
-	}
-
-	if w.config.ICEConfig.DisableIPv6Discovery {
-		agentConfig.NetworkTypes = []ice.NetworkType{ice.NetworkTypeUDP4}
-	}
-
 	w.sentExtraSrflx = false
-	agent, err := ice.NewAgent(agentConfig)
+
+	agent, err := newAgent(w.config, transportNet, relaySupport, w.localUfrag, w.localPwd)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("create agent: %w", err)
 	}
 
 	err = agent.OnCandidate(w.onICECandidate)
@@ -390,6 +365,36 @@ func (w *WorkerICE) turnAgentDial(ctx context.Context, remoteOfferAnswer *OfferA
 	}
 }
 
+func newAgent(config ConnConfig, transportNet *stdnet.Net, candidateTypes []ice.CandidateType, ufrag string, pwd string) (*ice.Agent, error) {
+	iceKeepAlive := iceKeepAlive()
+	iceDisconnectedTimeout := iceDisconnectedTimeout()
+	iceRelayAcceptanceMinWait := iceRelayAcceptanceMinWait()
+
+	agentConfig := &ice.AgentConfig{
+		MulticastDNSMode:       ice.MulticastDNSModeDisabled,
+		NetworkTypes:           []ice.NetworkType{ice.NetworkTypeUDP4, ice.NetworkTypeUDP6},
+		Urls:                   config.ICEConfig.StunTurn.Load().([]*stun.URI),
+		CandidateTypes:         candidateTypes,
+		InterfaceFilter:        stdnet.InterfaceFilter(config.ICEConfig.InterfaceBlackList),
+		UDPMux:                 config.ICEConfig.UDPMux,
+		UDPMuxSrflx:            config.ICEConfig.UDPMuxSrflx,
+		NAT1To1IPs:             config.ICEConfig.NATExternalIPs,
+		Net:                    transportNet,
+		FailedTimeout:          &failedTimeout,
+		DisconnectedTimeout:    &iceDisconnectedTimeout,
+		KeepaliveInterval:      &iceKeepAlive,
+		RelayAcceptanceMinWait: &iceRelayAcceptanceMinWait,
+		LocalUfrag:             ufrag,
+		LocalPwd:               pwd,
+	}
+
+	if config.ICEConfig.DisableIPv6Discovery {
+		agentConfig.NetworkTypes = []ice.NetworkType{ice.NetworkTypeUDP4}
+	}
+
+	return ice.NewAgent(agentConfig)
+}
+
 func extraSrflxCandidate(candidate ice.Candidate) (*ice.CandidateServerReflexive, error) {
 	relatedAdd := candidate.RelatedAddress()
 	return ice.NewCandidateServerReflexive(&ice.CandidateServerReflexiveConfig{

From 120710f749d22f63238b35fce18c269e778b1b0a Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Fri, 27 Sep 2024 18:59:13 +0200
Subject: [PATCH 02/14] Add missing const

---
 client/internal/peer/conn.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go
index 9d86b990f85..676eab99917 100644
--- a/client/internal/peer/conn.go
+++ b/client/internal/peer/conn.go
@@ -36,6 +36,7 @@ const (
 	reconnectMaxElapsedTime    = 30 * time.Minute
 	candidatesMonitorPeriod    = 5 * time.Minute
 	candidatedGatheringTimeout = 5 * time.Second
+	signalerMonitorPeriod      = 5 * time.Second
 )
 
 type WgConfig struct {

From a261d07ef4bad0808add3ca2e4e03fa30d2a7c07 Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Mon, 30 Sep 2024 10:04:26 +0200
Subject: [PATCH 03/14] Handle error

---
 client/internal/peer/conn.go | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go
index 676eab99917..03fd49b0e30 100644
--- a/client/internal/peer/conn.go
+++ b/client/internal/peer/conn.go
@@ -493,12 +493,15 @@ func (conn *Conn) handleCandidateTick(localCandidatesChanged chan<- struct{}, uf
 	}()
 
 	gatherDone := make(chan struct{})
-	agent.OnCandidate(func(c ice.Candidate) {
+	err = agent.OnCandidate(func(c ice.Candidate) {
 		log.Debugf("Got candidate: %v", c)
 		if c == nil {
 			close(gatherDone)
 		}
 	})
+	if err != nil {
+		return fmt.Errorf("set ICE candidate handler: %w", err)
+	}
 
 	if err := agent.GatherCandidates(); err != nil {
 		return fmt.Errorf("gather ICE candidates: %w", err)

From 6c60d04acb48635904b5f0dba74b17d2ca607186 Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Mon, 30 Sep 2024 10:09:19 +0200
Subject: [PATCH 04/14] Move code to separate file

---
 client/internal/peer/conn.go         | 173 -------------------------
 client/internal/peer/conn_monitor.go | 181 +++++++++++++++++++++++++++
 2 files changed, 181 insertions(+), 173 deletions(-)
 create mode 100644 client/internal/peer/conn_monitor.go

diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go
index 03fd49b0e30..768498c70f0 100644
--- a/client/internal/peer/conn.go
+++ b/client/internal/peer/conn.go
@@ -2,7 +2,6 @@ package peer
 
 import (
 	"context"
-	"fmt"
 	"math/rand"
 	"net"
 	"os"
@@ -359,7 +358,6 @@ func (conn *Conn) reconnectLoopWithRetry() {
 			}
 
 		case <-conn.reconnectCh:
-			conn.log.Debugf("Reconnect event received, resetting reconnect timer")
 			ticker.Stop()
 			ticker = conn.prepareExponentTicker()
 
@@ -387,177 +385,6 @@ func (conn *Conn) prepareExponentTicker() *backoff.Ticker {
 	return ticker
 }
 
-func (conn *Conn) monitorReconnectEvents() {
-	signalerReady := make(chan struct{}, 1)
-	go conn.monitorSignalerReady(signalerReady)
-
-	localCandidatesChanged := make(chan struct{}, 1)
-	go conn.monitorLocalCandidatesChanged(localCandidatesChanged)
-
-	for {
-		select {
-		case changed := <-conn.relayDisconnected:
-			if !changed {
-				continue
-			}
-
-			conn.log.Debugf("Relay state changed, triggering reconnect")
-			conn.triggerReconnect()
-
-		case changed := <-conn.iCEDisconnected:
-			if !changed {
-				continue
-			}
-
-			conn.log.Debugf("ICE state changed, triggering reconnect")
-			conn.triggerReconnect()
-
-		case <-signalerReady:
-			conn.log.Debugf("Signaler became ready, triggering reconnect")
-			conn.triggerReconnect()
-
-		case <-localCandidatesChanged:
-			conn.log.Debugf("Local candidates changed, triggering reconnect")
-			conn.triggerReconnect()
-
-		case <-conn.ctx.Done():
-			return
-		}
-	}
-}
-
-// monitorSignalerReady monitors the signaler ready state and triggers reconnect when it transitions from not ready to ready
-func (conn *Conn) monitorSignalerReady(signalerReady chan<- struct{}) {
-	ticker := time.NewTicker(signalerMonitorPeriod)
-	defer ticker.Stop()
-
-	lastReady := true
-	for {
-		select {
-		case <-ticker.C:
-			currentReady := conn.signaler.Ready()
-			if !lastReady && currentReady {
-				select {
-				case signalerReady <- struct{}{}:
-				default:
-				}
-			}
-			lastReady = currentReady
-		case <-conn.ctx.Done():
-			return
-		}
-	}
-}
-
-// monitorLocalCandidatesChanged monitors the local candidates and triggers reconnect when they change
-func (conn *Conn) monitorLocalCandidatesChanged(localCandidatesChanged chan<- struct{}) {
-	// TODO: make this global and not per-conn
-
-	ufrag, pwd, err := generateICECredentials()
-	if err != nil {
-		conn.log.Warnf("Failed to generate ICE credentials: %v", err)
-		return
-	}
-
-	ticker := time.NewTicker(candidatesMonitorPeriod)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ticker.C:
-			if err := conn.handleCandidateTick(localCandidatesChanged, ufrag, pwd); err != nil {
-				conn.log.Warnf("Failed to handle candidate tick: %v", err)
-			}
-		case <-conn.ctx.Done():
-			return
-		}
-	}
-}
-
-func (conn *Conn) handleCandidateTick(localCandidatesChanged chan<- struct{}, ufrag string, pwd string) error {
-	conn.log.Debugf("Gathering ICE candidates")
-
-	transportNet, err := newStdNet(conn.iFaceDiscover, conn.config.ICEConfig.InterfaceBlackList)
-	if err != nil {
-		conn.log.Errorf("failed to create pion's stdnet: %s", err)
-	}
-
-	agent, err := newAgent(conn.config, transportNet, candidateTypes(), ufrag, pwd)
-	if err != nil {
-		return fmt.Errorf("create ICE agent: %w", err)
-	}
-	defer func() {
-		if err := agent.Close(); err != nil {
-			conn.log.Warnf("Failed to close ICE agent: %v", err)
-		}
-	}()
-
-	gatherDone := make(chan struct{})
-	err = agent.OnCandidate(func(c ice.Candidate) {
-		log.Debugf("Got candidate: %v", c)
-		if c == nil {
-			close(gatherDone)
-		}
-	})
-	if err != nil {
-		return fmt.Errorf("set ICE candidate handler: %w", err)
-	}
-
-	if err := agent.GatherCandidates(); err != nil {
-		return fmt.Errorf("gather ICE candidates: %w", err)
-	}
-
-	ctx, cancel := context.WithTimeout(conn.ctx, candidatedGatheringTimeout)
-	defer cancel()
-
-	select {
-	case <-ctx.Done():
-		return fmt.Errorf("wait for gathering: %w", ctx.Err())
-	case <-gatherDone:
-	}
-
-	candidates, err := agent.GetLocalCandidates()
-	if err != nil {
-		return fmt.Errorf("get local candidates: %w", err)
-	}
-	log.Debugf("Got candidates: %v", candidates)
-
-	if changed := conn.updateCandidates(candidates); changed {
-		select {
-		case localCandidatesChanged <- struct{}{}:
-		default:
-		}
-	}
-
-	return nil
-}
-
-func (conn *Conn) updateCandidates(newCandidates []ice.Candidate) bool {
-	conn.candidatesMu.Lock()
-	defer conn.candidatesMu.Unlock()
-
-	if len(conn.currentCandidates) != len(newCandidates) {
-		conn.currentCandidates = newCandidates
-		return true
-	}
-
-	for i, candidate := range conn.currentCandidates {
-		if candidate.String() != newCandidates[i].String() {
-			conn.currentCandidates = newCandidates
-			return true
-		}
-	}
-
-	return false
-}
-
-func (conn *Conn) triggerReconnect() {
-	select {
-	case conn.reconnectCh <- struct{}{}:
-	default:
-	}
-}
-
 // reconnectLoopForOnDisconnectedEvent is used when the peer is not a controller and it should reconnect to the peer
 // when the connection is lost. It will try to establish a connection only once time if before the connection was established
 // It track separately the ice and relay connection status. Just because a lover priority connection reestablished it does not
diff --git a/client/internal/peer/conn_monitor.go b/client/internal/peer/conn_monitor.go
new file mode 100644
index 00000000000..97b5dd82fb3
--- /dev/null
+++ b/client/internal/peer/conn_monitor.go
@@ -0,0 +1,181 @@
+package peer
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/pion/ice/v3"
+	log "github.com/sirupsen/logrus"
+)
+
+func (conn *Conn) monitorReconnectEvents() {
+	signalerReady := make(chan struct{}, 1)
+	go conn.monitorSignalerReady(signalerReady)
+
+	localCandidatesChanged := make(chan struct{}, 1)
+	go conn.monitorLocalCandidatesChanged(localCandidatesChanged)
+
+	for {
+		select {
+		case changed := <-conn.relayDisconnected:
+			if !changed {
+				continue
+			}
+
+			conn.log.Debugf("Relay state changed, triggering reconnect")
+			conn.triggerReconnect()
+
+		case changed := <-conn.iCEDisconnected:
+			if !changed {
+				continue
+			}
+
+			conn.log.Debugf("ICE state changed, triggering reconnect")
+			conn.triggerReconnect()
+
+		case <-signalerReady:
+			conn.log.Debugf("Signaler became ready, triggering reconnect")
+			conn.triggerReconnect()
+
+		case <-localCandidatesChanged:
+			conn.log.Debugf("Local candidates changed, triggering reconnect")
+			conn.triggerReconnect()
+
+		case <-conn.ctx.Done():
+			return
+		}
+	}
+}
+
+// monitorSignalerReady monitors the signaler ready state and triggers reconnect when it transitions from not ready to ready
+func (conn *Conn) monitorSignalerReady(signalerReady chan<- struct{}) {
+	ticker := time.NewTicker(signalerMonitorPeriod)
+	defer ticker.Stop()
+
+	lastReady := true
+	for {
+		select {
+		case <-ticker.C:
+			currentReady := conn.signaler.Ready()
+			if !lastReady && currentReady {
+				select {
+				case signalerReady <- struct{}{}:
+				default:
+				}
+			}
+			lastReady = currentReady
+		case <-conn.ctx.Done():
+			return
+		}
+	}
+}
+
+// monitorLocalCandidatesChanged monitors the local candidates and triggers reconnect when they change
+func (conn *Conn) monitorLocalCandidatesChanged(localCandidatesChanged chan<- struct{}) {
+	// TODO: make this global and not per-conn
+
+	ufrag, pwd, err := generateICECredentials()
+	if err != nil {
+		conn.log.Warnf("Failed to generate ICE credentials: %v", err)
+		return
+	}
+
+	ticker := time.NewTicker(candidatesMonitorPeriod)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			if err := conn.handleCandidateTick(localCandidatesChanged, ufrag, pwd); err != nil {
+				conn.log.Warnf("Failed to handle candidate tick: %v", err)
+			}
+		case <-conn.ctx.Done():
+			return
+		}
+	}
+}
+
+func (conn *Conn) handleCandidateTick(localCandidatesChanged chan<- struct{}, ufrag string, pwd string) error {
+	conn.log.Debugf("Gathering ICE candidates")
+
+	transportNet, err := newStdNet(conn.iFaceDiscover, conn.config.ICEConfig.InterfaceBlackList)
+	if err != nil {
+		conn.log.Errorf("failed to create pion's stdnet: %s", err)
+	}
+
+	agent, err := newAgent(conn.config, transportNet, candidateTypes(), ufrag, pwd)
+	if err != nil {
+		return fmt.Errorf("create ICE agent: %w", err)
+	}
+	defer func() {
+		if err := agent.Close(); err != nil {
+			conn.log.Warnf("Failed to close ICE agent: %v", err)
+		}
+	}()
+
+	gatherDone := make(chan struct{})
+	err = agent.OnCandidate(func(c ice.Candidate) {
+		log.Debugf("Got candidate: %v", c)
+		if c == nil {
+			close(gatherDone)
+		}
+	})
+	if err != nil {
+		return fmt.Errorf("set ICE candidate handler: %w", err)
+	}
+
+	if err := agent.GatherCandidates(); err != nil {
+		return fmt.Errorf("gather ICE candidates: %w", err)
+	}
+
+	ctx, cancel := context.WithTimeout(conn.ctx, candidatedGatheringTimeout)
+	defer cancel()
+
+	select {
+	case <-ctx.Done():
+		return fmt.Errorf("wait for gathering: %w", ctx.Err())
+	case <-gatherDone:
+	}
+
+	candidates, err := agent.GetLocalCandidates()
+	if err != nil {
+		return fmt.Errorf("get local candidates: %w", err)
+	}
+	log.Debugf("Got candidates: %v", candidates)
+
+	if changed := conn.updateCandidates(candidates); changed {
+		select {
+		case localCandidatesChanged <- struct{}{}:
+		default:
+		}
+	}
+
+	return nil
+}
+
+func (conn *Conn) updateCandidates(newCandidates []ice.Candidate) bool {
+	conn.candidatesMu.Lock()
+	defer conn.candidatesMu.Unlock()
+
+	if len(conn.currentCandidates) != len(newCandidates) {
+		conn.currentCandidates = newCandidates
+		return true
+	}
+
+	for i, candidate := range conn.currentCandidates {
+		if candidate.String() != newCandidates[i].String() {
+			conn.currentCandidates = newCandidates
+			return true
+		}
+	}
+
+	return false
+}
+
+func (conn *Conn) triggerReconnect() {
+	select {
+	case conn.reconnectCh <- struct{}{}:
+	default:
+	}
+}

From be41a238ad3294198650bb5f9e8a44d243a4291e Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Mon, 30 Sep 2024 10:11:10 +0200
Subject: [PATCH 05/14] Fix return

---
 client/internal/peer/conn.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go
index 768498c70f0..db484c29fc4 100644
--- a/client/internal/peer/conn.go
+++ b/client/internal/peer/conn.go
@@ -334,7 +334,7 @@ func (conn *Conn) reconnectLoopWithRetry() {
 		case t := <-ticker.C:
 			if t.IsZero() {
 				// in case if the ticker has been canceled by context then avoid the temporary loop
-				continue
+				return
 			}
 
 			if conn.workerRelay.IsRelayConnectionSupportedWithPeer() {

From 34b7fb5ca8425b8af5801a7cbb0dc708818923e6 Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Mon, 30 Sep 2024 10:11:33 +0200
Subject: [PATCH 06/14] Add back sleep

---
 client/internal/peer/conn.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go
index db484c29fc4..b687da9e7ea 100644
--- a/client/internal/peer/conn.go
+++ b/client/internal/peer/conn.go
@@ -328,6 +328,7 @@ func (conn *Conn) reconnectLoopWithRetry() {
 
 	ticker := conn.prepareExponentTicker()
 	defer ticker.Stop()
+	time.Sleep(1 * time.Second)
 
 	for {
 		select {

From 1fc1cbe8336c75b6f2895cbdb21b36e4dd1cb2ab Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Mon, 30 Sep 2024 10:12:27 +0200
Subject: [PATCH 07/14] Add missing comment

---
 client/internal/peer/conn.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go
index b687da9e7ea..f15eaba2ee6 100644
--- a/client/internal/peer/conn.go
+++ b/client/internal/peer/conn.go
@@ -318,6 +318,7 @@ func (conn *Conn) GetKey() string {
 
 func (conn *Conn) reconnectLoopWithRetry() {
 	// Give chance to the peer to establish the initial connection.
+	// With it, we can decrease to send necessary offer
 	select {
 	case <-conn.ctx.Done():
 		return

From 7fc81807e8f275fd9744293d87090e3ca264d796 Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Tue, 1 Oct 2024 12:20:47 +0200
Subject: [PATCH 08/14] Don't consider relay candidates

---
 client/internal/peer/conn_monitor.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/client/internal/peer/conn_monitor.go b/client/internal/peer/conn_monitor.go
index 97b5dd82fb3..a9e89ff46ec 100644
--- a/client/internal/peer/conn_monitor.go
+++ b/client/internal/peer/conn_monitor.go
@@ -104,7 +104,7 @@ func (conn *Conn) handleCandidateTick(localCandidatesChanged chan<- struct{}, uf
 		conn.log.Errorf("failed to create pion's stdnet: %s", err)
 	}
 
-	agent, err := newAgent(conn.config, transportNet, candidateTypes(), ufrag, pwd)
+	agent, err := newAgent(conn.config, transportNet, candidateTypesP2P(), ufrag, pwd)
 	if err != nil {
 		return fmt.Errorf("create ICE agent: %w", err)
 	}

From efe3a04adac0260c420d4fd08add443716ce181d Mon Sep 17 00:00:00 2001
From: Viktor Liu <17948409+lixmal@users.noreply.github.com>
Date: Tue, 1 Oct 2024 15:28:12 +0200
Subject: [PATCH 09/14] Update client/internal/peer/conn_monitor.go

Co-authored-by: Maycon Santos <mlsmaycon@gmail.com>
---
 client/internal/peer/conn_monitor.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/client/internal/peer/conn_monitor.go b/client/internal/peer/conn_monitor.go
index a9e89ff46ec..7ee9a7f300e 100644
--- a/client/internal/peer/conn_monitor.go
+++ b/client/internal/peer/conn_monitor.go
@@ -142,7 +142,7 @@ func (conn *Conn) handleCandidateTick(localCandidatesChanged chan<- struct{}, uf
 	if err != nil {
 		return fmt.Errorf("get local candidates: %w", err)
 	}
-	log.Debugf("Got candidates: %v", candidates)
+	log.Tracef("Got candidates: %v", candidates)
 
 	if changed := conn.updateCandidates(candidates); changed {
 		select {

From ecc0629cabd5e10c67fb3da37932a4e020434a45 Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Tue, 1 Oct 2024 15:29:05 +0200
Subject: [PATCH 10/14] Use Address only

---
 client/internal/peer/conn_monitor.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/client/internal/peer/conn_monitor.go b/client/internal/peer/conn_monitor.go
index 7ee9a7f300e..07c975ec6ff 100644
--- a/client/internal/peer/conn_monitor.go
+++ b/client/internal/peer/conn_monitor.go
@@ -164,7 +164,7 @@ func (conn *Conn) updateCandidates(newCandidates []ice.Candidate) bool {
 	}
 
 	for i, candidate := range conn.currentCandidates {
-		if candidate.String() != newCandidates[i].String() {
+		if candidate.Address() != newCandidates[i].Address() {
 			conn.currentCandidates = newCandidates
 			return true
 		}

From c9f99da71d348a76654909950c7efe263accb060 Mon Sep 17 00:00:00 2001
From: Viktor Liu <17948409+lixmal@users.noreply.github.com>
Date: Tue, 1 Oct 2024 16:24:00 +0200
Subject: [PATCH 11/14] Update client/internal/peer/conn_monitor.go

Co-authored-by: Maycon Santos <mlsmaycon@gmail.com>
---
 client/internal/peer/conn_monitor.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/client/internal/peer/conn_monitor.go b/client/internal/peer/conn_monitor.go
index 07c975ec6ff..7563a9e1b74 100644
--- a/client/internal/peer/conn_monitor.go
+++ b/client/internal/peer/conn_monitor.go
@@ -116,7 +116,7 @@ func (conn *Conn) handleCandidateTick(localCandidatesChanged chan<- struct{}, uf
 
 	gatherDone := make(chan struct{})
 	err = agent.OnCandidate(func(c ice.Candidate) {
-		log.Debugf("Got candidate: %v", c)
+		log.Tracef("Got candidate: %v", c)
 		if c == nil {
 			close(gatherDone)
 		}

From 77ec0b004431dde1ba22e51818f03bf28aa72540 Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Fri, 4 Oct 2024 12:17:39 +0200
Subject: [PATCH 12/14] Move conn monitor to own struct

---
 client/internal/peer/conn.go         |  24 +--
 client/internal/peer/conn_monitor.go | 181 ----------------------
 client/internal/peer/connmonitor.go  | 218 +++++++++++++++++++++++++++
 3 files changed, 231 insertions(+), 192 deletions(-)
 delete mode 100644 client/internal/peer/conn_monitor.go
 create mode 100644 client/internal/peer/connmonitor.go

diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go
index f15eaba2ee6..77d1cc0d5a5 100644
--- a/client/internal/peer/conn.go
+++ b/client/internal/peer/conn.go
@@ -32,10 +32,7 @@ const (
 	connPriorityICETurn ConnPriority = 1
 	connPriorityICEP2P  ConnPriority = 2
 
-	reconnectMaxElapsedTime    = 30 * time.Minute
-	candidatesMonitorPeriod    = 5 * time.Minute
-	candidatedGatheringTimeout = 5 * time.Second
-	signalerMonitorPeriod      = 5 * time.Second
+	reconnectMaxElapsedTime = 30 * time.Minute
 )
 
 type WgConfig struct {
@@ -113,10 +110,8 @@ type Conn struct {
 	// for reconnection operations
 	iCEDisconnected   chan bool
 	relayDisconnected chan bool
-	reconnectCh       chan struct{}
-
-	currentCandidates []ice.Candidate
-	candidatesMu      sync.Mutex
+	connMonitor       *ConnMonitor
+	reconnectCh       <-chan struct{}
 }
 
 // NewConn creates a new not opened Conn to the remote peer.
@@ -147,9 +142,16 @@ func NewConn(engineCtx context.Context, config ConnConfig, statusRecorder *Statu
 
 		iCEDisconnected:   make(chan bool, 1),
 		relayDisconnected: make(chan bool, 1),
-		reconnectCh:       make(chan struct{}, 1),
 	}
 
+	conn.connMonitor, conn.reconnectCh = NewConnMonitor(
+		signaler,
+		iFaceDiscover,
+		config,
+		conn.relayDisconnected,
+		conn.iCEDisconnected,
+	)
+
 	rFns := WorkerRelayCallbacks{
 		OnConnReady:    conn.relayConnectionIsReady,
 		OnDisconnected: conn.onWorkerRelayStateDisconnected,
@@ -212,6 +214,8 @@ func (conn *Conn) startHandshakeAndReconnect() {
 		conn.log.Errorf("failed to send initial offer: %v", err)
 	}
 
+	go conn.connMonitor.Start(conn.ctx)
+
 	if conn.workerRelay.IsController() {
 		conn.reconnectLoopWithRetry()
 	} else {
@@ -325,8 +329,6 @@ func (conn *Conn) reconnectLoopWithRetry() {
 	case <-time.After(3 * time.Second):
 	}
 
-	go conn.monitorReconnectEvents()
-
 	ticker := conn.prepareExponentTicker()
 	defer ticker.Stop()
 	time.Sleep(1 * time.Second)
diff --git a/client/internal/peer/conn_monitor.go b/client/internal/peer/conn_monitor.go
deleted file mode 100644
index 7563a9e1b74..00000000000
--- a/client/internal/peer/conn_monitor.go
+++ /dev/null
@@ -1,181 +0,0 @@
-package peer
-
-import (
-	"context"
-	"fmt"
-	"time"
-
-	"github.com/pion/ice/v3"
-	log "github.com/sirupsen/logrus"
-)
-
-func (conn *Conn) monitorReconnectEvents() {
-	signalerReady := make(chan struct{}, 1)
-	go conn.monitorSignalerReady(signalerReady)
-
-	localCandidatesChanged := make(chan struct{}, 1)
-	go conn.monitorLocalCandidatesChanged(localCandidatesChanged)
-
-	for {
-		select {
-		case changed := <-conn.relayDisconnected:
-			if !changed {
-				continue
-			}
-
-			conn.log.Debugf("Relay state changed, triggering reconnect")
-			conn.triggerReconnect()
-
-		case changed := <-conn.iCEDisconnected:
-			if !changed {
-				continue
-			}
-
-			conn.log.Debugf("ICE state changed, triggering reconnect")
-			conn.triggerReconnect()
-
-		case <-signalerReady:
-			conn.log.Debugf("Signaler became ready, triggering reconnect")
-			conn.triggerReconnect()
-
-		case <-localCandidatesChanged:
-			conn.log.Debugf("Local candidates changed, triggering reconnect")
-			conn.triggerReconnect()
-
-		case <-conn.ctx.Done():
-			return
-		}
-	}
-}
-
-// monitorSignalerReady monitors the signaler ready state and triggers reconnect when it transitions from not ready to ready
-func (conn *Conn) monitorSignalerReady(signalerReady chan<- struct{}) {
-	ticker := time.NewTicker(signalerMonitorPeriod)
-	defer ticker.Stop()
-
-	lastReady := true
-	for {
-		select {
-		case <-ticker.C:
-			currentReady := conn.signaler.Ready()
-			if !lastReady && currentReady {
-				select {
-				case signalerReady <- struct{}{}:
-				default:
-				}
-			}
-			lastReady = currentReady
-		case <-conn.ctx.Done():
-			return
-		}
-	}
-}
-
-// monitorLocalCandidatesChanged monitors the local candidates and triggers reconnect when they change
-func (conn *Conn) monitorLocalCandidatesChanged(localCandidatesChanged chan<- struct{}) {
-	// TODO: make this global and not per-conn
-
-	ufrag, pwd, err := generateICECredentials()
-	if err != nil {
-		conn.log.Warnf("Failed to generate ICE credentials: %v", err)
-		return
-	}
-
-	ticker := time.NewTicker(candidatesMonitorPeriod)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ticker.C:
-			if err := conn.handleCandidateTick(localCandidatesChanged, ufrag, pwd); err != nil {
-				conn.log.Warnf("Failed to handle candidate tick: %v", err)
-			}
-		case <-conn.ctx.Done():
-			return
-		}
-	}
-}
-
-func (conn *Conn) handleCandidateTick(localCandidatesChanged chan<- struct{}, ufrag string, pwd string) error {
-	conn.log.Debugf("Gathering ICE candidates")
-
-	transportNet, err := newStdNet(conn.iFaceDiscover, conn.config.ICEConfig.InterfaceBlackList)
-	if err != nil {
-		conn.log.Errorf("failed to create pion's stdnet: %s", err)
-	}
-
-	agent, err := newAgent(conn.config, transportNet, candidateTypesP2P(), ufrag, pwd)
-	if err != nil {
-		return fmt.Errorf("create ICE agent: %w", err)
-	}
-	defer func() {
-		if err := agent.Close(); err != nil {
-			conn.log.Warnf("Failed to close ICE agent: %v", err)
-		}
-	}()
-
-	gatherDone := make(chan struct{})
-	err = agent.OnCandidate(func(c ice.Candidate) {
-		log.Tracef("Got candidate: %v", c)
-		if c == nil {
-			close(gatherDone)
-		}
-	})
-	if err != nil {
-		return fmt.Errorf("set ICE candidate handler: %w", err)
-	}
-
-	if err := agent.GatherCandidates(); err != nil {
-		return fmt.Errorf("gather ICE candidates: %w", err)
-	}
-
-	ctx, cancel := context.WithTimeout(conn.ctx, candidatedGatheringTimeout)
-	defer cancel()
-
-	select {
-	case <-ctx.Done():
-		return fmt.Errorf("wait for gathering: %w", ctx.Err())
-	case <-gatherDone:
-	}
-
-	candidates, err := agent.GetLocalCandidates()
-	if err != nil {
-		return fmt.Errorf("get local candidates: %w", err)
-	}
-	log.Tracef("Got candidates: %v", candidates)
-
-	if changed := conn.updateCandidates(candidates); changed {
-		select {
-		case localCandidatesChanged <- struct{}{}:
-		default:
-		}
-	}
-
-	return nil
-}
-
-func (conn *Conn) updateCandidates(newCandidates []ice.Candidate) bool {
-	conn.candidatesMu.Lock()
-	defer conn.candidatesMu.Unlock()
-
-	if len(conn.currentCandidates) != len(newCandidates) {
-		conn.currentCandidates = newCandidates
-		return true
-	}
-
-	for i, candidate := range conn.currentCandidates {
-		if candidate.Address() != newCandidates[i].Address() {
-			conn.currentCandidates = newCandidates
-			return true
-		}
-	}
-
-	return false
-}
-
-func (conn *Conn) triggerReconnect() {
-	select {
-	case conn.reconnectCh <- struct{}{}:
-	default:
-	}
-}
diff --git a/client/internal/peer/connmonitor.go b/client/internal/peer/connmonitor.go
new file mode 100644
index 00000000000..04ec1a32f59
--- /dev/null
+++ b/client/internal/peer/connmonitor.go
@@ -0,0 +1,218 @@
+package peer
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/pion/ice/v3"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/client/internal/stdnet"
+)
+
+const (
+	signalerMonitorPeriod     = 5 * time.Second
+	candidatesMonitorPeriod   = 5 * time.Minute
+	candidateGatheringTimeout = 5 * time.Second
+)
+
+type ConnMonitor struct {
+	signaler          *Signaler
+	iFaceDiscover     stdnet.ExternalIFaceDiscover
+	config            ConnConfig
+	relayDisconnected chan bool
+	iCEDisconnected   chan bool
+	reconnectCh       chan struct{}
+	currentCandidates []ice.Candidate
+	candidatesMu      sync.Mutex
+}
+
+func NewConnMonitor(signaler *Signaler, iFaceDiscover stdnet.ExternalIFaceDiscover, config ConnConfig, relayDisconnected, iCEDisconnected chan bool) (*ConnMonitor, <-chan struct{}) {
+	reconnectCh := make(chan struct{}, 1)
+	cm := &ConnMonitor{
+		signaler:          signaler,
+		iFaceDiscover:     iFaceDiscover,
+		config:            config,
+		relayDisconnected: relayDisconnected,
+		iCEDisconnected:   iCEDisconnected,
+		reconnectCh:       reconnectCh,
+	}
+	return cm, reconnectCh
+}
+
+func (cm *ConnMonitor) Start(ctx context.Context) {
+	signalerReady := make(chan struct{}, 1)
+	go cm.monitorSignalerReady(ctx, signalerReady)
+
+	localCandidatesChanged := make(chan struct{}, 1)
+	go cm.monitorLocalCandidatesChanged(ctx, localCandidatesChanged)
+
+	for {
+		select {
+		case changed := <-cm.relayDisconnected:
+			if !changed {
+				continue
+			}
+			log.Debugf("Relay state changed, triggering reconnect")
+			cm.triggerReconnect()
+
+		case changed := <-cm.iCEDisconnected:
+			if !changed {
+				continue
+			}
+			log.Debugf("ICE state changed, triggering reconnect")
+			cm.triggerReconnect()
+
+		case <-signalerReady:
+			log.Debugf("Signaler became ready, triggering reconnect")
+			cm.triggerReconnect()
+
+		case <-localCandidatesChanged:
+			log.Debugf("Local candidates changed, triggering reconnect")
+			cm.triggerReconnect()
+
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (cm *ConnMonitor) monitorSignalerReady(ctx context.Context, signalerReady chan<- struct{}) {
+	if cm.signaler == nil {
+		return
+	}
+
+	ticker := time.NewTicker(signalerMonitorPeriod)
+	defer ticker.Stop()
+
+	lastReady := true
+	for {
+		select {
+		case <-ticker.C:
+			currentReady := cm.signaler.Ready()
+			if !lastReady && currentReady {
+				select {
+				case signalerReady <- struct{}{}:
+				default:
+				}
+			}
+			lastReady = currentReady
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (cm *ConnMonitor) monitorLocalCandidatesChanged(ctx context.Context, localCandidatesChanged chan<- struct{}) {
+	ufrag, pwd, err := generateICECredentials()
+	if err != nil {
+		log.Warnf("Failed to generate ICE credentials: %v", err)
+		return
+	}
+
+	ticker := time.NewTicker(candidatesMonitorPeriod)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			if err := cm.handleCandidateTick(ctx, localCandidatesChanged, ufrag, pwd); err != nil {
+				log.Warnf("Failed to handle candidate tick: %v", err)
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (cm *ConnMonitor) handleCandidateTick(ctx context.Context, localCandidatesChanged chan<- struct{}, ufrag string, pwd string) error {
+	log.Debugf("Gathering ICE candidates")
+
+	transportNet, err := newStdNet(cm.iFaceDiscover, cm.config.ICEConfig.InterfaceBlackList)
+	if err != nil {
+		log.Errorf("failed to create pion's stdnet: %s", err)
+	}
+
+	agent, err := newAgent(cm.config, transportNet, candidateTypesP2P(), ufrag, pwd)
+	if err != nil {
+		return fmt.Errorf("create ICE agent: %w", err)
+	}
+	defer func() {
+		if err := agent.Close(); err != nil {
+			log.Warnf("Failed to close ICE agent: %v", err)
+		}
+	}()
+
+	gatherDone := make(chan struct{})
+	err = agent.OnCandidate(func(c ice.Candidate) {
+		log.Tracef("Got candidate: %v", c)
+		if c == nil {
+			close(gatherDone)
+		}
+	})
+	if err != nil {
+		return fmt.Errorf("set ICE candidate handler: %w", err)
+	}
+
+	if err := agent.GatherCandidates(); err != nil {
+		return fmt.Errorf("gather ICE candidates: %w", err)
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, candidateGatheringTimeout)
+	defer cancel()
+
+	select {
+	case <-ctx.Done():
+		return fmt.Errorf("wait for gathering: %w", ctx.Err())
+	case <-gatherDone:
+	}
+
+	candidates, err := agent.GetLocalCandidates()
+	if err != nil {
+		return fmt.Errorf("get local candidates: %w", err)
+	}
+	log.Tracef("Got candidates: %v", candidates)
+
+	if changed := cm.updateCandidates(candidates); changed {
+		select {
+		case localCandidatesChanged <- struct{}{}:
+		default:
+		}
+	}
+
+	return nil
+}
+
+func (cm *ConnMonitor) updateCandidates(newCandidates []ice.Candidate) bool {
+	cm.candidatesMu.Lock()
+	defer cm.candidatesMu.Unlock()
+
+	if len(cm.currentCandidates) != len(newCandidates) {
+		cm.currentCandidates = newCandidates
+		return true
+	}
+
+	for i, candidate := range cm.currentCandidates {
+		if candidate.Address() != newCandidates[i].Address() {
+			cm.currentCandidates = newCandidates
+			return true
+		}
+	}
+
+	return false
+}
+
+func (cm *ConnMonitor) triggerReconnect() {
+	select {
+	case cm.reconnectCh <- struct{}{}:
+	default:
+	}
+}
+
+func (cm *ConnMonitor) GetCurrentCandidates() []ice.Candidate {
+	cm.candidatesMu.Lock()
+	defer cm.candidatesMu.Unlock()
+	return append([]ice.Candidate{}, cm.currentCandidates...)
+}

From 0682483b1bd3a066b1119dac2b03eb31d54b2ff3 Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Fri, 4 Oct 2024 12:32:01 +0200
Subject: [PATCH 13/14] Remove unused method

---
 client/internal/peer/connmonitor.go | 6 ------
 1 file changed, 6 deletions(-)

diff --git a/client/internal/peer/connmonitor.go b/client/internal/peer/connmonitor.go
index 04ec1a32f59..75722c99011 100644
--- a/client/internal/peer/connmonitor.go
+++ b/client/internal/peer/connmonitor.go
@@ -210,9 +210,3 @@ func (cm *ConnMonitor) triggerReconnect() {
 	default:
 	}
 }
-
-func (cm *ConnMonitor) GetCurrentCandidates() []ice.Candidate {
-	cm.candidatesMu.Lock()
-	defer cm.candidatesMu.Unlock()
-	return append([]ice.Candidate{}, cm.currentCandidates...)
-}

From c11d80705f9c819da70168a806f5c2308ba01d50 Mon Sep 17 00:00:00 2001
From: Viktor Liu <viktor@netbird.io>
Date: Fri, 4 Oct 2024 12:32:49 +0200
Subject: [PATCH 14/14] Rename file

---
 client/internal/peer/{connmonitor.go => conn_monitor.go} | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename client/internal/peer/{connmonitor.go => conn_monitor.go} (100%)

diff --git a/client/internal/peer/connmonitor.go b/client/internal/peer/conn_monitor.go
similarity index 100%
rename from client/internal/peer/connmonitor.go
rename to client/internal/peer/conn_monitor.go