[client, android] Handle network changes without restarting the engine (#7144)

On network changes the client restarted the whole engine. That is heavy-handed and slow: it tears down working state to recover from a transition the engine could handle itself. This replaces the restart with proper network event handling.

Suspend the retry loops while no network is available. Instead of burning through backoff intervals against an unreachable network, the reconnection loops park until the OS reports a usable network again.

Reconnect immediately on a network switch. When the OS hands us a new network, connections bound to the old one are swept and re-dialed right away, rather than waiting for a timeout to notice they are dead.

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)
 	}

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()

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)
 	}
 }

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")
+}

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()

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,

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