mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-04 04:29:06 +02:00
[client] Sweep connections on network loss via a shared netevents manager
Losing the last network only flipped the availability state: the dead management, signal and relay sockets stayed silently connected until their own timeouts, so the client kept reporting Connected with no network at all. Introduce client/netevents with a Manager that ties the availability state, the connection sweeper and the status recorder together, and move the netstate and netsweep packages under it (netsweep renamed to sweep). SetNetworkAvailable(false) now also sweeps the registered connections so their owners redial and the listener reaches the NoNetwork state. The Android and iOS bindings own a Manager instance and inject it through the constructors; consumers hold the concrete *Manager whose nil zero value reports always-online and never sweeps, with interfaces kept only as parameter contracts. The relay guard settle wait moved into the Manager as WaitSettled, removing the netevents import from the relay package.
This commit is contained in:
+14
-26
@@ -38,8 +38,7 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/updater"
|
||||
"github.com/netbirdio/netbird/client/internal/updater/installer"
|
||||
nbnet "github.com/netbirdio/netbird/client/net"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netsweep"
|
||||
"github.com/netbirdio/netbird/client/netevents"
|
||||
cProto "github.com/netbirdio/netbird/client/proto"
|
||||
"github.com/netbirdio/netbird/client/ssh"
|
||||
sshconfig "github.com/netbirdio/netbird/client/ssh/config"
|
||||
@@ -73,28 +72,17 @@ type ConnectClient struct {
|
||||
|
||||
persistSyncResponse bool
|
||||
|
||||
// netState gates every reconnection loop on OS-reported network
|
||||
// availability. Nil (the default) disables gating; mobile platforms
|
||||
// inject it via WithNetworkState.
|
||||
netState *netstate.State
|
||||
|
||||
// sweeper cuts the management, signal and relay connections on network
|
||||
// change; nil disables it.
|
||||
sweeper *netsweep.Sweeper
|
||||
// netEvents gates every reconnection loop on OS-reported network
|
||||
// availability and sweeps connections on network change.
|
||||
netEvents *netevents.Manager
|
||||
}
|
||||
|
||||
// ConnectClientOption configures optional ConnectClient behavior.
|
||||
type ConnectClientOption func(*ConnectClient)
|
||||
|
||||
// WithNetworkState injects the OS network availability state that gates every
|
||||
// reconnection loop; without it gating is disabled.
|
||||
func WithNetworkState(netState *netstate.State) ConnectClientOption {
|
||||
return func(c *ConnectClient) { c.netState = netState }
|
||||
}
|
||||
|
||||
// WithSweeper injects the network change sweeper.
|
||||
func WithSweeper(sweeper *netsweep.Sweeper) ConnectClientOption {
|
||||
return func(c *ConnectClient) { c.sweeper = sweeper }
|
||||
// WithNetEvents injects the OS network event handling.
|
||||
func WithNetEvents(events *netevents.Manager) ConnectClientOption {
|
||||
return func(c *ConnectClient) { c.netEvents = events }
|
||||
}
|
||||
|
||||
func NewConnectClient(
|
||||
@@ -305,7 +293,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
}
|
||||
|
||||
// suspend connection attempts while the OS reports no usable network
|
||||
if waited, err := c.netState.Wait(c.ctx); err != nil {
|
||||
if waited, err := c.netEvents.Wait(c.ctx); err != nil {
|
||||
return nil
|
||||
} else if waited {
|
||||
backOff.Reset()
|
||||
@@ -323,7 +311,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
|
||||
log.Debugf("connecting to the Management service %s", c.config.ManagementURL.Host)
|
||||
mgmClient, err := mgm.NewClient(engineCtx, c.config.ManagementURL.Host, myPrivateKey, mgmTlsEnabled,
|
||||
mgm.WithNetworkState(c.netState), mgm.WithSweeper(c.sweeper))
|
||||
mgm.WithNetEvents(c.netEvents))
|
||||
if err != nil {
|
||||
// On daemon shutdown / Down() the parent context is cancelled
|
||||
// and the dial fails with "context canceled". Wrapping that
|
||||
@@ -398,7 +386,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
}()
|
||||
|
||||
// with the global Netbird config in hand connect (just a connection, no stream yet) Signal
|
||||
signalClient, err := connectToSignal(engineCtx, loginResp.GetNetbirdConfig(), myPrivateKey, c.netState, c.sweeper)
|
||||
signalClient, err := connectToSignal(engineCtx, loginResp.GetNetbirdConfig(), myPrivateKey, c.netEvents)
|
||||
if err != nil {
|
||||
log.Error(err)
|
||||
return wrapErr(err)
|
||||
@@ -435,7 +423,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
}
|
||||
|
||||
relayManager := relayClient.NewManager(engineCtx, relayURLs, myPrivateKey.PublicKey().String(), engineConfig.MTU,
|
||||
relayClient.WithNetworkState(c.netState), relayClient.WithSweeper(c.sweeper))
|
||||
relayClient.WithNetEvents(c.netEvents))
|
||||
c.statusRecorder.SetRelayMgr(relayManager)
|
||||
if len(relayURLs) > 0 {
|
||||
if token != nil {
|
||||
@@ -463,7 +451,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
|
||||
UpdateManager: c.updateManager,
|
||||
ClientMetrics: c.clientMetrics,
|
||||
MetricsCtx: c.ctx,
|
||||
NetState: c.netState,
|
||||
NetState: c.netEvents,
|
||||
}, mobileDependency)
|
||||
engine.SetSyncResponsePersistence(c.persistSyncResponse)
|
||||
c.engine = engine
|
||||
@@ -723,7 +711,7 @@ func selectMTU(localMTU uint16, peerMTU int32) uint16 {
|
||||
}
|
||||
|
||||
// connectToSignal creates Signal Service client and established a connection
|
||||
func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourPrivateKey wgtypes.Key, netState *netstate.State, sweeper *netsweep.Sweeper) (*signal.GrpcClient, error) {
|
||||
func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourPrivateKey wgtypes.Key, netEvents *netevents.Manager) (*signal.GrpcClient, error) {
|
||||
var sigTLSEnabled bool
|
||||
if wtConfig.Signal.Protocol == mgmProto.HostConfig_HTTPS {
|
||||
sigTLSEnabled = true
|
||||
@@ -732,7 +720,7 @@ func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourP
|
||||
}
|
||||
|
||||
signalClient, err := signal.NewClient(ctx, wtConfig.Signal.Uri, ourPrivateKey, sigTLSEnabled,
|
||||
signal.WithNetworkState(netState), signal.WithSweeper(sweeper))
|
||||
signal.WithNetEvents(netEvents))
|
||||
if err != nil {
|
||||
log.Errorf("error while connecting to the Signal Exchange Service %s: %s", wtConfig.Signal.Uri, err)
|
||||
return nil, gstatus.Errorf(codes.FailedPrecondition, "failed connecting to Signal Service : %s", err)
|
||||
|
||||
@@ -59,7 +59,7 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/syncstore"
|
||||
"github.com/netbirdio/netbird/client/internal/updater"
|
||||
"github.com/netbirdio/netbird/client/jobexec"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netevents"
|
||||
cProto "github.com/netbirdio/netbird/client/proto"
|
||||
"github.com/netbirdio/netbird/client/system"
|
||||
nbdns "github.com/netbirdio/netbird/dns"
|
||||
@@ -184,7 +184,7 @@ type EngineServices struct {
|
||||
MetricsCtx context.Context
|
||||
// NetState gates the reconnection loops on OS-reported network
|
||||
// availability; nil disables gating.
|
||||
NetState *netstate.State
|
||||
NetState *netevents.Manager
|
||||
}
|
||||
|
||||
// Engine is a mechanism responsible for reacting on Signal and Management stream events and managing connections to the remote peers.
|
||||
@@ -210,7 +210,7 @@ type Engine struct {
|
||||
|
||||
// netState gates the peer reconnection guards on OS-reported network
|
||||
// availability; nil disables gating.
|
||||
netState *netstate.State
|
||||
netState *netevents.Manager
|
||||
|
||||
// STUNs is a list of STUN servers used by ICE
|
||||
STUNs []*stun.URI
|
||||
|
||||
@@ -26,7 +26,7 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/portforward"
|
||||
"github.com/netbirdio/netbird/client/internal/rosenpass"
|
||||
"github.com/netbirdio/netbird/client/internal/stdnet"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netevents"
|
||||
"github.com/netbirdio/netbird/route"
|
||||
relayClient "github.com/netbirdio/netbird/shared/relay/client"
|
||||
)
|
||||
@@ -97,7 +97,7 @@ type ConnConfig struct {
|
||||
|
||||
// NetworkState gates the reconnection guard on OS-reported network
|
||||
// availability; nil disables gating.
|
||||
NetworkState *netstate.State
|
||||
NetworkState *netevents.Manager
|
||||
}
|
||||
|
||||
type Conn struct {
|
||||
|
||||
@@ -6,8 +6,6 @@ import (
|
||||
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
)
|
||||
|
||||
// ConnStatus represents the connection state as seen by the guard.
|
||||
@@ -24,6 +22,12 @@ const (
|
||||
|
||||
type connStatusFunc func() ConnStatus
|
||||
|
||||
// NetworkWatcher is the availability view the guard gates reconnects on.
|
||||
type NetworkWatcher interface {
|
||||
IsOnline() bool
|
||||
Changed() <-chan struct{}
|
||||
}
|
||||
|
||||
// Guard is responsible for the reconnection logic.
|
||||
// It will trigger to send an offer to the peer then has connection issues.
|
||||
// Watch these events:
|
||||
@@ -39,14 +43,14 @@ type Guard struct {
|
||||
srWatcher *SRWatcher
|
||||
// netState gates reconnect attempts on OS-reported network availability;
|
||||
// nil disables gating.
|
||||
netState *netstate.State
|
||||
netState NetworkWatcher
|
||||
relayedConnDisconnected chan struct{}
|
||||
iCEConnDisconnected chan struct{}
|
||||
}
|
||||
|
||||
// NewGuard creates a reconnection guard for a peer connection. A nil netState
|
||||
// disables network availability gating.
|
||||
func NewGuard(log *log.Entry, isConnectedFn connStatusFunc, timeout time.Duration, srWatcher *SRWatcher, netState *netstate.State) *Guard {
|
||||
func NewGuard(log *log.Entry, isConnectedFn connStatusFunc, timeout time.Duration, srWatcher *SRWatcher, netState NetworkWatcher) *Guard {
|
||||
return &Guard{
|
||||
log: log,
|
||||
isConnectedOnAllWay: isConnectedFn,
|
||||
@@ -104,14 +108,17 @@ func (g *Guard) reconnectLoopWithRetry(ctx context.Context, callback func()) {
|
||||
iceState := &iceRetryState{log: g.log}
|
||||
defer iceState.reset()
|
||||
|
||||
netChanged := g.netState.Changed()
|
||||
var netChanged <-chan struct{}
|
||||
if g.netState != nil {
|
||||
netChanged = g.netState.Changed()
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-tickerChannel:
|
||||
// skip attempts while the OS reports no usable network; the
|
||||
// netChanged case below resumes the loop once it returns
|
||||
if !g.netState.IsOnline() {
|
||||
if g.netState != nil && !g.netState.IsOnline() {
|
||||
continue
|
||||
}
|
||||
switch g.isConnectedOnAllWay() {
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/internal/peer/ice"
|
||||
"github.com/netbirdio/netbird/client/netstate"
|
||||
"github.com/netbirdio/netbird/client/netevents/netstate"
|
||||
)
|
||||
|
||||
// newTestGuardWithNetState builds a guard with a realistic MaxInterval: the
|
||||
|
||||
Reference in New Issue
Block a user