Merge remote-tracking branch 'origin/main' into revert/component-types

Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io>

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