diff --git a/client/android/client.go b/client/android/client.go index a21348f27..71bbe4380 100644 --- a/client/android/client.go +++ b/client/android/client.go @@ -26,6 +26,8 @@ import ( "github.com/netbirdio/netbird/client/internal/routemanager" "github.com/netbirdio/netbird/client/internal/stdnet" "github.com/netbirdio/netbird/client/net" + "github.com/netbirdio/netbird/client/netstate" + "github.com/netbirdio/netbird/client/netsweep" "github.com/netbirdio/netbird/client/system" "github.com/netbirdio/netbird/formatter" "github.com/netbirdio/netbird/route" @@ -40,11 +42,6 @@ const ( AnonymizeLevelStrict = nbAnonymize.LevelStrictString ) -// ConnectionListener export internal Listener for mobile -type ConnectionListener interface { - peer.Listener -} - // TunAdapter export internal TunAdapter for mobile type TunAdapter interface { device.TunAdapter @@ -85,6 +82,13 @@ type Client struct { deviceName string uiVersion string 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 + + // sweeper also outlives engine restarts; NotifyNetworkChange sweeps it. + sweeper *netsweep.Sweeper stateMu sync.RWMutex connectClient *internal.ConnectClient @@ -156,6 +160,8 @@ func NewClient(androidSDKVersion int, deviceName string, uiVersion string, tunAd recorder: peer.NewRecorder(""), ctxCancelLock: &sync.Mutex{}, networkChangeListener: networkChangeListener, + netState: netstate.New(), + sweeper: netsweep.New(), } } @@ -196,7 +202,8 @@ func (c *Client) Run(platformFiles PlatformFiles, urlOpener URLOpener, isAndroid } // todo do not throw error in case of cancelled context ctx = internal.CtxInitState(ctx) - connectClient := internal.NewConnectClient(ctx, cfg, c.recorder) + connectClient := internal.NewConnectClient(ctx, cfg, c.recorder, + internal.WithNetworkState(c.netState), internal.WithSweeper(c.sweeper)) c.setState(cfg, cacheDir, cfgFile, connectClient) // This path runs the interactive SSO flow, so reaching here means the peer // is authenticated again — release the latch Status() reports from. Clear @@ -237,7 +244,8 @@ func (c *Client) RunWithoutLogin(platformFiles PlatformFiles, dns *DNSList, dnsR // todo do not throw error in case of cancelled context ctx = internal.CtxInitState(ctx) - connectClient := internal.NewConnectClient(ctx, cfg, c.recorder) + connectClient := internal.NewConnectClient(ctx, cfg, c.recorder, + internal.WithNetworkState(c.netState), internal.WithSweeper(c.sweeper)) c.setState(cfg, cacheDir, cfgFile, connectClient) return connectClient.RunOnAndroid(c.tunAdapter, c.iFaceDiscover, c.networkChangeListener, slices.Clone(dns.items), dnsReadyListener, stateFile, cacheDir) } @@ -285,6 +293,24 @@ func (c *Client) GetTunSettings() (*TunSettings, error) { }, nil } +// 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) +} + +// NotifyNetworkChange marks the management, signal and relay connections +// stale after the OS switched networks and schedules a sweep that cuts +// whatever has not redialed on the new network by then. The engine and the +// TUN device stay untouched. +func (c *Client) NotifyNetworkChange() { + c.sweeper.MarkNetworkChange() + log.Infof("network change: connections marked stale") +} + // DebugBundle generates a debug bundle, uploads it, and returns the upload key. // It works both with and without a running engine. anonymizeLevel is "default" // or "strict"; strict also anonymizes internal IP ranges, peer names, and @@ -525,7 +551,11 @@ func (c *Client) OnUpdatedHostDNS(list *DNSList) error { // SetConnectionListener set the network connection listener func (c *Client) SetConnectionListener(listener ConnectionListener) { - c.recorder.SetConnectionListener(listener) + if listener == nil { + c.recorder.RemoveConnectionListener() + return + } + c.recorder.SetConnectionListener(connectionListenerAdapter{listener}) } // RemoveConnectionListener remove connection listener diff --git a/client/android/connection_listener.go b/client/android/connection_listener.go new file mode 100644 index 000000000..77c47574b --- /dev/null +++ b/client/android/connection_listener.go @@ -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)) +} diff --git a/client/grpc/dialer_generic.go b/client/grpc/dialer_generic.go index 479575996..8a80525e9 100644 --- a/client/grpc/dialer_generic.go +++ b/client/grpc/dialer_generic.go @@ -16,28 +16,47 @@ import ( "google.golang.org/grpc" nbnet "github.com/netbirdio/netbird/client/net" + "github.com/netbirdio/netbird/client/netsweep" ) func WithCustomDialer(_ bool, _ string) grpc.DialOption { + return grpc.WithContextDialer(dialContext) +} + +// WithSweeper dials like WithCustomDialer but registers connections and +// dials with the sweeper. Append it after WithCustomDialer: gRPC applies +// dial options in order, so the later context dialer wins. +func WithSweeper(sweeper *netsweep.Sweeper) grpc.DialOption { return grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) { - if runtime.GOOS == "linux" { - currentUser, err := user.Current() - if err != nil { - return nil, status.Errorf(codes.FailedPrecondition, "failed to get current user: %v", err) - } + dial := sweeper.StartDial(ctx) + defer dial.Release() - // the custom dialer requires root permissions which are not required for use cases run as non-root - if currentUser.Uid != "0" { - log.Debug("Not running as root, using standard dialer") - dialer := &net.Dialer{} - return dialer.DialContext(ctx, "tcp", addr) - } - } - - conn, err := nbnet.NewDialer().DialContext(ctx, "tcp", addr) + conn, err := dialContext(dial.Ctx(), addr) if err != nil { - return nil, fmt.Errorf("nbnet.NewDialer().DialContext: %w", err) + return nil, err } - return conn, nil + return dial.WrapConn(conn) }) } + +func dialContext(ctx context.Context, addr string) (net.Conn, error) { + if runtime.GOOS == "linux" { + currentUser, err := user.Current() + if err != nil { + return nil, status.Errorf(codes.FailedPrecondition, "failed to get current user: %v", err) + } + + // the custom dialer requires root permissions which are not required for use cases run as non-root + if currentUser.Uid != "0" { + log.Debug("Not running as root, using standard dialer") + dialer := &net.Dialer{} + return dialer.DialContext(ctx, "tcp", addr) + } + } + + conn, err := nbnet.NewDialer().DialContext(ctx, "tcp", addr) + if err != nil { + return nil, fmt.Errorf("nbnet.NewDialer().DialContext: %w", err) + } + return conn, nil +} diff --git a/client/grpc/dialer_js.go b/client/grpc/dialer_js.go index b89ec3c21..8863756d7 100644 --- a/client/grpc/dialer_js.go +++ b/client/grpc/dialer_js.go @@ -3,6 +3,7 @@ package grpc import ( "google.golang.org/grpc" + "github.com/netbirdio/netbird/client/netsweep" "github.com/netbirdio/netbird/util/wsproxy/client" ) @@ -11,3 +12,8 @@ import ( func WithCustomDialer(tlsEnabled bool, component string) grpc.DialOption { return client.WithWebSocketDialer(tlsEnabled, component) } + +// WithSweeper is a no-op on WASM/JS: there is no network change signal. +func WithSweeper(_ *netsweep.Sweeper) grpc.DialOption { + return grpc.EmptyDialOption{} +} diff --git a/client/grpc/retry.go b/client/grpc/retry.go new file mode 100644 index 000000000..754ffa341 --- /dev/null +++ b/client/grpc/retry.go @@ -0,0 +1,49 @@ +package grpc + +import ( + "context" + "errors" + "time" + + "github.com/cenkalti/backoff/v4" + + "github.com/netbirdio/netbird/client/netstate" +) + +// Retry mirrors backoff.Retry, but the sleep between attempts also wakes on +// OS network availability transitions: an operation cut down by a network +// change retries the moment the network settles instead of sleeping through +// the recovery. A nil netState never fires, leaving plain backoff.Retry +// behavior. +func Retry(ctx context.Context, operation backoff.Operation, bo backoff.BackOff, netState *netstate.State) error { + bo.Reset() + for { + err := operation() + if err == nil { + return nil + } + + var permanent *backoff.PermanentError + if errors.As(err, &permanent) { + return permanent.Err + } + + next := bo.NextBackOff() + if next == backoff.Stop { + if cerr := ctx.Err(); cerr != nil { + return cerr + } + return err + } + + timer := time.NewTimer(next) + select { + case <-timer.C: + case <-netState.Changed(): + timer.Stop() + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + } + } +} diff --git a/client/grpc/retry_test.go b/client/grpc/retry_test.go new file mode 100644 index 000000000..4edca47b6 --- /dev/null +++ b/client/grpc/retry_test.go @@ -0,0 +1,91 @@ +package grpc + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/cenkalti/backoff/v4" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/netbirdio/netbird/client/netstate" +) + +func TestRetryWakesOnNetworkChange(t *testing.T) { + ns := netstate.New() + attempts := 0 + operation := func() error { + attempts++ + if attempts == 1 { + return errors.New("cut by network change") + } + return nil + } + + go func() { + time.Sleep(20 * time.Millisecond) + ns.Set(false) + }() + + start := time.Now() + err := Retry(context.Background(), operation, backoff.NewConstantBackOff(time.Minute), ns) + + require.NoError(t, err) + assert.Equal(t, 2, attempts, "network change must cause one immediate retry") + assert.Less(t, time.Since(start), time.Second, "the transition must cut the minute-long sleep short") +} + +func TestRetryPermanentError(t *testing.T) { + sentinel := errors.New("permission denied") + operation := func() error { + return backoff.Permanent(sentinel) + } + + err := Retry(context.Background(), operation, backoff.NewConstantBackOff(time.Millisecond), nil) + assert.ErrorIs(t, err, sentinel, "permanent errors must stop retries") +} + +func TestRetryNilNetState(t *testing.T) { + attempts := 0 + operation := func() error { + attempts++ + if attempts < 3 { + return errors.New("transient") + } + return nil + } + + err := Retry(context.Background(), operation, backoff.NewConstantBackOff(time.Millisecond), nil) + require.NoError(t, err) + assert.Equal(t, 3, attempts, "nil network state must preserve timed retries") +} + +func TestRetryStops(t *testing.T) { + failure := errors.New("still failing") + operation := func() error { + return failure + } + + err := Retry(context.Background(), operation, &backoff.StopBackOff{}, nil) + assert.ErrorIs(t, err, failure, "stop backoff must return the operation error") +} + +func TestRetryCtxCancelDuringSleep(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + operation := func() error { + return errors.New("failing") + } + + go func() { + time.Sleep(20 * time.Millisecond) + cancel() + }() + + start := time.Now() + err := Retry(ctx, operation, backoff.NewConstantBackOff(time.Minute), netstate.New()) + + assert.ErrorIs(t, err, context.Canceled, "context cancellation must stop the retry loop") + assert.Less(t, time.Since(start), time.Second, "context cancellation must interrupt backoff sleep") +} diff --git a/client/internal/connect.go b/client/internal/connect.go index ceb39419e..e45ecca44 100644 --- a/client/internal/connect.go +++ b/client/internal/connect.go @@ -38,6 +38,8 @@ 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" cProto "github.com/netbirdio/netbird/client/proto" "github.com/netbirdio/netbird/client/ssh" sshconfig "github.com/netbirdio/netbird/client/ssh/config" @@ -70,18 +72,42 @@ type ConnectClient struct { updateManager *updater.Manager 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 +} + +// 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 } } func NewConnectClient( ctx context.Context, config *profilemanager.Config, statusRecorder *peer.Status, + opts ...ConnectClientOption, ) *ConnectClient { // 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 // racing the run loop's startup. Callers therefore need not cancel before Stop. runCtx, runCancel := context.WithCancel(ctx) - return &ConnectClient{ + c := &ConnectClient{ ctx: runCtx, runCancel: runCancel, runExited: make(chan struct{}), @@ -89,6 +115,10 @@ func NewConnectClient( statusRecorder: statusRecorder, engineMutex: sync.Mutex{}, } + for _, opt := range opts { + opt(c) + } + return c } func (c *ConnectClient) SetUpdateManager(um *updater.Manager) { @@ -274,6 +304,13 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan 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) engineCtx, cancel := context.WithCancel(c.ctx) @@ -285,7 +322,8 @@ 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) + mgmClient, err := mgm.NewClient(engineCtx, c.config.ManagementURL.Host, myPrivateKey, mgmTlsEnabled, + mgm.WithNetworkState(c.netState), mgm.WithSweeper(c.sweeper)) if err != nil { // On daemon shutdown / Down() the parent context is cancelled // and the dial fails with "context canceled". Wrapping that @@ -360,7 +398,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) + signalClient, err := connectToSignal(engineCtx, loginResp.GetNetbirdConfig(), myPrivateKey, c.netState, c.sweeper) if err != nil { log.Error(err) return wrapErr(err) @@ -396,7 +434,8 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan 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), relayClient.WithSweeper(c.sweeper)) c.statusRecorder.SetRelayMgr(relayManager) if len(relayURLs) > 0 { if token != nil { @@ -424,6 +463,7 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan UpdateManager: c.updateManager, ClientMetrics: c.clientMetrics, MetricsCtx: c.ctx, + NetState: c.netState, }, mobileDependency) engine.SetSyncResponsePersistence(c.persistSyncResponse) c.engine = engine @@ -480,6 +520,16 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan // status stream stuck at Connecting. err = backoff.Retry(operation, backoff.WithContext(backOff, c.ctx)) 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) if s, ok := gstatus.FromError(err); ok && (s.Code() == codes.PermissionDenied) { state.Set(StatusNeedsLogin) @@ -673,7 +723,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) (*signal.GrpcClient, error) { +func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourPrivateKey wgtypes.Key, netState *netstate.State, sweeper *netsweep.Sweeper) (*signal.GrpcClient, error) { var sigTLSEnabled bool if wtConfig.Signal.Protocol == mgmProto.HostConfig_HTTPS { sigTLSEnabled = true @@ -681,7 +731,8 @@ func connectToSignal(ctx context.Context, wtConfig *mgmProto.NetbirdConfig, ourP 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), signal.WithSweeper(sweeper)) 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) diff --git a/client/internal/engine.go b/client/internal/engine.go index d92c360f2..5380651a5 100644 --- a/client/internal/engine.go +++ b/client/internal/engine.go @@ -59,6 +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" cProto "github.com/netbirdio/netbird/client/proto" "github.com/netbirdio/netbird/client/system" nbdns "github.com/netbirdio/netbird/dns" @@ -181,6 +182,9 @@ type EngineServices struct { UpdateManager *updater.Manager ClientMetrics *metrics.ClientMetrics 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. @@ -204,6 +208,10 @@ type Engine struct { config *EngineConfig 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 []*stun.URI // TURNs is a list of STUN servers used by ICE @@ -337,6 +345,7 @@ func NewEngine( syncMsgMux: &sync.Mutex{}, config: config, mobileDep: mobileDep, + netState: services.NetState, STUNs: []*stun.URI{}, TURNs: []*stun.URI{}, networkSerial: 0, @@ -1893,7 +1902,8 @@ func (e *Engine) createPeerConn(pubKey string, allowedIPs []netip.Prefix, agentV Addr: e.getRosenpassAddr(), PermissiveMode: e.config.RosenpassPermissive, }, - ICEConfig: e.createICEConfig(), + ICEConfig: e.createICEConfig(), + NetworkState: e.netState, } serviceDependencies := peer.ServiceDependencies{ diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index f3235ec7f..a3c320027 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -26,6 +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/route" relayClient "github.com/netbirdio/netbird/shared/relay/client" ) @@ -93,6 +94,10 @@ type ConnConfig struct { // ICEConfig ICE protocol configuration ICEConfig icemaker.Config + + // NetworkState gates the reconnection guard on OS-reported network + // availability; nil disables gating. + NetworkState *netstate.State } type Conn struct { @@ -254,7 +259,7 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error { 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) go func() { diff --git a/client/internal/peer/guard/guard.go b/client/internal/peer/guard/guard.go index 6c2e846a9..68d77d318 100644 --- a/client/internal/peer/guard/guard.go +++ b/client/internal/peer/guard/guard.go @@ -6,6 +6,8 @@ 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. @@ -31,20 +33,26 @@ type connStatusFunc func() ConnStatus // - Relayed connection disconnected // - ICE candidate changes type Guard struct { - log *log.Entry - isConnectedOnAllWay connStatusFunc - timeout time.Duration - srWatcher *SRWatcher + log *log.Entry + isConnectedOnAllWay connStatusFunc + timeout time.Duration + srWatcher *SRWatcher + // netState gates reconnect attempts on OS-reported network availability; + // nil disables gating. + netState *netstate.State relayedConnDisconnected 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{ log: log, isConnectedOnAllWay: isConnectedFn, timeout: timeout, srWatcher: srWatcher, + netState: netState, relayedConnDisconnected: make(chan struct{}, 1), iCEConnDisconnected: make(chan struct{}, 1), } @@ -96,9 +104,16 @@ func (g *Guard) reconnectLoopWithRetry(ctx context.Context, callback func()) { iceState := &iceRetryState{log: g.log} defer iceState.reset() + 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() { + continue + } switch g.isConnectedOnAllWay() { case ConnStatusConnected: // all good, nothing to do @@ -135,6 +150,23 @@ func (g *Guard) reconnectLoopWithRetry(ctx context.Context, callback func()) { tickerChannel = ticker.C iceState.reset() + case <-netChanged: + // Re-arm for the next transition before acting on this one. + netChanged = g.netState.Changed() + if !g.netState.IsOnline() { + continue + } + // Ticks skipped while offline drove the backoff towards its + // maximum without ever attempting, and left the ICE budget + // frozen — possibly in hourly mode. Recover on our own so the + // peer does not depend on a signal or relay event that never + // comes when both stayed up across the outage. + g.log.Debugf("network is back, reset reconnection ticker") + ticker.Stop() + ticker = g.newReconnectTicker(ctx) + tickerChannel = ticker.C + iceState.reset() + case <-ctx.Done(): g.log.Debugf("context is done, stop reconnect loop") return diff --git a/client/internal/peer/guard/guard_leak_test.go b/client/internal/peer/guard/guard_leak_test.go index ded3e4aea..3d82ec591 100644 --- a/client/internal/peer/guard/guard_leak_test.go +++ b/client/internal/peer/guard/guard_leak_test.go @@ -15,7 +15,7 @@ import ( func newTestGuard(status connStatusFunc) *Guard { 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 diff --git a/client/internal/peer/guard/guard_netstate_test.go b/client/internal/peer/guard/guard_netstate_test.go new file mode 100644 index 000000000..2ab736428 --- /dev/null +++ b/client/internal/peer/guard/guard_netstate_test.go @@ -0,0 +1,107 @@ +package guard + +import ( + "context" + "sync/atomic" + "testing" + "time" + + log "github.com/sirupsen/logrus" + + "github.com/netbirdio/netbird/client/internal/peer/ice" + "github.com/netbirdio/netbird/client/netstate" +) + +// newTestGuardWithNetState builds a guard with a realistic MaxInterval: the +// backoff must be able to grow well past the outage, as it does in production +// where the timeout is seconds to minutes. +func newTestGuardWithNetState(status connStatusFunc, netState *netstate.State) *Guard { + srw := NewSRWatcher(nil, nil, nil, ice.Config{}) + return NewGuard(log.WithField("test", "guard"), status, 30*time.Second, srw, netState) +} + +// TestGuard_RecoversAfterOfflineToOnline covers a peer that stays disconnected +// across a network outage while neither signal nor relay reports an event — +// both stayed up, as on a short airplane mode toggle over Wi-Fi. +// +// Every tick taken while offline is skipped, but it still advances the +// exponential backoff, so by the time the network returns the next tick can be +// tens of seconds away. Without an explicit reaction to the transition the +// peer waits out that interval for a recovery that could start immediately. +func TestGuard_RecoversAfterOfflineToOnline(t *testing.T) { + netState := netstate.New() + + var attempts atomic.Int32 + g := newTestGuardWithNetState(func() ConnStatus { return ConnStatusDisconnected }, netState) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + // Start from the reconnect ticker (800ms initial interval), the state a + // peer is in after it loses its connection. + go g.Start(ctx, func() { attempts.Add(1) }) + g.SetRelayedConnDisconnected() + + // Let the backoff climb: 0.8s, 1.6s, 3.2s, 6.4s ... every tick is skipped + // while offline, but each one doubles the wait for the next. + netState.Set(false) + time.Sleep(8 * time.Second) + + offlineAttempts := attempts.Load() + if offlineAttempts != 0 { + t.Fatalf("callback ran %d times while offline, want 0", offlineAttempts) + } + + netState.Set(true) + + // The next organic tick is now several seconds out, so anything within + // this window can only come from reacting to the transition itself. + pollCtx, stopPolling := context.WithTimeout(ctx, 2*time.Second) + defer stopPolling() + + select { + case <-pollCtx.Done(): + t.Fatal("peer was not retried within 2s of the network coming back, " + + "with neither a signal nor a relay event to fall back on") + case <-pollUntil(pollCtx, func() bool { return attempts.Load() > 0 }): + } +} + +// TestGuard_OfflineTransitionDoesNotRetry checks the other direction: going +// offline must not itself trigger an attempt. +func TestGuard_OfflineTransitionDoesNotRetry(t *testing.T) { + netState := netstate.New() + + var attempts atomic.Int32 + g := newTestGuardWithNetState(func() ConnStatus { return ConnStatusDisconnected }, netState) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go g.Start(ctx, func() { attempts.Add(1) }) + + netState.Set(false) + time.Sleep(5 * time.Second) + + if got := attempts.Load(); got != 0 { + t.Fatalf("callback ran %d times after going offline, want 0", got) + } +} + +// pollUntil closes the returned channel once cond holds. It gives up when ctx +// is done, so the polling goroutine never outlives the test that started it. +func pollUntil(ctx context.Context, cond func() bool) <-chan struct{} { + done := make(chan struct{}) + go func() { + for { + if cond() { + close(done) + return + } + select { + case <-ctx.Done(): + return + case <-time.After(10 * time.Millisecond): + } + } + }() + return done +} diff --git a/client/internal/peer/listener.go b/client/internal/peer/listener.go index c601fe534..2bb7fcf32 100644 --- a/client/internal/peer/listener.go +++ b/client/internal/peer/listener.go @@ -1,11 +1,40 @@ 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 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() + // Deprecated: consume OnStateChanged instead. OnDisconnected() + // Deprecated: consume OnStateChanged instead. OnConnecting() + // Deprecated: consume OnStateChanged instead. OnDisconnecting() + OnAddressChanged(string, string) OnPeersListChanged(int) } diff --git a/client/internal/peer/notifier.go b/client/internal/peer/notifier.go index 8d1954fe5..1ee1d32ea 100644 --- a/client/internal/peer/notifier.go +++ b/client/internal/peer/notifier.go @@ -4,31 +4,64 @@ import ( "sync" ) -const ( - stateDisconnected = iota - stateConnected - stateConnecting - stateDisconnecting -) - type notifier struct { + // publishLock orders state publication: it is held across computing the + // effective state and handing it to the listener, so a transition cannot + // overtake a newer one and leave the listener on a stale state. + publishLock sync.Mutex serverStateLock sync.Mutex listenersLock sync.Mutex listener Listener currentClientState bool - lastNotification int + lastNotification ClientState lastNumberOfPeers int lastFqdnAddress string lastIPAddress string + networkAvailable bool } func newNotifier() *notifier { - return ¬ifier{} + return ¬ifier{ + 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.publishLock.Lock() + defer n.publishLock.Unlock() + + 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) { n.serverStateLock.Lock() - lastNotification := n.lastNotification + lastNotification := n.effectiveState(n.lastNotification) numOfPeers := n.lastNumberOfPeers fqdnAddress := n.lastFqdnAddress address := n.lastIPAddress @@ -52,6 +85,9 @@ func (n *notifier) removeListener() { } func (n *notifier) updateServerStates(mgmState bool, signalState bool) { + n.publishLock.Lock() + defer n.publishLock.Unlock() + n.serverStateLock.Lock() calculatedState := n.calculateState(mgmState, signalState) @@ -61,43 +97,54 @@ func (n *notifier) updateServerStates(mgmState bool, signalState bool) { } n.lastNotification = calculatedState + effective := n.effectiveState(calculatedState) n.serverStateLock.Unlock() - n.notify(calculatedState) + n.notify(effective) } func (n *notifier) clientStart() { + n.publishLock.Lock() + defer n.publishLock.Unlock() + n.serverStateLock.Lock() n.currentClientState = true - n.lastNotification = stateConnecting + n.lastNotification = ClientStateConnecting + effective := n.effectiveState(ClientStateConnecting) n.serverStateLock.Unlock() - n.notify(stateConnecting) + n.notify(effective) } func (n *notifier) clientStop() { + n.publishLock.Lock() + defer n.publishLock.Unlock() + n.serverStateLock.Lock() n.currentClientState = false - n.lastNotification = stateDisconnected + n.lastNotification = ClientStateDisconnected n.serverStateLock.Unlock() - n.notify(stateDisconnected) + n.notify(ClientStateDisconnected) } func (n *notifier) clientTearDown() { + n.publishLock.Lock() + defer n.publishLock.Unlock() + n.serverStateLock.Lock() n.currentClientState = false - n.lastNotification = stateDisconnecting + n.lastNotification = ClientStateDisconnecting 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 } -func (n *notifier) notify(state int) { +func (n *notifier) notify(state ClientState) { n.listenersLock.Lock() listener := n.listener n.listenersLock.Unlock() @@ -109,20 +156,20 @@ func (n *notifier) notify(state int) { notifyListener(listener, state) } -func (n *notifier) calculateState(managementConn, signalConn bool) int { +func (n *notifier) calculateState(managementConn, signalConn bool) ClientState { if managementConn && signalConn { - return stateConnected + return ClientStateConnected } if !managementConn && !signalConn && !n.currentClientState { - return stateDisconnected + return ClientStateDisconnected } - if n.lastNotification == stateDisconnecting { - return stateDisconnecting + if n.lastNotification == ClientStateDisconnecting { + return ClientStateDisconnecting } - return stateConnecting + return ClientStateConnecting } func (n *notifier) peerListChanged(numOfPeers int) { @@ -159,15 +206,19 @@ func (n *notifier) localAddressChanged(fqdn, address string) { 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 { - case stateDisconnected: + case ClientStateDisconnected: l.OnDisconnected() - case stateConnected: + case ClientStateConnected: l.OnConnected() - case stateConnecting: + case ClientStateConnecting: l.OnConnecting() - case stateDisconnecting: + case ClientStateDisconnecting: l.OnDisconnecting() } + + l.OnStateChanged(state) } diff --git a/client/internal/peer/notifier_concurrent_test.go b/client/internal/peer/notifier_concurrent_test.go new file mode 100644 index 000000000..fcaaaad3b --- /dev/null +++ b/client/internal/peer/notifier_concurrent_test.go @@ -0,0 +1,108 @@ +package peer + +import ( + "sync" + "testing" + "time" +) + +type recordingListener struct { + mu sync.Mutex + states []ClientState + onState func(ClientState) +} + +func (l *recordingListener) OnStateChanged(state ClientState) { + l.mu.Lock() + l.states = append(l.states, state) + hook := l.onState + l.mu.Unlock() + + if hook != nil { + hook(state) + } +} + +func (l *recordingListener) last() (ClientState, bool) { + l.mu.Lock() + defer l.mu.Unlock() + if len(l.states) == 0 { + return 0, false + } + return l.states[len(l.states)-1], true +} + +func (l *recordingListener) snapshot() []ClientState { + l.mu.Lock() + defer l.mu.Unlock() + return append([]ClientState(nil), l.states...) +} + +func (l *recordingListener) OnConnected() {} +func (l *recordingListener) OnDisconnected() {} +func (l *recordingListener) OnConnecting() {} +func (l *recordingListener) OnDisconnecting() {} +func (l *recordingListener) OnAddressChanged(string, string) {} +func (l *recordingListener) OnPeersListChanged(int) {} + +// TestNotifier_ConcurrentAvailabilityFlipOrdersPublication holds the first +// transition inside the listener callback and flips availability again from +// another goroutine while it is parked. The second flip must not publish +// ahead of the one in flight, otherwise the listener ends up on a state the +// notifier already superseded. +func TestNotifier_ConcurrentAvailabilityFlipOrdersPublication(t *testing.T) { + n := newNotifier() + n.currentClientState = true + n.lastNotification = ClientStateConnecting + + entered := make(chan struct{}) + release := make(chan struct{}) + + l := &recordingListener{} + l.onState = func(state ClientState) { + if state != ClientStateNoNetwork { + return + } + l.mu.Lock() + l.onState = nil + l.mu.Unlock() + close(entered) + <-release + } + n.listener = l + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + n.setNetworkAvailable(false) + }() + + <-entered + + flipped := make(chan struct{}) + go func() { + defer close(flipped) + n.setNetworkAvailable(true) + }() + + select { + case <-flipped: + t.Fatal("the online transition published while the offline one was " + + "still in flight; publication is not serialized") + case <-time.After(200 * time.Millisecond): + } + + close(release) + <-flipped + wg.Wait() + + got, ok := l.last() + if !ok { + t.Fatal("listener never observed a state") + } + if got != ClientStateConnecting { + t.Fatalf("listener holds %v after the network came back, want Connecting; sequence: %v", + got, l.snapshot()) + } +} diff --git a/client/internal/peer/notifier_test.go b/client/internal/peer/notifier_test.go index 0b7722b0c..a73016b05 100644 --- a/client/internal/peer/notifier_test.go +++ b/client/internal/peer/notifier_test.go @@ -6,29 +6,32 @@ import ( ) type mocListener struct { - lastState int + lastState ClientState wg sync.WaitGroup peersWg sync.WaitGroup peers int } func (l *mocListener) OnConnected() { - l.lastState = stateConnected + l.lastState = ClientStateConnected l.wg.Done() } func (l *mocListener) OnDisconnected() { - l.lastState = stateDisconnected + l.lastState = ClientStateDisconnected l.wg.Done() } func (l *mocListener) OnConnecting() { - l.lastState = stateConnecting + l.lastState = ClientStateConnecting l.wg.Done() } func (l *mocListener) OnDisconnecting() { - l.lastState = stateDisconnecting + l.lastState = ClientStateDisconnecting l.wg.Done() } +func (l *mocListener) OnStateChanged(state ClientState) { + +} func (l *mocListener) OnAddressChanged(host, addr string) { } @@ -57,15 +60,15 @@ func Test_notifier_serverState(t *testing.T) { type scenario struct { name string - expected int + expected ClientState mgmState bool signalState bool } scenarios := []scenario{ - {"connected", stateConnected, true, true}, - {"mgm down", stateConnecting, false, true}, - {"signal down", stateConnecting, true, false}, - {"disconnected", stateDisconnected, false, false}, + {"connected", ClientStateConnected, true, true}, + {"mgm down", ClientStateConnecting, false, true}, + {"signal down", ClientStateConnecting, true, false}, + {"disconnected", ClientStateDisconnected, false, false}, } for _, tt := range scenarios { @@ -85,7 +88,7 @@ func Test_notifier_SetListener(t *testing.T) { listener.setPeersWaiter() n := newNotifier() - n.lastNotification = stateConnecting + n.lastNotification = ClientStateConnecting n.setListener(listener) listener.wait() listener.waitPeers() @@ -99,7 +102,7 @@ func Test_notifier_RemoveListener(t *testing.T) { listener.setWaiter() listener.setPeersWaiter() n := newNotifier() - n.lastNotification = stateConnecting + n.lastNotification = ClientStateConnecting n.setListener(listener) // 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. diff --git a/client/internal/peer/status.go b/client/internal/peer/status.go index 423ce9b23..24e3e7fac 100644 --- a/client/internal/peer/status.go +++ b/client/internal/peer/status.go @@ -1211,6 +1211,12 @@ func (d *Status) ClientTeardown() { 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 func (d *Status) SetConnectionListener(listener Listener) { d.notifier.setListener(listener) diff --git a/client/ios/NetBirdSDK/client.go b/client/ios/NetBirdSDK/client.go index 5df6f92f1..f92f085ab 100644 --- a/client/ios/NetBirdSDK/client.go +++ b/client/ios/NetBirdSDK/client.go @@ -22,6 +22,8 @@ import ( "github.com/netbirdio/netbird/client/internal/listener" "github.com/netbirdio/netbird/client/internal/peer" "github.com/netbirdio/netbird/client/internal/profilemanager" + "github.com/netbirdio/netbird/client/netstate" + "github.com/netbirdio/netbird/client/netsweep" "github.com/netbirdio/netbird/client/system" "github.com/netbirdio/netbird/formatter" "github.com/netbirdio/netbird/route" @@ -36,11 +38,6 @@ const ( AnonymizeLevelStrict = nbAnonymize.LevelStrictString ) -// ConnectionListener export internal Listener for mobile -type ConnectionListener interface { - peer.Listener -} - // RouteListener export internal RouteListener for mobile type NetworkChangeListener interface { listener.NetworkChangeListener @@ -87,6 +84,12 @@ type Client struct { onHostDnsFn func([]string) dnsManager dns.IosDnsManager 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 + // sweeper also outlives engine restarts; NotifyNetworkChange sweeps it. + sweeper *netsweep.Sweeper // preloadedConfig holds config loaded from JSON (used on tvOS where file writes are blocked) preloadedConfig *profilemanager.Config @@ -109,6 +112,8 @@ func NewClient(cfgFile, stateFile, cacheDir, logFilePath, deviceName string, osV ctxCancelLock: &sync.Mutex{}, networkChangeListener: networkChangeListener, dnsManager: dnsManager, + netState: netstate.New(), + sweeper: netsweep.New(), } } @@ -184,7 +189,8 @@ func (c *Client) Run(fd int32, interfaceName string, envList *EnvList) error { c.onHostDnsFn = func([]string) {} cfg.WgIface = interfaceName - connectClient := internal.NewConnectClient(ctx, cfg, c.recorder) + connectClient := internal.NewConnectClient(ctx, cfg, c.recorder, + internal.WithNetworkState(c.netState), internal.WithSweeper(c.sweeper)) c.setState(cfg, connectClient) // 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 @@ -193,6 +199,25 @@ 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) } +// 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) +} + +// NotifyNetworkChange marks the management, signal and relay connections +// stale after the OS switched networks and schedules a sweep that cuts +// whatever has not redialed on the new network by then. The engine and the +// TUN device stay untouched. +func (c *Client) NotifyNetworkChange() { + c.sweeper.MarkNetworkChange() + log.Infof("network change: connections marked stale") +} + // Stop the internal client and free the resources func (c *Client) Stop() { c.ctxCancelLock.Lock() @@ -331,7 +356,11 @@ func (c *Client) GetStatusDetails() *StatusDetails { // SetConnectionListener set the network connection listener func (c *Client) SetConnectionListener(listener ConnectionListener) { - c.recorder.SetConnectionListener(listener) + if listener == nil { + c.recorder.RemoveConnectionListener() + return + } + c.recorder.SetConnectionListener(connectionListenerAdapter{listener}) } // RemoveConnectionListener remove connection listener diff --git a/client/ios/NetBirdSDK/connection_listener.go b/client/ios/NetBirdSDK/connection_listener.go new file mode 100644 index 000000000..d792537ba --- /dev/null +++ b/client/ios/NetBirdSDK/connection_listener.go @@ -0,0 +1,43 @@ +//go:build ios + +package NetBirdSDK + +import ( + "github.com/netbirdio/netbird/client/internal/peer" +) + +// Client state values, 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 intentionally lacks OnStateChanged for now: adding a method to a gomobile +// interface breaks every Swift implementation, so the iOS app keeps building +// against the legacy per-state callbacks. A follow-up will extend it together +// with the app. +type ConnectionListener interface { + OnConnected() + OnDisconnected() + OnConnecting() + OnDisconnecting() + OnAddressChanged(string, string) + OnPeersListChanged(int) +} + +// connectionListenerAdapter adapts the gomobile-facing ConnectionListener to +// peer.Listener. +type connectionListenerAdapter struct { + ConnectionListener +} + +// OnStateChanged is dropped on iOS until the app adopts the state callback; +// the legacy per-state callbacks continue to fire. +func (a connectionListenerAdapter) OnStateChanged(peer.ClientState) {} diff --git a/client/netstate/netstate.go b/client/netstate/netstate.go new file mode 100644 index 000000000..0d7a1268b --- /dev/null +++ b/client/netstate/netstate.go @@ -0,0 +1,110 @@ +// 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. Platforms without network tracking +// pass a nil *State instead: the read methods treat nil as always online and +// never block, so consumers need no nil guards. +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. Unlike the read methods, Set is not nil-safe: +// it is only for the platform owner that created the State with New. +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 +} + +// Changed returns a channel closed on the next availability transition, for +// callers that already own a select loop and cannot block in Wait. Re-read it +// after every fire: each transition installs a fresh channel. On a nil +// receiver — no State injected — it returns nil, which blocks forever in a +// select, so the caller simply never observes a transition. +func (s *State) Changed() <-chan struct{} { + if s == nil { + return nil + } + s.mu.Lock() + defer s.mu.Unlock() + return s.changed +} + +// 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: + } + } +} diff --git a/client/netstate/netstate_test.go b/client/netstate/netstate_test.go new file mode 100644 index 000000000..ea7015761 --- /dev/null +++ b/client/netstate/netstate_test.go @@ -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() +} diff --git a/client/netsweep/netsweep.go b/client/netsweep/netsweep.go new file mode 100644 index 000000000..46bc0a709 --- /dev/null +++ b/client/netsweep/netsweep.go @@ -0,0 +1,267 @@ +// Package netsweep cuts network-bound activity when the OS switches networks: +// a sweep closes the registered connections and aborts the in-flight dials, so +// their owners redial immediately instead of waiting for the old sockets to +// time out. +// +// A nil *Sweeper disables everything: all methods are nil-safe no-ops. +package netsweep + +import ( + "context" + "errors" + "net" + "sync" + "time" + + "github.com/cenkalti/backoff/v4" + log "github.com/sirupsen/logrus" + + "github.com/netbirdio/netbird/client/netstate" +) + +// DefaultSweepDelay absorbs network flapping while the OS settles on a +// default network before the stale registrations are cut. +const DefaultSweepDelay = 500 * time.Millisecond + +const recentMarkWindow = 3 * time.Second + +// Config customizes a Sweeper. The zero value applies the defaults. +type Config struct { + // SweepDelay overrides DefaultSweepDelay when positive. + SweepDelay time.Duration +} + +// ErrSwept reports that a dial finished after a network change swept its +// registration. The connection is already closed; the caller must treat it +// as a failed dial and redial on the new network. +var ErrSwept = errors.New("netsweep: connection swept by network change") + +// sweepID identifies one registration in a sweeper. Connections and dials +// draw from the same counter, so an id is unique across both registries. +type sweepID uint64 + +type connEntry struct { + conn net.Conn + gen uint64 +} + +// Dial tracks one dial from start to connection registration. It hands the +// dialed connection to the sweeper atomically, so a sweep can never fall +// between the dial finishing and the connection being registered. +type Dial struct { + sweeper *Sweeper + ctx context.Context + cancel context.CancelFunc + id sweepID + done bool // set by a sweep, WrapConn or Release; guarded by sweeper.mu + gen uint64 +} + +// Ctx returns the dial's context. A sweep cancels it, so a dial started on the +// old network aborts instead of waiting out its handshake timeout. +func (d *Dial) Ctx() context.Context { + return d.ctx +} + +// Release ends the dial's registration and cancels its context. It is +// idempotent and safe after WrapConn, so callers can defer it. +func (d *Dial) Release() { + s := d.sweeper + if s == nil { + return + } + + s.mu.Lock() + d.done = true + delete(s.dials, d.id) + s.mu.Unlock() + + d.cancel() +} + +// sweptConn deregisters itself from the sweeper when closed. +type sweptConn struct { + net.Conn + sweeper *Sweeper + id sweepID +} + +func (c *sweptConn) Close() error { + c.sweeper.deregister(c.id) + return c.Conn.Close() +} + +// Sweeper registers live connections and in-flight dials so the +// network-change sweep can cut everything registered before the change. +type Sweeper struct { + mu sync.Mutex + conns map[sweepID]connEntry + dials map[sweepID]*Dial + nextID sweepID + gen uint64 + timer *time.Timer + sweepDelay time.Duration + lastMark time.Time +} + +// New creates an empty sweeper with the default configuration. +func New() *Sweeper { + return NewWithConfig(Config{}) +} + +// NewWithConfig creates an empty sweeper customized by cfg. +func NewWithConfig(cfg Config) *Sweeper { + delay := cfg.SweepDelay + if delay <= 0 { + delay = DefaultSweepDelay + } + return &Sweeper{ + conns: make(map[sweepID]connEntry), + dials: make(map[sweepID]*Dial), + sweepDelay: delay, + } +} + +// StartDial registers an in-flight dial. Dial with Ctx, hand the result to +// WrapConn, and Release the dial when the attempt is over, typically deferred. +func (s *Sweeper) StartDial(ctx context.Context) *Dial { + if s == nil { + return &Dial{ctx: ctx} + } + + ctx, cancel := context.WithCancel(ctx) + d := &Dial{sweeper: s, ctx: ctx, cancel: cancel} + + s.mu.Lock() + d.id = s.nextID + s.nextID++ + d.gen = s.gen + s.dials[d.id] = d + s.mu.Unlock() + + return d +} + +// WrapConn hands conn over to the sweeper. If a sweep ran since StartDial, +// the connection belongs to the old network: it is closed and ErrSwept is +// returned. Otherwise conn is registered against the next sweep and returned +// wrapped, deregistering itself on Close. Call it once, before Release. +func (d *Dial) WrapConn(conn net.Conn) (net.Conn, error) { + s := d.sweeper + if s == nil { + return conn, nil + } + + s.mu.Lock() + if d.done { + s.mu.Unlock() + if err := conn.Close(); err != nil { + log.Debugf("swept dial close error: %v", err) + } + return nil, ErrSwept + } + d.done = true + delete(s.dials, d.id) + id := s.nextID + s.nextID++ + // The conn inherits the dial's generation: the socket was bound to the + // network that was default when the dial started, not when it finished. + s.conns[id] = connEntry{conn: conn, gen: d.gen} + s.mu.Unlock() + + return &sweptConn{Conn: conn, sweeper: s, id: id}, nil +} + +// MarkNetworkChange records that the OS switched networks: everything +// registered so far becomes stale, and a sweep is (re)scheduled after the +// configured delay to cut whatever is still stale by then. Owners that +// redialed in the meantime hold fresh-generation registrations and survive, +// so no cancellation is needed around the sweep. +func (s *Sweeper) MarkNetworkChange() { + if s == nil { + return + } + + s.mu.Lock() + s.gen++ + cutoff := s.gen + s.lastMark = time.Now() + if s.timer != nil { + s.timer.Stop() + } + s.timer = time.AfterFunc(s.sweepDelay, func() { + n := s.sweep(cutoff) + log.Infof("network change sweep: closed %d stale connections", n) + }) + s.mu.Unlock() +} + +// QuickRetryBackoff wraps bo so that after each Reset the first retry comes +// quickly when the disconnect followed a recent network change and the +// network is online. Any other failure keeps bo's spread, so the clients of +// a restarted server still scatter their reconnects. A nil sweeper returns +// bo unchanged. +func (s *Sweeper) QuickRetryBackoff(ctx context.Context, bo backoff.BackOff, netState *netstate.State) backoff.BackOff { + if s == nil { + return bo + } + return backoff.WithContext(newQuickRetryBackoff(bo, s, netState), ctx) +} + +func (s *Sweeper) markedRecently() bool { + if s == nil { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + return !s.lastMark.IsZero() && time.Since(s.lastMark) < recentMarkWindow +} + +// sweep closes the registered connections and aborts the in-flight dials +// older than cutoff, and returns how many connections it closed. A dial +// whose connection was not yet handed to WrapConn is marked, so the late +// WrapConn closes it instead of registering it. +func (s *Sweeper) sweep(cutoff uint64) int { + if s == nil { + return 0 + } + + s.mu.Lock() + var conns []net.Conn + for id, e := range s.conns { + if e.gen < cutoff { + delete(s.conns, id) + conns = append(conns, e.conn) + } + } + var dials []*Dial + for id, d := range s.dials { + if d.gen < cutoff { + d.done = true + delete(s.dials, id) + dials = append(dials, d) + } + } + s.mu.Unlock() + + if len(dials) > 0 { + log.Debugf("aborting %d in-flight dials", len(dials)) + for _, d := range dials { + d.cancel() + } + } + + for _, conn := range conns { + log.Debugf("sweeping connection %s -> %s", conn.LocalAddr(), conn.RemoteAddr()) + if err := conn.Close(); err != nil { + log.Debugf("swept connection close error: %v", err) + } + } + return len(conns) +} + +func (s *Sweeper) deregister(id sweepID) { + s.mu.Lock() + delete(s.conns, id) + s.mu.Unlock() +} diff --git a/client/netsweep/netsweep_test.go b/client/netsweep/netsweep_test.go new file mode 100644 index 000000000..88d660c2d --- /dev/null +++ b/client/netsweep/netsweep_test.go @@ -0,0 +1,241 @@ +package netsweep + +import ( + "context" + "math" + "net" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSweepClosesRegisteredConns(t *testing.T) { + sweeper := New() + + c1 := wrap(t, sweeper, connPair(t)) + c2 := wrap(t, sweeper, connPair(t)) + + assert.Equal(t, 2, sweeper.sweepAll(), "both live connections should be closed") + + // The wrappers must report closed now. + buf := make([]byte, 1) + _, err := c1.Read(buf) + assert.Error(t, err, "first connection should be unusable after the sweep") + _, err = c2.Read(buf) + assert.Error(t, err, "second connection should be unusable after the sweep") + + assert.Equal(t, 0, sweeper.sweepAll(), "second sweep should find nothing") +} + +func TestCloseDeregisters(t *testing.T) { + sweeper := New() + + conn := wrap(t, sweeper, connPair(t)) + require.NoError(t, conn.Close()) + + assert.Equal(t, 0, sweeper.sweepAll(), "closed connection must leave the registry") +} + +func TestCloseIsIdempotent(t *testing.T) { + sweeper := New() + + conn := wrap(t, sweeper, connPair(t)) + require.NoError(t, conn.Close()) + assert.Error(t, conn.Close(), "double close surfaces the underlying error but must not panic") +} + +func TestSweepOnlyAffectsOlderConns(t *testing.T) { + sweeper := New() + + _ = wrap(t, sweeper, connPair(t)) + assert.Equal(t, 1, sweeper.sweepAll()) + + // A connection dialed after the sweep must survive until the next one. + _ = wrap(t, sweeper, connPair(t)) + assert.Equal(t, 1, sweeper.sweepAll(), "post-sweep connection belongs to the next sweep") +} + +func TestSweepAbortsInFlightDials(t *testing.T) { + sweeper := New() + + dial := sweeper.StartDial(context.Background()) + defer dial.Release() + + sweeper.sweepAll() + + assert.ErrorIs(t, dial.Ctx().Err(), context.Canceled, "sweep must cancel the in-flight dial context") +} + +func TestReleasedDialIsNotAborted(t *testing.T) { + sweeper := New() + + // Simulate a dial that finished before the sweep. + released := sweeper.StartDial(context.Background()) + released.Release() + + // A dial still in flight during the sweep. + pending := sweeper.StartDial(context.Background()) + defer pending.Release() + + sweeper.sweepAll() + assert.ErrorIs(t, pending.Ctx().Err(), context.Canceled, "pending dial must be aborted") +} + +func TestSweepBetweenDialAndHandoffClosesConn(t *testing.T) { + sweeper := New() + + dial := sweeper.StartDial(context.Background()) + defer dial.Release() + + // The dial succeeds on the old network, then the sweep lands before the + // connection is handed over. + conn := connPair(t) + assert.Equal(t, 0, sweeper.sweepAll(), "the connection is not registered yet") + + wrapped, err := dial.WrapConn(conn) + require.ErrorIs(t, err, ErrSwept) + require.Nil(t, wrapped) + + buf := make([]byte, 1) + _, err = conn.Read(buf) + assert.Error(t, err, "the old-network connection must be closed, not leaked") + + assert.Equal(t, 0, sweeper.sweepAll(), "nothing may leak into the next sweep") +} + +func TestMarkNetworkChangeSparesFreshConns(t *testing.T) { + sweeper := NewWithConfig(Config{SweepDelay: 10 * time.Millisecond}) + + stale := wrap(t, sweeper, connPair(t)) + sweeper.MarkNetworkChange() + _ = wrap(t, sweeper, connPair(t)) + + _ = stale.SetReadDeadline(time.Now().Add(time.Second)) + buf := make([]byte, 1) + _, err := stale.Read(buf) + require.ErrorIs(t, err, net.ErrClosed, "stale connection must be closed by the delayed sweep") + + assert.Equal(t, 1, sweeper.sweepAll(), "the fresh connection must survive the stale sweep") +} + +func TestMarkNetworkChangeAbortsStaleDials(t *testing.T) { + sweeper := NewWithConfig(Config{SweepDelay: 10 * time.Millisecond}) + + stale := sweeper.StartDial(context.Background()) + defer stale.Release() + sweeper.MarkNetworkChange() + fresh := sweeper.StartDial(context.Background()) + defer fresh.Release() + + assert.Eventually(t, func() bool { + return stale.Ctx().Err() != nil + }, time.Second, 5*time.Millisecond, "stale dial must be aborted by the delayed sweep") + assert.NoError(t, fresh.Ctx().Err(), "post-mark dial must not be aborted") +} + +func TestConnInheritsDialGeneration(t *testing.T) { + sweeper := NewWithConfig(Config{SweepDelay: 20 * time.Millisecond}) + + // The dial starts before the network change but completes after it: the + // socket is bound to the old network, so the sweep must still cut it. + dial := sweeper.StartDial(context.Background()) + defer dial.Release() + sweeper.MarkNetworkChange() + + wrapped, err := dial.WrapConn(connPair(t)) + require.NoError(t, err) + + _ = wrapped.SetReadDeadline(time.Now().Add(time.Second)) + buf := make([]byte, 1) + _, err = wrapped.Read(buf) + require.ErrorIs(t, err, net.ErrClosed, "old-generation connection must be swept") +} + +func TestRepeatedMarksCoalesce(t *testing.T) { + sweeper := NewWithConfig(Config{SweepDelay: 20 * time.Millisecond}) + + first := wrap(t, sweeper, connPair(t)) + sweeper.MarkNetworkChange() + second := wrap(t, sweeper, connPair(t)) + sweeper.MarkNetworkChange() + _ = wrap(t, sweeper, connPair(t)) + + buf := make([]byte, 1) + for _, conn := range []net.Conn{first, second} { + _ = conn.SetReadDeadline(time.Now().Add(time.Second)) + _, err := conn.Read(buf) + require.ErrorIs(t, err, net.ErrClosed, "every pre-mark connection must be swept by the rescheduled sweep") + } + assert.Equal(t, 1, sweeper.sweepAll(), "only the newest-generation connection may remain") +} + +func TestNilSweeperIsNoop(t *testing.T) { + var sweeper *Sweeper + + conn := connPair(t) + dial := sweeper.StartDial(context.Background()) + defer dial.Release() + + wrapped, err := dial.WrapConn(conn) + require.NoError(t, err) + assert.Equal(t, conn, wrapped, "nil sweeper must return the conn unchanged") + assert.NoError(t, dial.Ctx().Err(), "nil sweeper must not cancel the dial context") + assert.Equal(t, 0, sweeper.sweepAll(), "nil sweeper closes nothing") +} + +// wrap registers conn with the sweeper through a completed dial. +func wrap(t *testing.T, sweeper *Sweeper, conn net.Conn) net.Conn { + t.Helper() + + dial := sweeper.StartDial(context.Background()) + defer dial.Release() + + wrapped, err := dial.WrapConn(conn) + require.NoError(t, err) + return wrapped +} + +// connPair dials a loopback TCP connection and keeps the accepted peer open +// until the test ends: a peer that closed early would make the connection +// unreadable on its own, so a read error after the sweep would prove nothing. +func connPair(t *testing.T) net.Conn { + t.Helper() + + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { + if err := l.Close(); err != nil { + t.Logf("listener close error: %v", err) + } + }) + + accepted := make(chan net.Conn, 1) + go func() { + conn, err := l.Accept() + if err != nil { + close(accepted) + return + } + accepted <- conn + }() + + conn, err := net.Dial("tcp", l.Addr().String()) + require.NoError(t, err) + + peer, ok := <-accepted + require.True(t, ok, "listener must accept the dialed connection") + t.Cleanup(func() { + if err := peer.Close(); err != nil { + t.Logf("peer close error: %v", err) + } + }) + + return conn +} + +// sweepAll cuts every registration regardless of generation. +func (s *Sweeper) sweepAll() int { + return s.sweep(math.MaxUint64) +} diff --git a/client/netsweep/quick_retry.go b/client/netsweep/quick_retry.go new file mode 100644 index 000000000..524a5c50c --- /dev/null +++ b/client/netsweep/quick_retry.go @@ -0,0 +1,39 @@ +package netsweep + +import ( + "time" + + "github.com/cenkalti/backoff/v4" + + "github.com/netbirdio/netbird/client/netstate" +) + +const quickRetryDelay = 200 * time.Millisecond + +type quickRetryBackoff struct { + backoff.BackOff + sweeper *Sweeper + netState *netstate.State + used bool +} + +func newQuickRetryBackoff(bo backoff.BackOff, sweeper *Sweeper, netState *netstate.State) *quickRetryBackoff { + return &quickRetryBackoff{ + BackOff: bo, + sweeper: sweeper, + netState: netState, + } +} + +func (b *quickRetryBackoff) NextBackOff() time.Duration { + if !b.used && b.sweeper.markedRecently() && b.netState.IsOnline() { + b.used = true + return quickRetryDelay + } + return b.BackOff.NextBackOff() +} + +func (b *quickRetryBackoff) Reset() { + b.used = false + b.BackOff.Reset() +} diff --git a/client/netsweep/quick_retry_test.go b/client/netsweep/quick_retry_test.go new file mode 100644 index 000000000..5505862c5 --- /dev/null +++ b/client/netsweep/quick_retry_test.go @@ -0,0 +1,58 @@ +package netsweep + +import ( + "context" + "testing" + "time" + + "github.com/cenkalti/backoff/v4" + "github.com/stretchr/testify/assert" +) + +func TestQuickRetryAfterRecentMark(t *testing.T) { + sweeper := New() + sweeper.MarkNetworkChange() + + slow := backoff.NewConstantBackOff(5 * time.Second) + bo := sweeper.QuickRetryBackoff(context.Background(), slow, nil) + + assert.Equal(t, quickRetryDelay, bo.NextBackOff(), "first retry after a mark must be quick") + assert.Equal(t, 5*time.Second, bo.NextBackOff(), "second retry must fall back to the wrapped backoff") + + bo.Reset() + assert.Equal(t, quickRetryDelay, bo.NextBackOff(), "reset must re-arm the quick retry") +} + +func TestQuickRetryWithoutMarkKeepsSpread(t *testing.T) { + sweeper := New() + + slow := backoff.NewConstantBackOff(5 * time.Second) + bo := sweeper.QuickRetryBackoff(context.Background(), slow, nil) + + assert.Equal(t, 5*time.Second, bo.NextBackOff(), "without a mark the wrapped backoff decides") + + sweeper.mu.Lock() + sweeper.lastMark = time.Now().Add(-recentMarkWindow) + sweeper.mu.Unlock() + assert.Equal(t, 5*time.Second, bo.NextBackOff(), "a stale mark must not trigger the quick retry") +} + +func TestQuickRetryNilSweeperPassthrough(t *testing.T) { + var sweeper *Sweeper + + slow := backoff.NewConstantBackOff(5 * time.Second) + bo := sweeper.QuickRetryBackoff(context.Background(), slow, nil) + + assert.Equal(t, backoff.BackOff(slow), bo, "nil sweeper must return the backoff unchanged") +} + +func TestQuickRetryHonorsContext(t *testing.T) { + sweeper := New() + sweeper.MarkNetworkChange() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + bo := sweeper.QuickRetryBackoff(ctx, backoff.NewConstantBackOff(time.Millisecond), nil) + + assert.Equal(t, backoff.Stop, bo.NextBackOff(), "cancelled context must stop the retry loop") +} diff --git a/shared/management/client/grpc.go b/shared/management/client/grpc.go index 81f25900a..cd250b5f7 100644 --- a/shared/management/client/grpc.go +++ b/shared/management/client/grpc.go @@ -21,6 +21,8 @@ import ( "google.golang.org/grpc/connectivity" nbgrpc "github.com/netbirdio/netbird/client/grpc" + "github.com/netbirdio/netbird/client/netstate" + "github.com/netbirdio/netbird/client/netsweep" "github.com/netbirdio/netbird/client/system" "github.com/netbirdio/netbird/encryption" "github.com/netbirdio/netbird/shared/management/domain" @@ -62,6 +64,13 @@ type GrpcClient struct { connStateCallbackLock sync.RWMutex serverURL string + // netState gates the stream retry loop on OS-reported network + // availability; nil (the default) disables gating. + netState *netstate.State + + // sweeper cuts the transport connections on network change; nil disables it. + sweeper *netsweep.Sweeper + // syncStreamErr holds the last Sync stream error, or nil while the stream // is established and healthy. GetServerKey succeeds even when the peer // cannot sync (e.g. the server returns "settings not found"), so the @@ -111,16 +120,43 @@ func MaxRecvMsgSize() int { return size } +// Option configures optional GrpcClient behavior. +type Option 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) Option { + return func(c *GrpcClient) { c.netState = netState } +} + +// WithSweeper injects the network change sweeper. +func WithSweeper(sweeper *netsweep.Sweeper) Option { + return func(c *GrpcClient) { c.sweeper = sweeper } +} + // NewClient creates a new client to Management service -func NewClient(ctx context.Context, addr string, ourPrivateKey wgtypes.Key, tlsEnabled bool) (*GrpcClient, error) { - var conn *grpc.ClientConn +func NewClient(ctx context.Context, addr string, ourPrivateKey wgtypes.Key, tlsEnabled bool, opts ...Option) (*GrpcClient, error) { + // Options apply before dialing: the sweeper must wrap the first connection too. + c := &GrpcClient{ + key: ourPrivateKey, + ctx: ctx, + connStateCallbackLock: sync.RWMutex{}, + serverURL: addr, + } + for _, opt := range opts { + opt(c) + } var extraOpts []grpc.DialOption if maxSize := MaxRecvMsgSize(); maxSize > 0 { extraOpts = append(extraOpts, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxSize))) log.Infof("management gRPC max receive message size set to %d bytes", maxSize) } + if c.sweeper != nil { + extraOpts = append(extraOpts, nbgrpc.WithSweeper(c.sweeper)) + } + var conn *grpc.ClientConn operation := func() error { var err error conn, err = nbgrpc.CreateConnection(ctx, addr, tlsEnabled, wsproxy.ManagementComponent, extraOpts...) @@ -136,16 +172,9 @@ func NewClient(ctx context.Context, addr string, ourPrivateKey wgtypes.Key, tlsE return nil, err } - realClient := proto.NewManagementServiceClient(conn) - - return &GrpcClient{ - key: ourPrivateKey, - realClient: realClient, - ctx: ctx, - conn: conn, - connStateCallbackLock: sync.RWMutex{}, - serverURL: addr, - }, nil + c.conn = conn + c.realClient = proto.NewManagementServiceClient(conn) + return c, nil } // GetServerURL returns the management server URL @@ -206,16 +235,33 @@ func (c *GrpcClient) withMgmtStream( ctx context.Context, handler func(ctx context.Context, serverPubKey wgtypes.Key, backOff backoff.BackOff) error, ) error { - backOff := defaultBackoff(ctx) + backOff := c.sweeper.QuickRetryBackoff(ctx, defaultBackoff(ctx), c.netState) operation := func() error { - log.Debugf("management connection state %v", c.conn.GetState()) - connState := c.conn.GetState() + // 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 //nolint:nilerr // a cancelled context means shutdown, not a retryable failure + } else if waited { + backOff.Reset() + } + connState := c.conn.GetState() + log.Debugf("management connection state %v", connState) if connState == connectivity.Shutdown { return backoff.Permanent(fmt.Errorf("connection to management has been shut down")) - } else if !(connState == connectivity.Ready || connState == connectivity.Idle) { + } + if !(connState == connectivity.Ready || connState == connectivity.Idle) { + // A dial may already be in flight (e.g. the other stream triggered + // it after a network change); wait for it to settle and proceed if + // the channel became usable, instead of burning a backoff round on + // a successful dial. A failed dial errors out as before. c.conn.WaitForStateChange(ctx, connState) - return fmt.Errorf("connection to management is not ready and in %s state", connState) + connState = c.conn.GetState() + if !(connState == connectivity.Ready || connState == connectivity.Idle) { + return fmt.Errorf("connection to management is not ready and in %s state", connState) + } } serverPubKey, err := c.getServerPublicKey() @@ -227,7 +273,7 @@ func (c *GrpcClient) withMgmtStream( return handler(ctx, *serverPubKey, backOff) } - err := backoff.Retry(operation, backOff) + err := nbgrpc.Retry(ctx, operation, backOff, c.netState) if err != nil { log.Warnf("exiting the Management service connection retry loop due to the unrecoverable error: %s", err) } diff --git a/shared/relay/client/client.go b/shared/relay/client/client.go index 8d4aa6020..4fb30b8d9 100644 --- a/shared/relay/client/client.go +++ b/shared/relay/client/client.go @@ -14,6 +14,7 @@ import ( log "github.com/sirupsen/logrus" + "github.com/netbirdio/netbird/client/netsweep" auth "github.com/netbirdio/netbird/shared/relay/auth/hmac" "github.com/netbirdio/netbird/shared/relay/client/dialer" netErr "github.com/netbirdio/netbird/shared/relay/client/dialer/net" @@ -184,6 +185,10 @@ type Client struct { // datagram-sized transport is avoided on subsequent connects. Shared via // the manager. transportFallback *transportFallback + + // sweeper cuts the relay connection on network change; the read loop + // reports the disconnect and the guard reconnects. Shared via the manager. + sweeper *netsweep.Sweeper // datagramFallbackTriggered guards a single fallback per connection so a // burst of oversized datagrams triggers one reconnect, not many. datagramFallbackTriggered atomic.Bool @@ -393,6 +398,12 @@ func (c *Client) Close() error { } func (c *Client) connect(ctx context.Context) (*RelayAddr, error) { + // A sweep cancels this context, so a dial started on the old network + // aborts instead of waiting out its handshake timeout. + dial := c.sweeper.StartDial(ctx) + defer dial.Release() + ctx = dial.Ctx() + mode := transportModeFromEnv() dialers := c.getDialers(mode) @@ -417,12 +428,19 @@ func (c *Client) connect(ctx context.Context) (*RelayAddr, error) { return nil, fmt.Errorf("dial via FQDN: %w", err) } } - c.relayConn = conn - c.datagramFallbackTriggered.Store(false) + // Read the transport off the concrete connection: the sweeper's wrapper + // embeds net.Conn only, so it does not promote Protocol(). if tc, ok := conn.(transportConn); ok { c.transport = tc.Protocol() } + conn, err := dial.WrapConn(conn) + if err != nil { + return nil, fmt.Errorf("register connection: %w", err) + } + c.relayConn = conn + c.datagramFallbackTriggered.Store(false) + instanceURL, err := c.handShake(ctx) if err != nil { cErr := conn.Close() diff --git a/shared/relay/client/guard.go b/shared/relay/client/guard.go index d18534d9d..a62f8772d 100644 --- a/shared/relay/client/guard.go +++ b/shared/relay/client/guard.go @@ -7,9 +7,22 @@ import ( "github.com/cenkalti/backoff/v4" log "github.com/sirupsen/logrus" + + "github.com/netbirdio/netbird/client/netstate" ) -const defaultMaxBackoffInterval = 60 * time.Second +const ( + defaultMaxBackoffInterval = 60 * time.Second + + // quickReconnectBudget bounds how long a quick reconnect waits for the + // network before handing the retry over to the ticker. + quickReconnectBudget = 1500 * time.Millisecond + + // verdictSettleWindow is how long an online verdict must hold before it + // is trusted: the disconnect often precedes the OS offline flag by a few + // milliseconds. + verdictSettleWindow = 200 * time.Millisecond +) // Guard manage the reconnection tries to the Relay server in case of disconnection event. type Guard struct { @@ -22,14 +35,19 @@ type Guard struct { // attempts. 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, // surfaced as the home relay status while disconnected. lastErr atomic.Pointer[error] } // NewGuard creates a new guard for the relay client. A non-positive -// maxBackoffInterval falls back to defaultMaxBackoffInterval. -func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration) *Guard { +// maxBackoffInterval falls back to defaultMaxBackoffInterval. A nil netState +// disables network availability gating. +func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration, netState *netstate.State) *Guard { if maxBackoffInterval <= 0 { maxBackoffInterval = defaultMaxBackoffInterval } @@ -38,6 +56,7 @@ func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration) *Guard { OnReconnected: make(chan struct{}, 1), serverPicker: sp, maxBackoffInterval: maxBackoffInterval, + netState: netState, } return g } @@ -70,11 +89,21 @@ func (g *Guard) StartReconnectTrys(ctx context.Context, relayClient *Client) { // start a ticker to pick a new server ticker := g.exponentTicker(ctx) - defer ticker.Stop() + defer func() { + ticker.Stop() + }() for { select { 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 { log.Errorf("failed to pick new Relay server: %s", err) g.setLastError(err) @@ -100,7 +129,12 @@ func (g *Guard) tryToQuickReconnect(parentCtx context.Context, rc *Client) bool return false } - if cancelled := waiteBeforeRetry(parentCtx); !cancelled { + if ok := g.waitForNetwork(parentCtx); !ok { + return false + } + + // Still offline after the budget: leave the retry to the ticker. + if !g.netState.IsOnline() { return false } @@ -166,14 +200,47 @@ func (g *Guard) exponentTicker(ctx context.Context) *backoff.Ticker { return backoff.NewTicker(bo) } -func waiteBeforeRetry(ctx context.Context) bool { - timer := time.NewTimer(1500 * time.Millisecond) - defer timer.Stop() +// waitForNetwork waits out the settle window while online, or waits for the +// network to return while offline, within the budget. Returns false when ctx +// is cancelled. Without an injected netState it degrades to a fixed +// budget-long sleep, the pre-netstate behavior. +func (g *Guard) waitForNetwork(ctx context.Context) bool { + budget := time.NewTimer(quickReconnectBudget) + defer budget.Stop() - select { - case <-timer.C: - return true - case <-ctx.Done(): - return false + settleWindow := verdictSettleWindow + if g.netState == nil { + settleWindow = quickReconnectBudget + } + settle := time.NewTimer(settleWindow) + defer settle.Stop() + + for { + // Channel first, flag second: a flip in between still fires the channel. + changedCh := g.netState.Changed() + if g.netState.IsOnline() { + select { + case <-settle.C: + return true + case <-changedCh: + case <-ctx.Done(): + return false + } + } else { + select { + case <-budget.C: + return true + case <-changedCh: + case <-ctx.Done(): + return false + } + } + if !settle.Stop() { + select { + case <-settle.C: + default: + } + } + settle.Reset(settleWindow) } } diff --git a/shared/relay/client/guard_test.go b/shared/relay/client/guard_test.go new file mode 100644 index 000000000..0e05783e0 --- /dev/null +++ b/shared/relay/client/guard_test.go @@ -0,0 +1,30 @@ +package client + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + + "github.com/netbirdio/netbird/client/netstate" +) + +func TestWaitForNetworkSettlesAfterOutage(t *testing.T) { + ns := netstate.New() + ns.Set(false) + g := NewGuard(nil, 0, ns) + + const outage = 2 * verdictSettleWindow + start := time.Now() + go func() { + time.Sleep(outage) + ns.Set(true) + }() + + ok := g.waitForNetwork(context.Background()) + elapsed := time.Since(start) + + assert.True(t, ok, "recovered network must let the quick reconnect proceed") + assert.GreaterOrEqual(t, elapsed, outage+verdictSettleWindow, "reconnect must wait a full settle window after the network returns") +} diff --git a/shared/relay/client/manager.go b/shared/relay/client/manager.go index 2f2839d94..80e38ae2d 100644 --- a/shared/relay/client/manager.go +++ b/shared/relay/client/manager.go @@ -12,6 +12,8 @@ import ( log "github.com/sirupsen/logrus" + "github.com/netbirdio/netbird/client/netstate" + "github.com/netbirdio/netbird/client/netsweep" relayAuth "github.com/netbirdio/netbird/shared/relay/auth/hmac" ) @@ -65,6 +67,17 @@ func WithMaxBackoffInterval(d time.Duration) ManagerOption { 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 } +} + +// WithSweeper injects the network change sweeper. +func WithSweeper(sweeper *netsweep.Sweeper) ManagerOption { + return func(m *Manager) { m.sweeper = sweeper } +} + // 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. // The manager also manage temporary relay connection. If a client wants to communicate with a client on a @@ -92,6 +105,8 @@ type Manager struct { mtu uint16 maxBackoffInterval time.Duration + netState *netstate.State + sweeper *netsweep.Sweeper cleanupInterval time.Duration keepUnusedServerTime time.Duration @@ -128,8 +143,9 @@ func NewManager(ctx context.Context, serverURLs []string, peerID string, mtu uin for _, opt := range opts { opt(m) } + m.serverPicker.Sweeper = m.sweeper m.serverPicker.ServerURLs.Store(serverURLs) - m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval) + m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval, m.netState) return m } @@ -354,6 +370,7 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string relayClient := NewClientWithServerIP(serverAddress, serverIP, m.tokenStore, m.peerID, m.mtu) relayClient.SetTransportFallback(m.transportFallback) + relayClient.sweeper = m.sweeper err := relayClient.Connect(m.ctx) if err != nil { rt.Lock() diff --git a/shared/relay/client/picker.go b/shared/relay/client/picker.go index bb721e4ad..72789fadc 100644 --- a/shared/relay/client/picker.go +++ b/shared/relay/client/picker.go @@ -9,6 +9,7 @@ import ( log "github.com/sirupsen/logrus" + "github.com/netbirdio/netbird/client/netsweep" auth "github.com/netbirdio/netbird/shared/relay/auth/hmac" ) @@ -30,6 +31,7 @@ type ServerPicker struct { MTU uint16 ConnectionTimeout time.Duration TransportFallback *transportFallback + Sweeper *netsweep.Sweeper } func (sp *ServerPicker) PickServer(parentCtx context.Context) (*Client, error) { @@ -73,6 +75,7 @@ func (sp *ServerPicker) startConnection(ctx context.Context, resultChan chan con log.Infof("try to connecting to relay server: %s", url) relayClient := NewClient(url, sp.TokenStore, sp.PeerID, sp.MTU) relayClient.SetTransportFallback(sp.TransportFallback) + relayClient.sweeper = sp.Sweeper err := relayClient.Connect(ctx) resultChan <- connResult{ RelayClient: relayClient, diff --git a/shared/signal/client/grpc.go b/shared/signal/client/grpc.go index a07867263..73c482e8f 100644 --- a/shared/signal/client/grpc.go +++ b/shared/signal/client/grpc.go @@ -19,6 +19,8 @@ import ( "google.golang.org/grpc/status" nbgrpc "github.com/netbirdio/netbird/client/grpc" + "github.com/netbirdio/netbird/client/netstate" + "github.com/netbirdio/netbird/client/netsweep" "github.com/netbirdio/netbird/encryption" "github.com/netbirdio/netbird/shared/management/client" "github.com/netbirdio/netbird/shared/signal/proto" @@ -65,6 +67,13 @@ type GrpcClient struct { connStateCallback ConnStateNotifier connStateCallbackLock sync.RWMutex + // netState gates the Receive retry loop on OS-reported network + // availability; nil (the default) disables gating. + netState *netstate.State + + // sweeper cuts the transport connections on network change; nil disables it. + sweeper *netsweep.Sweeper + onReconnectedListenerFn func() decryptionWorker *Worker @@ -88,13 +97,43 @@ type GrpcClient struct { watchdogWg sync.WaitGroup } -// NewClient creates a new Signal client -func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled bool) (*GrpcClient, error) { - var conn *grpc.ClientConn +// Option configures optional GrpcClient behavior. +type Option 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) Option { + return func(c *GrpcClient) { c.netState = netState } +} + +// WithSweeper injects the network change sweeper. +func WithSweeper(sweeper *netsweep.Sweeper) Option { + return func(c *GrpcClient) { c.sweeper = sweeper } +} + +// NewClient creates a new Signal client +func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled bool, opts ...Option) (*GrpcClient, error) { + // Options apply before dialing: the sweeper must wrap the first connection too. + c := &GrpcClient{ + ctx: ctx, + key: key, + mux: sync.Mutex{}, + status: StreamDisconnected, + connStateCallbackLock: sync.RWMutex{}, + } + for _, opt := range opts { + opt(c) + } + + var extraOpts []grpc.DialOption + if c.sweeper != nil { + extraOpts = append(extraOpts, nbgrpc.WithSweeper(c.sweeper)) + } + + var conn *grpc.ClientConn operation := func() error { var err error - conn, err = nbgrpc.CreateConnection(ctx, addr, tlsEnabled, wsproxy.SignalComponent) + conn, err = nbgrpc.CreateConnection(ctx, addr, tlsEnabled, wsproxy.SignalComponent, extraOpts...) if err != nil { return fmt.Errorf("create connection: %w", err) } @@ -109,15 +148,9 @@ func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled boo log.Debugf("connected to Signal Service: %v", conn.Target()) - return &GrpcClient{ - realClient: proto.NewSignalExchangeClient(conn), - ctx: ctx, - signalConn: conn, - key: key, - mux: sync.Mutex{}, - status: StreamDisconnected, - connStateCallbackLock: sync.RWMutex{}, - }, nil + c.signalConn = conn + c.realClient = proto.NewSignalExchangeClient(conn) + return c, nil } func (c *GrpcClient) StreamConnected() bool { @@ -165,19 +198,36 @@ func defaultBackoff(ctx context.Context) backoff.BackOff { // The connection retry logic will try to reconnect for 30 min and if wasn't successful will propagate the error to the function caller. func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Message) error) error { - var backOff = defaultBackoff(ctx) + backOff := c.sweeper.QuickRetryBackoff(ctx, defaultBackoff(ctx), c.netState) 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() - log.Debugf("signal connection state %v", c.signalConn.GetState()) connState := c.signalConn.GetState() + log.Debugf("signal connection state %v", connState) if connState == connectivity.Shutdown { return backoff.Permanent(fmt.Errorf("connection to signal has been shut down")) - } else if !(connState == connectivity.Ready || connState == connectivity.Idle) { + } + if !(connState == connectivity.Ready || connState == connectivity.Idle) { + // A dial may already be in flight (e.g. triggered by another RPC + // after a network change); wait for it to settle and proceed if + // the channel became usable, instead of burning a backoff round on + // a successful dial. A failed dial errors out as before. c.signalConn.WaitForStateChange(ctx, connState) - return fmt.Errorf("connection to signal is not ready and in %s state", connState) + connState = c.signalConn.GetState() + if !(connState == connectivity.Ready || connState == connectivity.Idle) { + return fmt.Errorf("connection to signal is not ready and in %s state", connState) + } } // connect to Signal stream identifying ourselves with a public WireGuard key @@ -231,7 +281,7 @@ func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Mes return nil } - err := backoff.Retry(operation, backOff) + err := nbgrpc.Retry(ctx, operation, backOff, c.netState) if err != nil { log.Errorf("exiting the Signal service connection retry loop due to the unrecoverable error: %v", err) return err