[client] Suspend reconnection loops while the OS reports no network

On mobile the client kept dialing management, signal, relay and peer
connections while the device had no usable network at all (airplane
mode), burning battery for attempts that cannot succeed. Stopping the
engine is not an option: tearing it down destroys the TUN device, and
traffic can leak outside the tunnel until it is rebuilt.

Add client/netstate, a small gate the platform feeds from its own
connectivity callbacks. Every reconnection loop waits on it instead of
retrying blindly, and resets its backoff when the network returns so
recovery is immediate. The state is injected through functional options
and consumers hold a *State that may be nil, so every platform that does
not report availability behaves exactly as before.

The relay quick-reconnect rechecks availability after its 1.5s wait: the
disconnect that triggers it is usually the first symptom of the network
going away, so the flag typically arrives while it sleeps.

Report the suspension to the UI as well. peer.Listener grows
OnStateChanged with a typed ClientState, re-exported across the gomobile
boundary as integer constants, and the notifier maps Connecting to a new
NoNetwork state while the OS reports no network, so mobile clients can
show "no network available" instead of a misleading "connecting".

Finally, exit the client retry loop cleanly when its context is
cancelled. backoff.WithContext surfaces the bare context error, which
callers could not distinguish from a real failure — on Android that
turned an engine restart into an unrecoverable error.
This commit is contained in:
Zoltan Papp
2026-08-11 00:58:50 +02:00
parent 2ce6323602
commit 902263ac96
19 changed files with 673 additions and 81 deletions
+18 -8
View File
@@ -25,6 +25,7 @@ import (
"github.com/netbirdio/netbird/client/internal/routemanager" "github.com/netbirdio/netbird/client/internal/routemanager"
"github.com/netbirdio/netbird/client/internal/stdnet" "github.com/netbirdio/netbird/client/internal/stdnet"
"github.com/netbirdio/netbird/client/net" "github.com/netbirdio/netbird/client/net"
"github.com/netbirdio/netbird/client/netstate"
"github.com/netbirdio/netbird/client/system" "github.com/netbirdio/netbird/client/system"
"github.com/netbirdio/netbird/formatter" "github.com/netbirdio/netbird/formatter"
"github.com/netbirdio/netbird/route" "github.com/netbirdio/netbird/route"
@@ -32,11 +33,6 @@ import (
types "github.com/netbirdio/netbird/upload-server/types" types "github.com/netbirdio/netbird/upload-server/types"
) )
// ConnectionListener export internal Listener for mobile
type ConnectionListener interface {
peer.Listener
}
// TunAdapter export internal TunAdapter for mobile // TunAdapter export internal TunAdapter for mobile
type TunAdapter interface { type TunAdapter interface {
device.TunAdapter device.TunAdapter
@@ -77,6 +73,10 @@ type Client struct {
deviceName string deviceName string
uiVersion string uiVersion string
networkChangeListener listener.NetworkChangeListener networkChangeListener listener.NetworkChangeListener
// netState outlives engine restarts: it mirrors the OS connectivity, not
// the engine lifecycle. Run and RunWithoutLogin inject it into each new
// ConnectClient, which distributes it to every reconnection loop.
netState *netstate.State
stateMu sync.RWMutex stateMu sync.RWMutex
connectClient *internal.ConnectClient connectClient *internal.ConnectClient
@@ -148,9 +148,19 @@ func NewClient(androidSDKVersion int, deviceName string, uiVersion string, tunAd
recorder: peer.NewRecorder(""), recorder: peer.NewRecorder(""),
ctxCancelLock: &sync.Mutex{}, ctxCancelLock: &sync.Mutex{},
networkChangeListener: networkChangeListener, networkChangeListener: networkChangeListener,
netState: netstate.New(),
} }
} }
// SetNetworkAvailable feeds OS-reported network availability into the client.
// While unavailable, the internal reconnect loops suspend their attempts and
// the connection listener reports NoNetwork instead of Connecting; when
// availability returns, the loops resume immediately with a fresh backoff.
func (c *Client) SetNetworkAvailable(available bool) {
c.netState.Set(available)
c.recorder.SetNetworkAvailable(available)
}
// Run start the internal client. It is a blocker function // Run start the internal client. It is a blocker function
func (c *Client) Run(platformFiles PlatformFiles, urlOpener URLOpener, isAndroidTV bool, dns *DNSList, dnsReadyListener DnsReadyListener, envList *EnvList) error { func (c *Client) Run(platformFiles PlatformFiles, urlOpener URLOpener, isAndroidTV bool, dns *DNSList, dnsReadyListener DnsReadyListener, envList *EnvList) error {
exportEnvList(envList) exportEnvList(envList)
@@ -188,7 +198,7 @@ func (c *Client) Run(platformFiles PlatformFiles, urlOpener URLOpener, isAndroid
} }
// todo do not throw error in case of cancelled context // todo do not throw error in case of cancelled context
ctx = internal.CtxInitState(ctx) ctx = internal.CtxInitState(ctx)
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder) connectClient := internal.NewConnectClient(ctx, cfg, c.recorder, internal.WithNetworkState(c.netState))
c.setState(cfg, cacheDir, cfgFile, connectClient) c.setState(cfg, cacheDir, cfgFile, connectClient)
// This path runs the interactive SSO flow, so reaching here means the peer // This path runs the interactive SSO flow, so reaching here means the peer
// is authenticated again — release the latch Status() reports from. Clear // is authenticated again — release the latch Status() reports from. Clear
@@ -229,7 +239,7 @@ func (c *Client) RunWithoutLogin(platformFiles PlatformFiles, dns *DNSList, dnsR
// todo do not throw error in case of cancelled context // todo do not throw error in case of cancelled context
ctx = internal.CtxInitState(ctx) ctx = internal.CtxInitState(ctx)
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder) connectClient := internal.NewConnectClient(ctx, cfg, c.recorder, internal.WithNetworkState(c.netState))
c.setState(cfg, cacheDir, cfgFile, connectClient) c.setState(cfg, cacheDir, cfgFile, connectClient)
return connectClient.RunOnAndroid(c.tunAdapter, c.iFaceDiscover, c.networkChangeListener, slices.Clone(dns.items), dnsReadyListener, stateFile, cacheDir) return connectClient.RunOnAndroid(c.tunAdapter, c.iFaceDiscover, c.networkChangeListener, slices.Clone(dns.items), dnsReadyListener, stateFile, cacheDir)
} }
@@ -513,7 +523,7 @@ func (c *Client) OnUpdatedHostDNS(list *DNSList) error {
// SetConnectionListener set the network connection listener // SetConnectionListener set the network connection listener
func (c *Client) SetConnectionListener(listener ConnectionListener) { func (c *Client) SetConnectionListener(listener ConnectionListener) {
c.recorder.SetConnectionListener(listener) c.recorder.SetConnectionListener(connectionListenerAdapter{listener})
} }
// RemoveConnectionListener remove connection listener // RemoveConnectionListener remove connection listener
+41
View File
@@ -0,0 +1,41 @@
//go:build android
package android
import (
"github.com/netbirdio/netbird/client/internal/peer"
)
// Client state values delivered via ConnectionListener.OnStateChanged,
// re-exported as basic constants so gomobile emits them into the generated
// Java bindings. They mirror peer.ClientState*: append-only, never reorder.
const (
ClientStateDisconnected = int(peer.ClientStateDisconnected)
ClientStateConnected = int(peer.ClientStateConnected)
ClientStateConnecting = int(peer.ClientStateConnecting)
ClientStateDisconnecting = int(peer.ClientStateDisconnecting)
ClientStateNoNetwork = int(peer.ClientStateNoNetwork)
)
// ConnectionListener export internal Listener for mobile. It mirrors
// peer.Listener with OnStateChanged taking a plain int (one of the
// ClientState* constants), because gomobile cannot bind named types.
type ConnectionListener interface {
OnStateChanged(state int)
OnConnected()
OnDisconnected()
OnConnecting()
OnDisconnecting()
OnAddressChanged(string, string)
OnPeersListChanged(int)
}
// connectionListenerAdapter adapts the gomobile-facing ConnectionListener to
// peer.Listener, converting the typed state to the int the binding carries.
type connectionListenerAdapter struct {
ConnectionListener
}
func (a connectionListenerAdapter) OnStateChanged(state peer.ClientState) {
a.ConnectionListener.OnStateChanged(int(state))
}
+45 -6
View File
@@ -38,6 +38,7 @@ import (
"github.com/netbirdio/netbird/client/internal/updater" "github.com/netbirdio/netbird/client/internal/updater"
"github.com/netbirdio/netbird/client/internal/updater/installer" "github.com/netbirdio/netbird/client/internal/updater/installer"
nbnet "github.com/netbirdio/netbird/client/net" nbnet "github.com/netbirdio/netbird/client/net"
"github.com/netbirdio/netbird/client/netstate"
cProto "github.com/netbirdio/netbird/client/proto" cProto "github.com/netbirdio/netbird/client/proto"
"github.com/netbirdio/netbird/client/ssh" "github.com/netbirdio/netbird/client/ssh"
sshconfig "github.com/netbirdio/netbird/client/ssh/config" sshconfig "github.com/netbirdio/netbird/client/ssh/config"
@@ -70,18 +71,33 @@ type ConnectClient struct {
updateManager *updater.Manager updateManager *updater.Manager
persistSyncResponse bool 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
}
// 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 }
} }
func NewConnectClient( func NewConnectClient(
ctx context.Context, ctx context.Context,
config *profilemanager.Config, config *profilemanager.Config,
statusRecorder *peer.Status, statusRecorder *peer.Status,
opts ...ConnectClientOption,
) *ConnectClient { ) *ConnectClient {
// Derive the run context here so Stop owns the cancel that unblocks the run // Derive the run context here so Stop owns the cancel that unblocks the run
// loop. runCancel is set once at construction, so Stop can call it without // loop. runCancel is set once at construction, so Stop can call it without
// racing the run loop's startup. Callers therefore need not cancel before Stop. // racing the run loop's startup. Callers therefore need not cancel before Stop.
runCtx, runCancel := context.WithCancel(ctx) runCtx, runCancel := context.WithCancel(ctx)
return &ConnectClient{ c := &ConnectClient{
ctx: runCtx, ctx: runCtx,
runCancel: runCancel, runCancel: runCancel,
runExited: make(chan struct{}), runExited: make(chan struct{}),
@@ -89,6 +105,10 @@ func NewConnectClient(
statusRecorder: statusRecorder, statusRecorder: statusRecorder,
engineMutex: sync.Mutex{}, engineMutex: sync.Mutex{},
} }
for _, opt := range opts {
opt(c)
}
return c
} }
func (c *ConnectClient) SetUpdateManager(um *updater.Manager) { func (c *ConnectClient) SetUpdateManager(um *updater.Manager) {
@@ -274,6 +294,13 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
return nil return nil
} }
// suspend connection attempts while the OS reports no usable network
if waited, err := c.netState.Wait(c.ctx); err != nil {
return nil
} else if waited {
backOff.Reset()
}
state.Set(StatusConnecting) state.Set(StatusConnecting)
engineCtx, cancel := context.WithCancel(c.ctx) engineCtx, cancel := context.WithCancel(c.ctx)
@@ -285,7 +312,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
}() }()
log.Debugf("connecting to the Management service %s", c.config.ManagementURL.Host) log.Debugf("connecting to the Management service %s", c.config.ManagementURL.Host)
mgmClient, err := mgm.NewClient(engineCtx, c.config.ManagementURL.Host, myPrivateKey, mgmTlsEnabled) mgmClient, err := mgm.NewClient(engineCtx, c.config.ManagementURL.Host, myPrivateKey, mgmTlsEnabled, mgm.WithNetworkState(c.netState))
if err != nil { if err != nil {
// On daemon shutdown / Down() the parent context is cancelled // On daemon shutdown / Down() the parent context is cancelled
// and the dial fails with "context canceled". Wrapping that // and the dial fails with "context canceled". Wrapping that
@@ -360,7 +387,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
}() }()
// with the global Netbird config in hand connect (just a connection, no stream yet) Signal // with the global Netbird config in hand connect (just a connection, no stream yet) Signal
signalClient, err := connectToSignal(engineCtx, loginResp.GetNetbirdConfig(), myPrivateKey) signalClient, err := connectToSignal(engineCtx, loginResp.GetNetbirdConfig(), myPrivateKey, c.netState)
if err != nil { if err != nil {
log.Error(err) log.Error(err)
return wrapErr(err) return wrapErr(err)
@@ -396,7 +423,8 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
engineConfig.StateDir = filepath.Dir(path) engineConfig.StateDir = filepath.Dir(path)
} }
relayManager := relayClient.NewManager(engineCtx, relayURLs, myPrivateKey.PublicKey().String(), engineConfig.MTU) relayManager := relayClient.NewManager(engineCtx, relayURLs, myPrivateKey.PublicKey().String(), engineConfig.MTU,
relayClient.WithNetworkState(c.netState))
c.statusRecorder.SetRelayMgr(relayManager) c.statusRecorder.SetRelayMgr(relayManager)
if len(relayURLs) > 0 { if len(relayURLs) > 0 {
if token != nil { if token != nil {
@@ -424,6 +452,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
UpdateManager: c.updateManager, UpdateManager: c.updateManager,
ClientMetrics: c.clientMetrics, ClientMetrics: c.clientMetrics,
MetricsCtx: c.ctx, MetricsCtx: c.ctx,
NetState: c.netState,
}, mobileDependency) }, mobileDependency)
engine.SetSyncResponsePersistence(c.persistSyncResponse) engine.SetSyncResponsePersistence(c.persistSyncResponse)
c.engine = engine c.engine = engine
@@ -480,6 +509,16 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
// status stream stuck at Connecting. // status stream stuck at Connecting.
err = backoff.Retry(operation, backoff.WithContext(backOff, c.ctx)) err = backoff.Retry(operation, backoff.WithContext(backOff, c.ctx))
if err != nil { if err != nil {
// Once the client context is cancelled backoff.WithContext surfaces the
// bare context error, and any attempt torn down mid-flight reports the
// same. That cancellation is the caller asking us to stop (Stop, Down or
// an engine restart), so exit cleanly instead of handing back a failure
// the caller would have to distinguish from a real one.
if c.ctx.Err() != nil && errors.Is(err, context.Canceled) {
log.Info("exiting client retry loop, context cancelled")
return nil
}
log.Debugf("exiting client retry loop due to unrecoverable error: %s", err) log.Debugf("exiting client retry loop due to unrecoverable error: %s", err)
if s, ok := gstatus.FromError(err); ok && (s.Code() == codes.PermissionDenied) { if s, ok := gstatus.FromError(err); ok && (s.Code() == codes.PermissionDenied) {
state.Set(StatusNeedsLogin) state.Set(StatusNeedsLogin)
@@ -673,7 +712,7 @@ func selectMTU(localMTU uint16, peerMTU int32) uint16 {
} }
// connectToSignal creates Signal Service client and established a connection // connectToSignal creates Signal Service client and established a connection
func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourPrivateKey wgtypes.Key) (*signal.GrpcClient, error) { func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourPrivateKey wgtypes.Key, netState *netstate.State) (*signal.GrpcClient, error) {
var sigTLSEnabled bool var sigTLSEnabled bool
if wtConfig.Signal.Protocol == mgmProto.HostConfig_HTTPS { if wtConfig.Signal.Protocol == mgmProto.HostConfig_HTTPS {
sigTLSEnabled = true sigTLSEnabled = true
@@ -681,7 +720,7 @@ func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourP
sigTLSEnabled = false sigTLSEnabled = false
} }
signalClient, err := signal.NewClient(ctx, wtConfig.Signal.Uri, ourPrivateKey, sigTLSEnabled) signalClient, err := signal.NewClient(ctx, wtConfig.Signal.Uri, ourPrivateKey, sigTLSEnabled, signal.WithNetworkState(netState))
if err != nil { if err != nil {
log.Errorf("error while connecting to the Signal Exchange Service %s: %s", wtConfig.Signal.Uri, err) 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) return nil, gstatus.Errorf(codes.FailedPrecondition, "failed connecting to Signal Service : %s", err)
+11 -1
View File
@@ -58,6 +58,7 @@ import (
"github.com/netbirdio/netbird/client/internal/syncstore" "github.com/netbirdio/netbird/client/internal/syncstore"
"github.com/netbirdio/netbird/client/internal/updater" "github.com/netbirdio/netbird/client/internal/updater"
"github.com/netbirdio/netbird/client/jobexec" "github.com/netbirdio/netbird/client/jobexec"
"github.com/netbirdio/netbird/client/netstate"
cProto "github.com/netbirdio/netbird/client/proto" cProto "github.com/netbirdio/netbird/client/proto"
"github.com/netbirdio/netbird/client/system" "github.com/netbirdio/netbird/client/system"
nbdns "github.com/netbirdio/netbird/dns" nbdns "github.com/netbirdio/netbird/dns"
@@ -180,6 +181,9 @@ type EngineServices struct {
UpdateManager *updater.Manager UpdateManager *updater.Manager
ClientMetrics *metrics.ClientMetrics ClientMetrics *metrics.ClientMetrics
MetricsCtx context.Context MetricsCtx context.Context
// NetState gates the reconnection loops on OS-reported network
// availability; nil disables gating.
NetState *netstate.State
} }
// Engine is a mechanism responsible for reacting on Signal and Management stream events and managing connections to the remote peers. // Engine is a mechanism responsible for reacting on Signal and Management stream events and managing connections to the remote peers.
@@ -203,6 +207,10 @@ type Engine struct {
config *EngineConfig config *EngineConfig
mobileDep MobileDependency mobileDep MobileDependency
// netState gates the peer reconnection guards on OS-reported network
// availability; nil disables gating.
netState *netstate.State
// STUNs is a list of STUN servers used by ICE // STUNs is a list of STUN servers used by ICE
STUNs []*stun.URI STUNs []*stun.URI
// TURNs is a list of STUN servers used by ICE // TURNs is a list of STUN servers used by ICE
@@ -336,6 +344,7 @@ func NewEngine(
syncMsgMux: &sync.Mutex{}, syncMsgMux: &sync.Mutex{},
config: config, config: config,
mobileDep: mobileDep, mobileDep: mobileDep,
netState: services.NetState,
STUNs: []*stun.URI{}, STUNs: []*stun.URI{},
TURNs: []*stun.URI{}, TURNs: []*stun.URI{},
networkSerial: 0, networkSerial: 0,
@@ -1891,7 +1900,8 @@ func (e *Engine) createPeerConn(pubKey string, allowedIPs []netip.Prefix, agentV
Addr: e.getRosenpassAddr(), Addr: e.getRosenpassAddr(),
PermissiveMode: e.config.RosenpassPermissive, PermissiveMode: e.config.RosenpassPermissive,
}, },
ICEConfig: e.createICEConfig(), ICEConfig: e.createICEConfig(),
NetworkState: e.netState,
} }
serviceDependencies := peer.ServiceDependencies{ serviceDependencies := peer.ServiceDependencies{
+6 -1
View File
@@ -26,6 +26,7 @@ import (
"github.com/netbirdio/netbird/client/internal/portforward" "github.com/netbirdio/netbird/client/internal/portforward"
"github.com/netbirdio/netbird/client/internal/rosenpass" "github.com/netbirdio/netbird/client/internal/rosenpass"
"github.com/netbirdio/netbird/client/internal/stdnet" "github.com/netbirdio/netbird/client/internal/stdnet"
"github.com/netbirdio/netbird/client/netstate"
"github.com/netbirdio/netbird/route" "github.com/netbirdio/netbird/route"
relayClient "github.com/netbirdio/netbird/shared/relay/client" relayClient "github.com/netbirdio/netbird/shared/relay/client"
) )
@@ -93,6 +94,10 @@ type ConnConfig struct {
// ICEConfig ICE protocol configuration // ICEConfig ICE protocol configuration
ICEConfig icemaker.Config ICEConfig icemaker.Config
// NetworkState gates the reconnection guard on OS-reported network
// availability; nil disables gating.
NetworkState *netstate.State
} }
type Conn struct { type Conn struct {
@@ -254,7 +259,7 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error {
conn.handshaker.AddICEListener(conn.workerICE.OnNewOffer) conn.handshaker.AddICEListener(conn.workerICE.OnNewOffer)
} }
conn.guard = guard.NewGuard(conn.Log, conn.isConnectedOnAllWay, conn.config.Timeout, conn.srWatcher) conn.guard = guard.NewGuard(conn.Log, conn.isConnectedOnAllWay, conn.config.Timeout, conn.srWatcher, conn.config.NetworkState)
conn.wg.Add(1) conn.wg.Add(1)
go func() { go func() {
+19 -5
View File
@@ -6,6 +6,8 @@ import (
"github.com/cenkalti/backoff/v4" "github.com/cenkalti/backoff/v4"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/client/netstate"
) )
// ConnStatus represents the connection state as seen by the guard. // ConnStatus represents the connection state as seen by the guard.
@@ -31,20 +33,26 @@ type connStatusFunc func() ConnStatus
// - Relayed connection disconnected // - Relayed connection disconnected
// - ICE candidate changes // - ICE candidate changes
type Guard struct { type Guard struct {
log *log.Entry log *log.Entry
isConnectedOnAllWay connStatusFunc isConnectedOnAllWay connStatusFunc
timeout time.Duration timeout time.Duration
srWatcher *SRWatcher srWatcher *SRWatcher
// netState gates reconnect attempts on OS-reported network availability;
// nil disables gating.
netState *netstate.State
relayedConnDisconnected chan struct{} relayedConnDisconnected chan struct{}
iCEConnDisconnected chan struct{} iCEConnDisconnected chan struct{}
} }
func NewGuard(log *log.Entry, isConnectedFn connStatusFunc, timeout time.Duration, srWatcher *SRWatcher) *Guard { // 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 {
return &Guard{ return &Guard{
log: log, log: log,
isConnectedOnAllWay: isConnectedFn, isConnectedOnAllWay: isConnectedFn,
timeout: timeout, timeout: timeout,
srWatcher: srWatcher, srWatcher: srWatcher,
netState: netState,
relayedConnDisconnected: make(chan struct{}, 1), relayedConnDisconnected: make(chan struct{}, 1),
iCEConnDisconnected: make(chan struct{}, 1), iCEConnDisconnected: make(chan struct{}, 1),
} }
@@ -99,6 +107,12 @@ func (g *Guard) reconnectLoopWithRetry(ctx context.Context, callback func()) {
for { for {
select { select {
case <-tickerChannel: case <-tickerChannel:
// skip attempts while the OS reports no usable network; the ticker
// keeps running so other events remain responsive, and the guard
// resumes via the signal/relay reconnect events once network returns
if !g.netState.IsOnline() {
continue
}
switch g.isConnectedOnAllWay() { switch g.isConnectedOnAllWay() {
case ConnStatusConnected: case ConnStatusConnected:
// all good, nothing to do // all good, nothing to do
@@ -15,7 +15,7 @@ import (
func newTestGuard(status connStatusFunc) *Guard { func newTestGuard(status connStatusFunc) *Guard {
srw := NewSRWatcher(nil, nil, nil, ice.Config{}) srw := NewSRWatcher(nil, nil, nil, ice.Config{})
return NewGuard(log.WithField("test", "guard"), status, 50*time.Millisecond, srw) return NewGuard(log.WithField("test", "guard"), status, 50*time.Millisecond, srw, nil)
} }
// countBackoffTickerGoroutines returns how many goroutines are currently sitting // countBackoffTickerGoroutines returns how many goroutines are currently sitting
+29
View File
@@ -1,11 +1,40 @@
package peer package peer
// ClientState identifies the client connection state delivered via
// Listener.OnStateChanged.
type ClientState int
// Client states. The numeric values cross the gomobile boundary (the mobile
// bindings re-export them as integer constants), so they are a wire format:
// append new states at the end, never reorder or insert.
const (
ClientStateDisconnected ClientState = iota
ClientStateConnected
ClientStateConnecting
ClientStateDisconnecting
// ClientStateNoNetwork is an overlay state: it is never stored as the
// last notification, only derived from ClientStateConnecting while the
// OS reports no usable network (see notifier.effectiveState).
ClientStateNoNetwork
)
// Listener is a callback type about the NetBird network connection state // Listener is a callback type about the NetBird network connection state
type Listener interface { type Listener interface {
// OnStateChanged reports every client state transition. New states are
// delivered only through this callback; the per-state callbacks below
// are kept for compatibility and will be removed once all consumers
// have migrated.
OnStateChanged(state ClientState)
// Deprecated: consume OnStateChanged instead.
OnConnected() OnConnected()
// Deprecated: consume OnStateChanged instead.
OnDisconnected() OnDisconnected()
// Deprecated: consume OnStateChanged instead.
OnConnecting() OnConnecting()
// Deprecated: consume OnStateChanged instead.
OnDisconnecting() OnDisconnecting()
OnAddressChanged(string, string) OnAddressChanged(string, string)
OnPeersListChanged(int) OnPeersListChanged(int)
} }
+62 -30
View File
@@ -4,31 +4,57 @@ import (
"sync" "sync"
) )
const (
stateDisconnected = iota
stateConnected
stateConnecting
stateDisconnecting
)
type notifier struct { type notifier struct {
serverStateLock sync.Mutex serverStateLock sync.Mutex
listenersLock sync.Mutex listenersLock sync.Mutex
listener Listener listener Listener
currentClientState bool currentClientState bool
lastNotification int lastNotification ClientState
lastNumberOfPeers int lastNumberOfPeers int
lastFqdnAddress string lastFqdnAddress string
lastIPAddress string lastIPAddress string
networkAvailable bool
} }
func newNotifier() *notifier { func newNotifier() *notifier {
return &notifier{} return &notifier{
networkAvailable: true,
}
}
// effectiveState maps the computed state to what listeners should see:
// while the OS reports no usable network, "Connecting" would be a lie —
// connection attempts are suspended — so it is reported as NoNetwork.
// Caller must hold serverStateLock.
func (n *notifier) effectiveState(state ClientState) ClientState {
if !n.networkAvailable && state == ClientStateConnecting {
return ClientStateNoNetwork
}
return state
}
// setNetworkAvailable records the OS network availability and re-notifies
// the listener when the flag flips the effective state (Connecting <->
// NoNetwork).
func (n *notifier) setNetworkAvailable(available bool) {
n.serverStateLock.Lock()
if n.networkAvailable == available {
n.serverStateLock.Unlock()
return
}
previous := n.effectiveState(n.lastNotification)
n.networkAvailable = available
current := n.effectiveState(n.lastNotification)
n.serverStateLock.Unlock()
if previous != current {
n.notify(current)
}
} }
func (n *notifier) setListener(listener Listener) { func (n *notifier) setListener(listener Listener) {
n.serverStateLock.Lock() n.serverStateLock.Lock()
lastNotification := n.lastNotification lastNotification := n.effectiveState(n.lastNotification)
numOfPeers := n.lastNumberOfPeers numOfPeers := n.lastNumberOfPeers
fqdnAddress := n.lastFqdnAddress fqdnAddress := n.lastFqdnAddress
address := n.lastIPAddress address := n.lastIPAddress
@@ -61,43 +87,45 @@ func (n *notifier) updateServerStates(mgmState bool, signalState bool) {
} }
n.lastNotification = calculatedState n.lastNotification = calculatedState
effective := n.effectiveState(calculatedState)
n.serverStateLock.Unlock() n.serverStateLock.Unlock()
n.notify(calculatedState) n.notify(effective)
} }
func (n *notifier) clientStart() { func (n *notifier) clientStart() {
n.serverStateLock.Lock() n.serverStateLock.Lock()
n.currentClientState = true n.currentClientState = true
n.lastNotification = stateConnecting n.lastNotification = ClientStateConnecting
effective := n.effectiveState(ClientStateConnecting)
n.serverStateLock.Unlock() n.serverStateLock.Unlock()
n.notify(stateConnecting) n.notify(effective)
} }
func (n *notifier) clientStop() { func (n *notifier) clientStop() {
n.serverStateLock.Lock() n.serverStateLock.Lock()
n.currentClientState = false n.currentClientState = false
n.lastNotification = stateDisconnected n.lastNotification = ClientStateDisconnected
n.serverStateLock.Unlock() n.serverStateLock.Unlock()
n.notify(stateDisconnected) n.notify(ClientStateDisconnected)
} }
func (n *notifier) clientTearDown() { func (n *notifier) clientTearDown() {
n.serverStateLock.Lock() n.serverStateLock.Lock()
n.currentClientState = false n.currentClientState = false
n.lastNotification = stateDisconnecting n.lastNotification = ClientStateDisconnecting
n.serverStateLock.Unlock() n.serverStateLock.Unlock()
n.notify(stateDisconnecting) n.notify(ClientStateDisconnecting)
} }
func (n *notifier) isServerStateChanged(newState int) bool { func (n *notifier) isServerStateChanged(newState ClientState) bool {
return n.lastNotification != newState return n.lastNotification != newState
} }
func (n *notifier) notify(state int) { func (n *notifier) notify(state ClientState) {
n.listenersLock.Lock() n.listenersLock.Lock()
listener := n.listener listener := n.listener
n.listenersLock.Unlock() n.listenersLock.Unlock()
@@ -109,20 +137,20 @@ func (n *notifier) notify(state int) {
notifyListener(listener, state) notifyListener(listener, state)
} }
func (n *notifier) calculateState(managementConn, signalConn bool) int { func (n *notifier) calculateState(managementConn, signalConn bool) ClientState {
if managementConn && signalConn { if managementConn && signalConn {
return stateConnected return ClientStateConnected
} }
if !managementConn && !signalConn && !n.currentClientState { if !managementConn && !signalConn && !n.currentClientState {
return stateDisconnected return ClientStateDisconnected
} }
if n.lastNotification == stateDisconnecting { if n.lastNotification == ClientStateDisconnecting {
return stateDisconnecting return ClientStateDisconnecting
} }
return stateConnecting return ClientStateConnecting
} }
func (n *notifier) peerListChanged(numOfPeers int) { func (n *notifier) peerListChanged(numOfPeers int) {
@@ -159,15 +187,19 @@ func (n *notifier) localAddressChanged(fqdn, address string) {
listener.OnAddressChanged(fqdn, address) listener.OnAddressChanged(fqdn, address)
} }
func notifyListener(l Listener, state int) { func notifyListener(l Listener, state ClientState) {
// legacy per-state callbacks; NoNetwork is delivered only via
// OnStateChanged below
switch state { switch state {
case stateDisconnected: case ClientStateDisconnected:
l.OnDisconnected() l.OnDisconnected()
case stateConnected: case ClientStateConnected:
l.OnConnected() l.OnConnected()
case stateConnecting: case ClientStateConnecting:
l.OnConnecting() l.OnConnecting()
case stateDisconnecting: case ClientStateDisconnecting:
l.OnDisconnecting() l.OnDisconnecting()
} }
l.OnStateChanged(state)
} }
+15 -12
View File
@@ -6,29 +6,32 @@ import (
) )
type mocListener struct { type mocListener struct {
lastState int lastState ClientState
wg sync.WaitGroup wg sync.WaitGroup
peersWg sync.WaitGroup peersWg sync.WaitGroup
peers int peers int
} }
func (l *mocListener) OnConnected() { func (l *mocListener) OnConnected() {
l.lastState = stateConnected l.lastState = ClientStateConnected
l.wg.Done() l.wg.Done()
} }
func (l *mocListener) OnDisconnected() { func (l *mocListener) OnDisconnected() {
l.lastState = stateDisconnected l.lastState = ClientStateDisconnected
l.wg.Done() l.wg.Done()
} }
func (l *mocListener) OnConnecting() { func (l *mocListener) OnConnecting() {
l.lastState = stateConnecting l.lastState = ClientStateConnecting
l.wg.Done() l.wg.Done()
} }
func (l *mocListener) OnDisconnecting() { func (l *mocListener) OnDisconnecting() {
l.lastState = stateDisconnecting l.lastState = ClientStateDisconnecting
l.wg.Done() l.wg.Done()
} }
func (l *mocListener) OnStateChanged(state ClientState) {
}
func (l *mocListener) OnAddressChanged(host, addr string) { func (l *mocListener) OnAddressChanged(host, addr string) {
} }
@@ -57,15 +60,15 @@ func Test_notifier_serverState(t *testing.T) {
type scenario struct { type scenario struct {
name string name string
expected int expected ClientState
mgmState bool mgmState bool
signalState bool signalState bool
} }
scenarios := []scenario{ scenarios := []scenario{
{"connected", stateConnected, true, true}, {"connected", ClientStateConnected, true, true},
{"mgm down", stateConnecting, false, true}, {"mgm down", ClientStateConnecting, false, true},
{"signal down", stateConnecting, true, false}, {"signal down", ClientStateConnecting, true, false},
{"disconnected", stateDisconnected, false, false}, {"disconnected", ClientStateDisconnected, false, false},
} }
for _, tt := range scenarios { for _, tt := range scenarios {
@@ -85,7 +88,7 @@ func Test_notifier_SetListener(t *testing.T) {
listener.setPeersWaiter() listener.setPeersWaiter()
n := newNotifier() n := newNotifier()
n.lastNotification = stateConnecting n.lastNotification = ClientStateConnecting
n.setListener(listener) n.setListener(listener)
listener.wait() listener.wait()
listener.waitPeers() listener.waitPeers()
@@ -99,7 +102,7 @@ func Test_notifier_RemoveListener(t *testing.T) {
listener.setWaiter() listener.setWaiter()
listener.setPeersWaiter() listener.setPeersWaiter()
n := newNotifier() n := newNotifier()
n.lastNotification = stateConnecting n.lastNotification = ClientStateConnecting
n.setListener(listener) n.setListener(listener)
// setListener replays cached state on a goroutine; wait for both the state // setListener replays cached state on a goroutine; wait for both the state
// and peers callbacks to finish so we don't race on listener.peers. // and peers callbacks to finish so we don't race on listener.peers.
+6
View File
@@ -1211,6 +1211,12 @@ func (d *Status) ClientTeardown() {
d.notifyStateChange() d.notifyStateChange()
} }
// SetNetworkAvailable records the OS-reported network availability; while
// unavailable, listeners see NoNetwork instead of Connecting.
func (d *Status) SetNetworkAvailable(available bool) {
d.notifier.setNetworkAvailable(available)
}
// SetConnectionListener set a listener to the notifier // SetConnectionListener set a listener to the notifier
func (d *Status) SetConnectionListener(listener Listener) { func (d *Status) SetConnectionListener(listener Listener) {
d.notifier.setListener(listener) d.notifier.setListener(listener)
+18 -7
View File
@@ -21,6 +21,7 @@ import (
"github.com/netbirdio/netbird/client/internal/listener" "github.com/netbirdio/netbird/client/internal/listener"
"github.com/netbirdio/netbird/client/internal/peer" "github.com/netbirdio/netbird/client/internal/peer"
"github.com/netbirdio/netbird/client/internal/profilemanager" "github.com/netbirdio/netbird/client/internal/profilemanager"
"github.com/netbirdio/netbird/client/netstate"
"github.com/netbirdio/netbird/client/system" "github.com/netbirdio/netbird/client/system"
"github.com/netbirdio/netbird/formatter" "github.com/netbirdio/netbird/formatter"
"github.com/netbirdio/netbird/route" "github.com/netbirdio/netbird/route"
@@ -28,11 +29,6 @@ import (
types "github.com/netbirdio/netbird/upload-server/types" types "github.com/netbirdio/netbird/upload-server/types"
) )
// ConnectionListener export internal Listener for mobile
type ConnectionListener interface {
peer.Listener
}
// RouteListener export internal RouteListener for mobile // RouteListener export internal RouteListener for mobile
type NetworkChangeListener interface { type NetworkChangeListener interface {
listener.NetworkChangeListener listener.NetworkChangeListener
@@ -79,6 +75,10 @@ type Client struct {
onHostDnsFn func([]string) onHostDnsFn func([]string)
dnsManager dns.IosDnsManager dnsManager dns.IosDnsManager
loginComplete bool loginComplete bool
// netState outlives engine restarts: it mirrors the OS connectivity, not
// the engine lifecycle. Run injects it into each new ConnectClient, which
// distributes it to every reconnection loop.
netState *netstate.State
// preloadedConfig holds config loaded from JSON (used on tvOS where file writes are blocked) // preloadedConfig holds config loaded from JSON (used on tvOS where file writes are blocked)
preloadedConfig *profilemanager.Config preloadedConfig *profilemanager.Config
@@ -101,6 +101,7 @@ func NewClient(cfgFile, stateFile, cacheDir, logFilePath, deviceName string, osV
ctxCancelLock: &sync.Mutex{}, ctxCancelLock: &sync.Mutex{},
networkChangeListener: networkChangeListener, networkChangeListener: networkChangeListener,
dnsManager: dnsManager, dnsManager: dnsManager,
netState: netstate.New(),
} }
} }
@@ -176,7 +177,7 @@ func (c *Client) Run(fd int32, interfaceName string, envList *EnvList) error {
c.onHostDnsFn = func([]string) {} c.onHostDnsFn = func([]string) {}
cfg.WgIface = interfaceName cfg.WgIface = interfaceName
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder) connectClient := internal.NewConnectClient(ctx, cfg, c.recorder, internal.WithNetworkState(c.netState))
c.setState(cfg, connectClient) c.setState(cfg, connectClient)
// Persist the latest sync response so DebugBundle can include the network // Persist the latest sync response so DebugBundle can include the network
// map. On iOS this is backed by disk to keep it out of the constrained // map. On iOS this is backed by disk to keep it out of the constrained
@@ -185,6 +186,16 @@ func (c *Client) Run(fd int32, interfaceName string, envList *EnvList) error {
return connectClient.RunOniOS(fd, c.networkChangeListener, c.dnsManager, c.stateFile, c.cacheDir, c.logFilePath) return connectClient.RunOniOS(fd, c.networkChangeListener, c.dnsManager, c.stateFile, c.cacheDir, c.logFilePath)
} }
// SetNetworkAvailable feeds OS-reported network availability into the client
// (e.g. from NWPathMonitor). While unavailable, the internal reconnect loops
// suspend their attempts and the connection listener reports NoNetwork
// instead of Connecting; when availability returns, the loops resume
// immediately with a fresh backoff.
func (c *Client) SetNetworkAvailable(available bool) {
c.netState.Set(available)
c.recorder.SetNetworkAvailable(available)
}
// Stop the internal client and free the resources // Stop the internal client and free the resources
func (c *Client) Stop() { func (c *Client) Stop() {
c.ctxCancelLock.Lock() c.ctxCancelLock.Lock()
@@ -320,7 +331,7 @@ func (c *Client) GetStatusDetails() *StatusDetails {
// SetConnectionListener set the network connection listener // SetConnectionListener set the network connection listener
func (c *Client) SetConnectionListener(listener ConnectionListener) { func (c *Client) SetConnectionListener(listener ConnectionListener) {
c.recorder.SetConnectionListener(listener) c.recorder.SetConnectionListener(connectionListenerAdapter{listener})
} }
// RemoveConnectionListener remove connection listener // RemoveConnectionListener remove connection listener
@@ -0,0 +1,41 @@
//go:build ios
package NetBirdSDK
import (
"github.com/netbirdio/netbird/client/internal/peer"
)
// Client state values delivered via ConnectionListener.OnStateChanged,
// re-exported as basic constants so gomobile emits them into the generated
// bindings. They mirror peer.ClientState*: append-only, never reorder.
const (
ClientStateDisconnected = int(peer.ClientStateDisconnected)
ClientStateConnected = int(peer.ClientStateConnected)
ClientStateConnecting = int(peer.ClientStateConnecting)
ClientStateDisconnecting = int(peer.ClientStateDisconnecting)
ClientStateNoNetwork = int(peer.ClientStateNoNetwork)
)
// ConnectionListener export internal Listener for mobile. It mirrors
// peer.Listener with OnStateChanged taking a plain int (one of the
// ClientState* constants), because gomobile cannot bind named types.
type ConnectionListener interface {
OnStateChanged(state int)
OnConnected()
OnDisconnected()
OnConnecting()
OnDisconnecting()
OnAddressChanged(string, string)
OnPeersListChanged(int)
}
// connectionListenerAdapter adapts the gomobile-facing ConnectionListener to
// peer.Listener, converting the typed state to the int the binding carries.
type connectionListenerAdapter struct {
ConnectionListener
}
func (a connectionListenerAdapter) OnStateChanged(state peer.ClientState) {
a.ConnectionListener.OnStateChanged(int(state))
}
+93
View File
@@ -0,0 +1,93 @@
// Package netstate tracks OS-reported network availability for the client.
//
// A State instance is owned by the platform integration (e.g. the Android or
// iOS bindings, fed from ConnectivityManager callbacks or NWPathMonitor) and
// is injected into the connection retry loops (management, signal, relay,
// peer guards and the top-level connect loop), which consult it to avoid
// burning CPU and battery on reconnect attempts while the device has no
// network at all (e.g. airplane mode), and to reset their backoff as soon as
// the network returns.
//
// Consumers hold a *State that may be nil — every non-mobile platform leaves
// it unset. The read methods are safe on a nil receiver: they report online
// and never block, so consumers behave as if this package did not exist.
package netstate
import (
"context"
"sync"
log "github.com/sirupsen/logrus"
)
// State holds the OS-reported network availability. The zero value is not
// usable; create instances with New.
type State struct {
mu sync.Mutex
online bool
changed chan struct{}
}
// New creates a State that starts online.
func New() *State {
return &State{
online: true,
changed: make(chan struct{}),
}
}
// Set records whether the OS reports any usable network. Transitions wake up
// all Wait callers immediately.
func (s *State) Set(online bool) {
s.mu.Lock()
defer s.mu.Unlock()
if s.online == online {
return
}
s.online = online
close(s.changed)
s.changed = make(chan struct{})
log.Infof("OS network availability changed: online=%t", online)
}
// IsOnline reports whether the OS reports at least one usable network. On a
// nil receiver — no State injected — it reports online.
func (s *State) IsOnline() bool {
if s == nil {
return true
}
s.mu.Lock()
defer s.mu.Unlock()
return s.online
}
// Wait blocks while the network is offline. It reports whether it had to
// wait, so callers can reset their backoff after an outage. It returns early
// with the context error when ctx is done. On a nil receiver — no State
// injected — it returns immediately.
func (s *State) Wait(ctx context.Context) (bool, error) {
if s == nil {
return false, nil
}
waited := false
for {
s.mu.Lock()
if s.online {
s.mu.Unlock()
return waited, nil
}
ch := s.changed
s.mu.Unlock()
if !waited {
waited = true
log.Debugf("network is offline, pausing connection attempts")
}
select {
case <-ctx.Done():
return waited, ctx.Err()
case <-ch:
}
}
}
+170
View File
@@ -0,0 +1,170 @@
package netstate
import (
"context"
"sync"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestNewStateIsOnline(t *testing.T) {
assert.True(t, New().IsOnline(), "a fresh State should start online")
}
func TestSetTogglesOnlineState(t *testing.T) {
s := New()
s.Set(false)
assert.False(t, s.IsOnline(), "state should be offline after Set(false)")
s.Set(true)
assert.True(t, s.IsOnline(), "state should be online after Set(true)")
}
func TestWaitReturnsImmediatelyWhenOnline(t *testing.T) {
s := New()
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
waited, err := s.Wait(ctx)
require.NoError(t, err)
assert.False(t, waited, "Wait should not block when the network is online")
}
func TestWaitBlocksUntilOnline(t *testing.T) {
s := New()
s.Set(false)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
result := make(chan bool, 1)
go func() {
waited, err := s.Wait(ctx)
if err != nil {
result <- false
return
}
result <- waited
}()
// Verify Wait is actually blocking while offline
select {
case <-result:
t.Fatal("Wait should block while the network is offline")
case <-time.After(100 * time.Millisecond):
}
s.Set(true)
select {
case waited := <-result:
assert.True(t, waited, "Wait should report that it had to wait for the network")
case <-time.After(2 * time.Second):
t.Fatal("Wait should return promptly after the network becomes available")
}
}
func TestWaitReturnsOnContextCancel(t *testing.T) {
s := New()
s.Set(false)
ctx, cancel := context.WithCancel(context.Background())
result := make(chan error, 1)
go func() {
_, err := s.Wait(ctx)
result <- err
}()
cancel()
select {
case err := <-result:
assert.ErrorIs(t, err, context.Canceled)
case <-time.After(2 * time.Second):
t.Fatal("Wait should return promptly after context cancellation")
}
}
func TestWaitWakesAllWaiters(t *testing.T) {
s := New()
s.Set(false)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
const waiters = 10
var wg sync.WaitGroup
results := make(chan bool, waiters)
for i := 0; i < waiters; i++ {
wg.Add(1)
go func() {
defer wg.Done()
waited, err := s.Wait(ctx)
if err != nil {
results <- false
return
}
results <- waited
}()
}
time.Sleep(100 * time.Millisecond)
s.Set(true)
wg.Wait()
close(results)
count := 0
for waited := range results {
assert.True(t, waited, "every waiter should report that it waited")
count++
}
assert.Equal(t, waiters, count, "all waiters should have returned")
}
func TestNilStateReadsAreNoops(t *testing.T) {
var s *State
assert.True(t, s.IsOnline(), "nil State should report online")
waited, err := s.Wait(context.Background())
require.NoError(t, err)
assert.False(t, waited, "nil State's Wait should not block")
}
func TestConcurrentSetAndWait(t *testing.T) {
s := New()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var wg sync.WaitGroup
for i := 0; i < 4; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := 0; j < 100; j++ {
s.Set(j%2 == 0)
s.IsOnline()
}
}()
}
for i := 0; i < 4; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := 0; j < 100; j++ {
if _, err := s.Wait(ctx); err != nil {
return
}
}
}()
}
wg.Wait()
}
+31 -3
View File
@@ -21,6 +21,7 @@ import (
"google.golang.org/grpc/connectivity" "google.golang.org/grpc/connectivity"
nbgrpc "github.com/netbirdio/netbird/client/grpc" nbgrpc "github.com/netbirdio/netbird/client/grpc"
"github.com/netbirdio/netbird/client/netstate"
"github.com/netbirdio/netbird/client/system" "github.com/netbirdio/netbird/client/system"
"github.com/netbirdio/netbird/encryption" "github.com/netbirdio/netbird/encryption"
"github.com/netbirdio/netbird/shared/management/domain" "github.com/netbirdio/netbird/shared/management/domain"
@@ -62,6 +63,10 @@ type GrpcClient struct {
connStateCallbackLock sync.RWMutex connStateCallbackLock sync.RWMutex
serverURL string serverURL string
// netState gates the stream retry loop on OS-reported network
// availability; nil (the default) disables gating.
netState *netstate.State
// syncStreamErr holds the last Sync stream error, or nil while the stream // syncStreamErr holds the last Sync stream error, or nil while the stream
// is established and healthy. GetServerKey succeeds even when the peer // is established and healthy. GetServerKey succeeds even when the peer
// cannot sync (e.g. the server returns "settings not found"), so the // cannot sync (e.g. the server returns "settings not found"), so the
@@ -111,8 +116,17 @@ func MaxRecvMsgSize() int {
return size return size
} }
// ClientOption configures optional GrpcClient behavior.
type ClientOption func(*GrpcClient)
// WithNetworkState injects the OS network availability state that gates the
// stream retry loop; without it gating is disabled.
func WithNetworkState(netState *netstate.State) ClientOption {
return func(c *GrpcClient) { c.netState = netState }
}
// NewClient creates a new client to Management service // NewClient creates a new client to Management service
func NewClient(ctx context.Context, addr string, ourPrivateKey wgtypes.Key, tlsEnabled bool) (*GrpcClient, error) { func NewClient(ctx context.Context, addr string, ourPrivateKey wgtypes.Key, tlsEnabled bool, opts ...ClientOption) (*GrpcClient, error) {
var conn *grpc.ClientConn var conn *grpc.ClientConn
var extraOpts []grpc.DialOption var extraOpts []grpc.DialOption
@@ -138,14 +152,18 @@ func NewClient(ctx context.Context, addr string, ourPrivateKey wgtypes.Key, tlsE
realClient := proto.NewManagementServiceClient(conn) realClient := proto.NewManagementServiceClient(conn)
return &GrpcClient{ c := &GrpcClient{
key: ourPrivateKey, key: ourPrivateKey,
realClient: realClient, realClient: realClient,
ctx: ctx, ctx: ctx,
conn: conn, conn: conn,
connStateCallbackLock: sync.RWMutex{}, connStateCallbackLock: sync.RWMutex{},
serverURL: addr, serverURL: addr,
}, nil }
for _, opt := range opts {
opt(c)
}
return c, nil
} }
// GetServerURL returns the management server URL // GetServerURL returns the management server URL
@@ -208,6 +226,16 @@ func (c *GrpcClient) withMgmtStream(
) error { ) error {
backOff := defaultBackoff(ctx) backOff := defaultBackoff(ctx)
operation := func() error { operation := func() error {
// suspend reconnect attempts while the OS reports no usable network.
// Wait only errors on a cancelled context, which means shutdown, so
// stop the loop without reporting a failure.
if waited, err := c.netState.Wait(ctx); err != nil {
log.Debugf("management connection context has been canceled while offline, this usually indicates shutdown")
return nil
} else if waited {
backOff.Reset()
}
log.Debugf("management connection state %v", c.conn.GetState()) log.Debugf("management connection state %v", c.conn.GetState())
connState := c.conn.GetState() connState := c.conn.GetState()
+28 -3
View File
@@ -7,6 +7,8 @@ import (
"github.com/cenkalti/backoff/v4" "github.com/cenkalti/backoff/v4"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/client/netstate"
) )
const defaultMaxBackoffInterval = 60 * time.Second const defaultMaxBackoffInterval = 60 * time.Second
@@ -22,14 +24,19 @@ type Guard struct {
// attempts. // attempts.
maxBackoffInterval time.Duration maxBackoffInterval time.Duration
// netState gates reconnect attempts on OS-reported network availability;
// nil disables gating.
netState *netstate.State
// lastErr is the error from the most recent failed reconnect attempt, // lastErr is the error from the most recent failed reconnect attempt,
// surfaced as the home relay status while disconnected. // surfaced as the home relay status while disconnected.
lastErr atomic.Pointer[error] lastErr atomic.Pointer[error]
} }
// NewGuard creates a new guard for the relay client. A non-positive // NewGuard creates a new guard for the relay client. A non-positive
// maxBackoffInterval falls back to defaultMaxBackoffInterval. // maxBackoffInterval falls back to defaultMaxBackoffInterval. A nil netState
func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration) *Guard { // disables network availability gating.
func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration, netState *netstate.State) *Guard {
if maxBackoffInterval <= 0 { if maxBackoffInterval <= 0 {
maxBackoffInterval = defaultMaxBackoffInterval maxBackoffInterval = defaultMaxBackoffInterval
} }
@@ -38,6 +45,7 @@ func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration) *Guard {
OnReconnected: make(chan struct{}, 1), OnReconnected: make(chan struct{}, 1),
serverPicker: sp, serverPicker: sp,
maxBackoffInterval: maxBackoffInterval, maxBackoffInterval: maxBackoffInterval,
netState: netState,
} }
return g return g
} }
@@ -70,11 +78,21 @@ func (g *Guard) StartReconnectTrys(ctx context.Context, relayClient *Client) {
// start a ticker to pick a new server // start a ticker to pick a new server
ticker := g.exponentTicker(ctx) ticker := g.exponentTicker(ctx)
defer ticker.Stop() defer func() {
ticker.Stop()
}()
for { for {
select { select {
case <-ticker.C: case <-ticker.C:
// suspend reconnect attempts while the OS reports no usable network
if waited, err := g.netState.Wait(ctx); err != nil {
return
} else if waited {
ticker.Stop()
ticker = g.exponentTicker(ctx)
continue
}
if err := g.retry(ctx); err != nil { if err := g.retry(ctx); err != nil {
log.Errorf("failed to pick new Relay server: %s", err) log.Errorf("failed to pick new Relay server: %s", err)
g.setLastError(err) g.setLastError(err)
@@ -104,6 +122,13 @@ func (g *Guard) tryToQuickReconnect(parentCtx context.Context, rc *Client) bool
return false return false
} }
// Re-check after the wait: the disconnect that triggered this reconnect
// is often the first symptom of the network going away, so the
// availability flag typically arrives while we sleep here.
if !g.netState.IsOnline() {
return false
}
log.Infof("try to reconnect to Relay server: %s", rc.connectionURL) log.Infof("try to reconnect to Relay server: %s", rc.connectionURL)
if err := rc.Connect(parentCtx); err != nil { if err := rc.Connect(parentCtx); err != nil {
+9 -1
View File
@@ -12,6 +12,7 @@ import (
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/client/netstate"
relayAuth "github.com/netbirdio/netbird/shared/relay/auth/hmac" relayAuth "github.com/netbirdio/netbird/shared/relay/auth/hmac"
) )
@@ -65,6 +66,12 @@ func WithMaxBackoffInterval(d time.Duration) ManagerOption {
return func(m *Manager) { m.maxBackoffInterval = d } return func(m *Manager) { m.maxBackoffInterval = d }
} }
// WithNetworkState injects the OS network availability state that gates the
// reconnect guard; without it reconnect attempts are not gated.
func WithNetworkState(netState *netstate.State) ManagerOption {
return func(m *Manager) { m.netState = netState }
}
// Manager is a manager for the relay client instances. It establishes one persistent connection to the given relay URL // Manager is a manager for the relay client instances. It establishes one persistent connection to the given relay URL
// and automatically reconnect to them in case disconnection. // and automatically reconnect to them in case disconnection.
// The manager also manage temporary relay connection. If a client wants to communicate with a client on a // The manager also manage temporary relay connection. If a client wants to communicate with a client on a
@@ -92,6 +99,7 @@ type Manager struct {
mtu uint16 mtu uint16
maxBackoffInterval time.Duration maxBackoffInterval time.Duration
netState *netstate.State
cleanupInterval time.Duration cleanupInterval time.Duration
keepUnusedServerTime time.Duration keepUnusedServerTime time.Duration
@@ -129,7 +137,7 @@ func NewManager(ctx context.Context, serverURLs []string, peerID string, mtu uin
opt(m) opt(m)
} }
m.serverPicker.ServerURLs.Store(serverURLs) m.serverPicker.ServerURLs.Store(serverURLs)
m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval) m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval, m.netState)
return m return m
} }
+30 -3
View File
@@ -19,6 +19,7 @@ import (
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
nbgrpc "github.com/netbirdio/netbird/client/grpc" nbgrpc "github.com/netbirdio/netbird/client/grpc"
"github.com/netbirdio/netbird/client/netstate"
"github.com/netbirdio/netbird/encryption" "github.com/netbirdio/netbird/encryption"
"github.com/netbirdio/netbird/shared/management/client" "github.com/netbirdio/netbird/shared/management/client"
"github.com/netbirdio/netbird/shared/signal/proto" "github.com/netbirdio/netbird/shared/signal/proto"
@@ -65,6 +66,10 @@ type GrpcClient struct {
connStateCallback ConnStateNotifier connStateCallback ConnStateNotifier
connStateCallbackLock sync.RWMutex connStateCallbackLock sync.RWMutex
// netState gates the Receive retry loop on OS-reported network
// availability; nil (the default) disables gating.
netState *netstate.State
onReconnectedListenerFn func() onReconnectedListenerFn func()
decryptionWorker *Worker decryptionWorker *Worker
@@ -89,7 +94,16 @@ type GrpcClient struct {
} }
// NewClient creates a new Signal client // NewClient creates a new Signal client
func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled bool) (*GrpcClient, error) { // ClientOption configures optional GrpcClient behavior.
type ClientOption func(*GrpcClient)
// WithNetworkState injects the OS network availability state that gates the
// Receive retry loop; without it gating is disabled.
func WithNetworkState(netState *netstate.State) ClientOption {
return func(c *GrpcClient) { c.netState = netState }
}
func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled bool, opts ...ClientOption) (*GrpcClient, error) {
var conn *grpc.ClientConn var conn *grpc.ClientConn
operation := func() error { operation := func() error {
@@ -109,7 +123,7 @@ func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled boo
log.Debugf("connected to Signal Service: %v", conn.Target()) log.Debugf("connected to Signal Service: %v", conn.Target())
return &GrpcClient{ c := &GrpcClient{
realClient: proto.NewSignalExchangeClient(conn), realClient: proto.NewSignalExchangeClient(conn),
ctx: ctx, ctx: ctx,
signalConn: conn, signalConn: conn,
@@ -117,7 +131,11 @@ func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled boo
mux: sync.Mutex{}, mux: sync.Mutex{},
status: StreamDisconnected, status: StreamDisconnected,
connStateCallbackLock: sync.RWMutex{}, connStateCallbackLock: sync.RWMutex{},
}, nil }
for _, opt := range opts {
opt(c)
}
return c, nil
} }
func (c *GrpcClient) StreamConnected() bool { func (c *GrpcClient) StreamConnected() bool {
@@ -168,6 +186,15 @@ func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Mes
var backOff = defaultBackoff(ctx) var backOff = defaultBackoff(ctx)
operation := func() error { operation := func() error {
// suspend reconnect attempts while the OS reports no usable network.
// Wait only errors on a cancelled context, which means shutdown, so
// stop the loop without reporting a failure.
if waited, err := c.netState.Wait(ctx); err != nil {
log.Debugf("signal connection context has been canceled while offline, this usually indicates shutdown")
return nil
} else if waited {
backOff.Reset()
}
c.notifyStreamDisconnected() c.notifyStreamDisconnected()