Merge branch 'main' into agent-network/provider-credential-check

shared/management/client/grpc.go

@@ -21,8 +21,7 @@ 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/netevents"
 	"github.com/netbirdio/netbird/client/system"
 	"github.com/netbirdio/netbird/encryption"
 	"github.com/netbirdio/netbird/shared/management/domain"
@@ -64,12 +63,9 @@ 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
+	// netMgr gates the stream retry loop on OS-reported network
+	// availability and sweeps the transport on network change.
+	netMgr *netevents.Manager
 
 	// syncStreamErr holds the last Sync stream error, or nil while the stream
 	// is established and healthy. GetServerKey succeeds even when the peer
@@ -123,15 +119,9 @@ func MaxRecvMsgSize() int {
 // 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 }
+// WithNetEvents injects the OS network event handling.
+func WithNetEvents(events *netevents.Manager) Option {
+	return func(c *GrpcClient) { c.netMgr = events }
 }
 
 // NewClient creates a new client to Management service
@@ -152,8 +142,8 @@ func NewClient(ctx context.Context, addr string, ourPrivateKey wgtypes.Key, tlsE
 		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))
+	if c.netMgr != nil {
+		extraOpts = append(extraOpts, nbgrpc.WithSweeper(c.netMgr))
 	}
 
 	var conn *grpc.ClientConn
@@ -235,16 +225,19 @@ func (c *GrpcClient) withMgmtStream(
 	ctx context.Context,
 	handler func(ctx context.Context, serverPubKey wgtypes.Key, backOff backoff.BackOff) error,
 ) error {
-	backOff := c.sweeper.QuickRetryBackoff(ctx, defaultBackoff(ctx), c.netState)
+	backOff := c.netMgr.QuickRetryBackoff(ctx, defaultBackoff(ctx))
 	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 {
+		if waited, err := c.netMgr.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()
+			// dials attempted while offline grew the channel's internal backoff;
+			// reset it too, or the reconnect waits out that timer first
+			c.conn.ResetConnectBackoff()
 		}
 
 		connState := c.conn.GetState()
@@ -273,7 +266,7 @@ func (c *GrpcClient) withMgmtStream(
 		return handler(ctx, *serverPubKey, backOff)
 	}
 
-	err := nbgrpc.Retry(ctx, operation, backOff, c.netState)
+	err := nbgrpc.Retry(ctx, operation, backOff, c.netMgr)
 	if err != nil {
 		log.Warnf("exiting the Management service connection retry loop due to the unrecoverable error: %s", err)
 	}

shared/management/networkmap/decode.go

@@ -275,6 +275,7 @@ func decodePeerCompact(pc *proto.PeerCompact, peerID string) *types.ComponentPee
 		SupportsIPv6:           pc.SupportsIpv6,
 		ServerSSHAllowed:       pc.ServerSshAllowed,
 		AddedWithSSOLogin:      pc.AddedWithSsoLogin,
+		ProxyEmbedded:          pc.ProxyEmbedded,
 	}
 	if pc.LastLoginUnixNano != 0 {
 		peer.LastLogin = time.Unix(0, pc.LastLoginUnixNano)

shared/management/networkmap/encode.go

@@ -272,8 +272,9 @@ func ToProtocolDNSConfig(update nbdns.Config, cache DNSConfigCache, forwardPort
 }
 
 // AppendRemotePeerConfig appends typed peers as proto.RemotePeerConfig
-// entries to dst and returns the result.
-func AppendRemotePeerConfig(dst []*proto.RemotePeerConfig, peers []*types.ComponentPeer, dnsName string, includeIPv6 bool) []*proto.RemotePeerConfig {
+// entries to dst and returns the result. localIsProxy reports whether the peer
+// receiving this config is itself an embedded proxy.
+func AppendRemotePeerConfig(dst []*proto.RemotePeerConfig, peers []*types.ComponentPeer, dnsName string, includeIPv6 bool, localIsProxy bool) []*proto.RemotePeerConfig {
 	for _, rPeer := range peers {
 		allowedIPs := []string{rPeer.IP.String() + "/32"}
 		if includeIPv6 && rPeer.IPv6.IsValid() {
@@ -285,11 +286,24 @@ func AppendRemotePeerConfig(dst []*proto.RemotePeerConfig, peers []*types.Compon
 			SshConfig:    &proto.SSHConfig{SshPubKey: []byte(rPeer.SSHKey)},
 			Fqdn:         rPeer.FQDN(dnsName),
 			AgentVersion: rPeer.AgentVersion,
+			LazyState:    lazyStateFor(localIsProxy, rPeer),
 		})
 	}
 	return dst
 }
 
+// lazyStateFor returns the per-peer lazy override for a remote peer. Connections
+// involving an ephemeral proxy peer on either endpoint default to lazy so shared
+// proxy infrastructure is not kept permanently connected to every peer. All
+// other peers follow the account-wide flag. A future admin-facing per-peer
+// setting can return LazyStateEager here to force a peer always-active.
+func lazyStateFor(localIsProxy bool, rPeer *types.ComponentPeer) proto.LazyState {
+	if localIsProxy || rPeer.ProxyEmbedded {
+		return proto.LazyState_LazyStateLazy
+	}
+	return proto.LazyState_LazyStateDefault
+}
+
 // BuildAuthorizedUsersProto deduplicates user-IDs into a hashed list and
 // builds per-machine-user index maps. Returns (hashedUsers, machineUsers).
 // Errors from individual hash failures are logged via the provided context;

shared/management/networkmap/envelope.go

@@ -74,11 +74,11 @@ func EnvelopeToNetworkMap(ctx context.Context, env *proto.NetworkMapEnvelope, lo
 	protoNM.Routes = ToProtocolRoutes(typedNM.Routes)
 	protoNM.DNSConfig = ToProtocolDNSConfig(typedNM.DNSConfig, nil, dnsFwdPort)
 
-	remotePeers := AppendRemotePeerConfig(nil, typedNM.Peers, dnsName, includeIPv6)
+	remotePeers := AppendRemotePeerConfig(nil, typedNM.Peers, dnsName, includeIPv6, localPeer.ProxyEmbedded)
 	protoNM.RemotePeers = remotePeers
 	protoNM.RemotePeersIsEmpty = len(remotePeers) == 0
 
-	protoNM.OfflinePeers = AppendRemotePeerConfig(nil, typedNM.OfflinePeers, dnsName, includeIPv6)
+	protoNM.OfflinePeers = AppendRemotePeerConfig(nil, typedNM.OfflinePeers, dnsName, includeIPv6, localPeer.ProxyEmbedded)
 
 	firewallRules := ToProtocolFirewallRules(typedNM.FirewallRules, includeIPv6, useSourcePrefixes)
 	protoNM.FirewallRules = firewallRules

shared/management/proto/management.proto

@@ -501,6 +501,22 @@ message RemotePeerConfig {
   string fqdn = 4;
 
   string agentVersion = 5;
+
+  // lazyState is the management per-peer override for lazy (on-demand)
+  // connections to this remote peer. LazyStateDefault follows the account-wide
+  // flag; LazyStateLazy forces lazy; LazyStateEager forces an always-active
+  // connection. A local NB_LAZY_CONN/MDM override still wins over this.
+  LazyState lazyState = 6;
+}
+
+// LazyState is the management per-peer override for lazy connections.
+enum LazyState {
+  // Follow the account-wide lazy connection flag.
+  LazyStateDefault = 0;
+  // Force a lazy (on-demand) connection regardless of the account flag.
+  LazyStateLazy = 1;
+  // Force an always-active connection regardless of the account flag.
+  LazyStateEager = 2;
 }
 
 // SSHConfig represents SSH configurations of a peer.
@@ -1016,6 +1032,11 @@ message PeerCompact {
   // (port 22022) is only added when this flag is set and the peer agent
   // version supports it.
   bool server_ssh_allowed = 13;
+
+  // Mirror of types.Peer.ProxyMeta.Embedded. Connections involving an
+  // ephemeral proxy peer on either endpoint default to lazy, so this bit
+  // feeds the per-peer lazyState emitted in RemotePeerConfig.
+  bool proxy_embedded = 14;
 }
 
 // PolicyCompact is the compact form of a policy rule. Group references use

shared/management/types/component_types.go

@@ -25,6 +25,9 @@ type ComponentPeer struct {
 	LoginExpirationEnabled bool
 	AddedWithSSOLogin      bool
 	LastLogin              time.Time
+	// ProxyEmbedded marks an ephemeral embedded proxy peer. Connections
+	// involving such a peer on either endpoint default to lazy.
+	ProxyEmbedded bool
 }
 
 // FQDN returns the peer's FQDN combined of the peer's DNS label and the system's DNS domain.

shared/relay/client/client.go

@@ -14,7 +14,7 @@ import (
 
 	log "github.com/sirupsen/logrus"
 
-	"github.com/netbirdio/netbird/client/netsweep"
+	"github.com/netbirdio/netbird/client/netevents/sweep"
 	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"
@@ -151,6 +151,14 @@ type transportConn interface {
 	Protocol() string
 }
 
+// NetEvents is the OS network event view the relay consumes: availability
+// gating for the reconnect guard and dial registration for the network change
+// sweep.
+type NetEvents interface {
+	NetworkWatcher
+	StartDial(ctx context.Context) *sweep.Dial
+}
+
 // Client is a client for the relay server. It is responsible for establishing a connection to the relay server and
 // managing connections to other peers. All exported functions are safe to call concurrently. After close the connection,
 // the client can be reused by calling Connect again. When the client is closed, all connections are closed too.
@@ -186,9 +194,10 @@ type Client struct {
 	// 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
+	// netEvents registers the relay dial for the network change sweep; the
+	// read loop reports the disconnect and the guard reconnects. Shared via
+	// the manager.
+	netEvents NetEvents
 	// datagramFallbackTriggered guards a single fallback per connection so a
 	// burst of oversized datagrams triggers one reconnect, not many.
 	datagramFallbackTriggered atomic.Bool
@@ -400,7 +409,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)
+	var dial *sweep.Dial
+	if c.netEvents != nil {
+		dial = c.netEvents.StartDial(ctx)
+	} else {
+		dial = (*sweep.Sweeper)(nil).StartDial(ctx)
+	}
 	defer dial.Release()
 	ctx = dial.Ctx()
 

shared/relay/client/guard.go

@@ -7,8 +7,6 @@ import (
 
 	"github.com/cenkalti/backoff/v4"
 	log "github.com/sirupsen/logrus"
-
-	"github.com/netbirdio/netbird/client/netstate"
 )
 
 const (
@@ -24,6 +22,13 @@ const (
 	verdictSettleWindow = 200 * time.Millisecond
 )
 
+// NetworkWatcher is the availability view the guard gates reconnects on.
+type NetworkWatcher interface {
+	Wait(ctx context.Context) (bool, error)
+	IsOnline() bool
+	WaitSettled(ctx context.Context, budget, settleWindow time.Duration) bool
+}
+
 // Guard manage the reconnection tries to the Relay server in case of disconnection event.
 type Guard struct {
 	// OnNewRelayClient is a channel that is used to notify the relay manager about a new relay client instance.
@@ -35,9 +40,8 @@ type Guard struct {
 	// attempts.
 	maxBackoffInterval time.Duration
 
-	// netState gates reconnect attempts on OS-reported network availability;
-	// nil disables gating.
-	netState *netstate.State
+	// netWatcher gates reconnect attempts on OS-reported network availability.
+	netWatcher NetworkWatcher
 
 	// lastErr is the error from the most recent failed reconnect attempt,
 	// surfaced as the home relay status while disconnected.
@@ -45,9 +49,8 @@ type Guard struct {
 }
 
 // NewGuard creates a new guard for the relay client. A non-positive
-// maxBackoffInterval falls back to defaultMaxBackoffInterval. A nil netState
-// disables network availability gating.
-func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration, netState *netstate.State) *Guard {
+// maxBackoffInterval falls back to defaultMaxBackoffInterval.
+func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration, netWatcher NetworkWatcher) *Guard {
 	if maxBackoffInterval <= 0 {
 		maxBackoffInterval = defaultMaxBackoffInterval
 	}
@@ -56,7 +59,7 @@ func NewGuard(sp *ServerPicker, maxBackoffInterval time.Duration, netState *nets
 		OnReconnected:      make(chan struct{}, 1),
 		serverPicker:       sp,
 		maxBackoffInterval: maxBackoffInterval,
-		netState:           netState,
+		netWatcher:         netWatcher,
 	}
 	return g
 }
@@ -97,12 +100,14 @@ func (g *Guard) StartReconnectTrys(ctx context.Context, relayClient *Client) {
 		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 g.netWatcher != nil {
+				if waited, err := g.netWatcher.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)
@@ -129,13 +134,18 @@ func (g *Guard) tryToQuickReconnect(parentCtx context.Context, rc *Client) bool
 		return false
 	}
 
-	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
+	if g.netWatcher != nil {
+		if ok := g.netWatcher.WaitSettled(parentCtx, quickReconnectBudget, verdictSettleWindow); !ok {
+			return false
+		}
+		// Still offline after the budget: leave the retry to the ticker.
+		if !g.netWatcher.IsOnline() {
+			return false
+		}
+	} else {
+		if cancelled := waitBeforeRetry(parentCtx); !cancelled {
+			return false
+		}
 	}
 
 	log.Infof("try to reconnect to Relay server: %s", rc.connectionURL)
@@ -200,47 +210,14 @@ func (g *Guard) exponentTicker(ctx context.Context) *backoff.Ticker {
 	return backoff.NewTicker(bo)
 }
 
-// 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()
+func waitBeforeRetry(ctx context.Context) bool {
+	timer := time.NewTimer(quickReconnectBudget)
+	defer timer.Stop()
 
-	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)
+	select {
+	case <-timer.C:
+		return true
+	case <-ctx.Done():
+		return false
 	}
 }

shared/relay/client/guard_test.go

@@ -1,30 +0,0 @@
-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,8 +12,6 @@ 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"
 )
 
@@ -67,15 +65,9 @@ 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 }
+// WithNetEvents injects the OS network event handling.
+func WithNetEvents(events NetEvents) ManagerOption {
+	return func(m *Manager) { m.netEvents = events }
 }
 
 // Manager is a manager for the relay client instances. It establishes one persistent connection to the given relay URL
@@ -105,8 +97,7 @@ type Manager struct {
 
 	mtu                uint16
 	maxBackoffInterval time.Duration
-	netState           *netstate.State
-	sweeper            *netsweep.Sweeper
+	netEvents          NetEvents
 
 	cleanupInterval      time.Duration
 	keepUnusedServerTime time.Duration
@@ -143,9 +134,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.NetEvents = m.netEvents
 	m.serverPicker.ServerURLs.Store(serverURLs)
-	m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval, m.netState)
+	m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval, m.netEvents)
 	return m
 }
 
@@ -370,7 +361,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
+	relayClient.netEvents = m.netEvents
 	err := relayClient.Connect(m.ctx)
 	if err != nil {
 		rt.Lock()

shared/relay/client/picker.go

@@ -9,7 +9,6 @@ import (
 
 	log "github.com/sirupsen/logrus"
 
-	"github.com/netbirdio/netbird/client/netsweep"
 	auth "github.com/netbirdio/netbird/shared/relay/auth/hmac"
 )
 
@@ -31,7 +30,7 @@ type ServerPicker struct {
 	MTU               uint16
 	ConnectionTimeout time.Duration
 	TransportFallback *transportFallback
-	Sweeper           *netsweep.Sweeper
+	NetEvents         NetEvents
 }
 
 func (sp *ServerPicker) PickServer(parentCtx context.Context) (*Client, error) {
@@ -75,7 +74,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
+	relayClient.netEvents = sp.NetEvents
 	err := relayClient.Connect(ctx)
 	resultChan <- connResult{
 		RelayClient: relayClient,

shared/signal/client/grpc.go

@@ -19,8 +19,7 @@ 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/client/netevents"
 	"github.com/netbirdio/netbird/encryption"
 	"github.com/netbirdio/netbird/shared/management/client"
 	"github.com/netbirdio/netbird/shared/signal/proto"
@@ -67,12 +66,9 @@ 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
+	// netMgr gates the Receive retry loop on OS-reported network
+	// availability and sweeps the transport on network change.
+	netMgr *netevents.Manager
 
 	onReconnectedListenerFn func()
 
@@ -100,15 +96,9 @@ type GrpcClient struct {
 // 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 }
+// WithNetEvents injects the OS network event handling.
+func WithNetEvents(events *netevents.Manager) Option {
+	return func(c *GrpcClient) { c.netMgr = events }
 }
 
 // NewClient creates a new Signal client
@@ -126,8 +116,8 @@ func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled boo
 	}
 
 	var extraOpts []grpc.DialOption
-	if c.sweeper != nil {
-		extraOpts = append(extraOpts, nbgrpc.WithSweeper(c.sweeper))
+	if c.netMgr != nil {
+		extraOpts = append(extraOpts, nbgrpc.WithSweeper(c.netMgr))
 	}
 
 	var conn *grpc.ClientConn
@@ -198,17 +188,20 @@ 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 {
 
-	backOff := c.sweeper.QuickRetryBackoff(ctx, defaultBackoff(ctx), c.netState)
+	backOff := c.netMgr.QuickRetryBackoff(ctx, defaultBackoff(ctx))
 
 	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 {
+		if waited, err := c.netMgr.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()
+			// dials attempted while offline grew the channel's internal backoff;
+			// reset it too, or the reconnect waits out that timer first
+			c.signalConn.ResetConnectBackoff()
 		}
 
 		c.notifyStreamDisconnected()
@@ -281,7 +274,7 @@ func (c *GrpcClient) Receive(ctx context.Context, msgHandler func(msg *proto.Mes
 		return nil
 	}
 
-	err := nbgrpc.Retry(ctx, operation, backOff, c.netState)
+	err := nbgrpc.Retry(ctx, operation, backOff, c.netMgr)
 	if err != nil {
 		log.Errorf("exiting the Signal service connection retry loop due to the unrecoverable error: %v", err)
 		return err