Support per-peer lazy connection state and default proxy peers to lazy

client/internal/conn_mgr.go

@@ -14,6 +14,7 @@ import (
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/client/internal/peerstore"
 	"github.com/netbirdio/netbird/route"
+	mgmProto "github.com/netbirdio/netbird/shared/management/proto"
 )
 
 // lazyForce is the resolved local decision for lazy connections, layered above the
@@ -42,6 +43,9 @@ type ConnMgr struct {
 	iface            lazyconn.WGIface
 	force            lazyForce
 	rosenpassEnabled bool
+	// remoteLazyEnabled caches the account-wide lazy feature flag from management.
+	// It is the default for peers that do not carry a per-peer lazy hint.
+	remoteLazyEnabled bool
 
 	lazyConnMgr *manager.Manager
 	// lazyConnMgrMu guards the lazyConnMgr pointer for readers outside the
@@ -75,69 +79,56 @@ func NewConnMgr(engineConfig *EngineConfig, statusRecorder *peer.Status, peerSto
 	return e
 }
 
-// Start initializes the connection manager. It starts the lazy connection manager when a
-// local override forces it on; with no local override it waits for the management feature flag.
+// Start initializes the connection manager. The lazy connection manager always runs so that
+// per-peer lazy defaults (e.g. proxy peers) work even when the account flag is off; the
+// account flag and the local override decide the default lazy state per peer (see
+// PeerLazyDefault). Rosenpass is the only condition that disables it.
 func (e *ConnMgr) Start(ctx context.Context) {
 	if e.lazyConnMgr != nil {
 		log.Errorf("lazy connection manager is already started")
 		return
 	}
 
-	switch e.force {
-	case lazyForceOff:
-		log.Infof("lazy connection manager is disabled by local override (%s or MDM policy)", lazyconn.EnvLazyConn)
-		e.statusRecorder.UpdateLazyConnection(false)
-		return
-	case lazyForceNone:
-		log.Infof("lazy connection manager is managed by the management feature flag")
-		e.statusRecorder.UpdateLazyConnection(false)
-		return
-	}
-
 	if e.rosenpassEnabled {
-		log.Warnf("rosenpass connection manager is enabled, lazy connection manager will not be started")
+		log.Warnf("rosenpass is enabled, lazy connection manager will not be started")
 		e.statusRecorder.UpdateLazyConnection(false)
 		return
 	}
 
 	e.initLazyManager(ctx)
-	e.statusRecorder.UpdateLazyConnection(true)
+	e.statusRecorder.UpdateLazyConnection(e.PeerLazyDefault(mgmProto.LazyState_LazyStateDefault))
 }
 
-// UpdatedRemoteFeatureFlag is called when the remote feature flag is updated.
-// If enabled, it initializes the lazy connection manager and start it. Do not need to call Start() again.
-// If disabled, then it closes the lazy connection manager and open the connections to all peers.
-func (e *ConnMgr) UpdatedRemoteFeatureFlag(ctx context.Context, enabled bool) error {
-	// a local override (NB_LAZY_CONN or local config) takes precedence over management
-	if e.force != lazyForceNone {
-		return nil
+// UpdatedRemoteFeatureFlag caches the account-wide lazy feature flag. The manager itself is
+// not started or stopped here; the per-sync exclude-list reconciliation moves normal peers
+// between the lazy and always-active sets when the flag flips.
+func (e *ConnMgr) UpdatedRemoteFeatureFlag(_ context.Context, enabled bool) error {
+	e.remoteLazyEnabled = enabled
+	if e.isStartedWithLazyMgr() {
+		e.statusRecorder.UpdateLazyConnection(e.PeerLazyDefault(mgmProto.LazyState_LazyStateDefault))
+	}
+	return nil
+}
+
+// PeerLazyDefault reports whether a peer should be lazy. The local override
+// (NB_LAZY_CONN/MDM) wins over everything; without a local override the
+// management per-peer state applies (LazyStateLazy/Eager force the decision),
+// and LazyStateDefault follows the account-wide flag.
+func (e *ConnMgr) PeerLazyDefault(state mgmProto.LazyState) bool {
+	switch e.force {
+	case lazyForceOn:
+		return true
+	case lazyForceOff:
+		return false
 	}
 
-	if enabled {
-		// if the lazy connection manager is already started, do not start it again
-		if e.lazyConnMgr != nil {
-			return nil
-		}
-
-		if e.rosenpassEnabled {
-			log.Infof("rosenpass connection manager is enabled, lazy connection manager will not be started")
-			e.statusRecorder.UpdateLazyConnection(false)
-			return nil
-		}
-
-		log.Infof("lazy connection manager is enabled by the management feature flag")
-		e.initLazyManager(ctx)
-		e.statusRecorder.UpdateLazyConnection(true)
-		return e.addPeersToLazyConnManager()
-	} else {
-		if e.lazyConnMgr == nil {
-			e.statusRecorder.UpdateLazyConnection(false)
-			return nil
-		}
-		log.Infof("lazy connection manager is disabled by management feature flag")
-		e.closeManager(ctx)
-		e.statusRecorder.UpdateLazyConnection(false)
-		return nil
+	switch state {
+	case mgmProto.LazyState_LazyStateLazy:
+		return true
+	case mgmProto.LazyState_LazyStateEager:
+		return false
+	default:
+		return e.remoteLazyEnabled
 	}
 }
 
@@ -192,12 +183,16 @@ func (e *ConnMgr) SetExcludeList(ctx context.Context, peerIDs map[string]bool) {
 	}
 }
 
-func (e *ConnMgr) AddPeerConn(ctx context.Context, peerKey string, conn *peer.Conn) (exists bool) {
+// AddPeerConn registers a peer connection. permanent requests an always-active connection
+// (the peer belongs to the exclude set: a forwarder, or a peer that is not lazy by policy).
+// Non-permanent peers are handed to the lazy manager. The subsequent SetExcludeList call
+// reconciles membership for existing peers across flag flips.
+func (e *ConnMgr) AddPeerConn(ctx context.Context, peerKey string, conn *peer.Conn, permanent bool) (exists bool) {
 	if success := e.peerStore.AddPeerConn(peerKey, conn); !success {
 		return true
 	}
 
-	if !e.isStartedWithLazyMgr() {
+	if !e.isStartedWithLazyMgr() || permanent {
 		if err := conn.Open(ctx); err != nil {
 			conn.Log.Errorf("failed to open connection: %v", err)
 		}
@@ -316,46 +311,6 @@ func (e *ConnMgr) initLazyManager(engineCtx context.Context) {
 	}()
 }
 
-func (e *ConnMgr) addPeersToLazyConnManager() error {
-	peers := e.peerStore.PeersPubKey()
-	lazyPeerCfgs := make([]lazyconn.PeerConfig, 0, len(peers))
-	for _, peerID := range peers {
-		var peerConn *peer.Conn
-		var exists bool
-		if peerConn, exists = e.peerStore.PeerConn(peerID); !exists {
-			log.Warnf("failed to find peer conn for peerID: %s", peerID)
-			continue
-		}
-
-		lazyPeerCfg := lazyconn.PeerConfig{
-			PublicKey:  peerID,
-			AllowedIPs: peerConn.WgConfig().AllowedIps,
-			PeerConnID: peerConn.ConnID(),
-			Log:        peerConn.Log,
-		}
-		lazyPeerCfgs = append(lazyPeerCfgs, lazyPeerCfg)
-	}
-
-	return e.lazyConnMgr.AddActivePeers(lazyPeerCfgs)
-}
-
-func (e *ConnMgr) closeManager(ctx context.Context) {
-	if e.lazyConnMgr == nil {
-		return
-	}
-
-	e.lazyCtxCancel()
-	e.wg.Wait()
-
-	e.lazyConnMgrMu.Lock()
-	e.lazyConnMgr = nil
-	e.lazyConnMgrMu.Unlock()
-
-	for _, peerID := range e.peerStore.PeersPubKey() {
-		e.peerStore.PeerConnOpen(ctx, peerID)
-	}
-}
-
 func (e *ConnMgr) isStartedWithLazyMgr() bool {
 	return e.lazyConnMgr != nil && e.lazyCtxCancel != nil
 }

client/internal/conn_mgr_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/client/internal/peerstore"
 	"github.com/netbirdio/netbird/monotime"
+	mgmProto "github.com/netbirdio/netbird/shared/management/proto"
 )
 
 func TestResolveLazyForce(t *testing.T) {
@@ -138,4 +139,91 @@ func TestInactivityThresholdEnv(t *testing.T) {
 	}
 }
 
+func TestPeerLazyDefault(t *testing.T) {
+	tests := []struct {
+		name          string
+		force         lazyForce
+		remoteEnabled bool
+		state         mgmProto.LazyState
+		want          bool
+	}{
+		{name: "force on wins over eager state", force: lazyForceOn, state: mgmProto.LazyState_LazyStateEager, want: true},
+		{name: "force off wins over lazy state", force: lazyForceOff, remoteEnabled: true, state: mgmProto.LazyState_LazyStateLazy, want: false},
+		{name: "none, default, account off -> active", force: lazyForceNone, state: mgmProto.LazyState_LazyStateDefault, want: false},
+		{name: "none, default, account on -> lazy", force: lazyForceNone, remoteEnabled: true, state: mgmProto.LazyState_LazyStateDefault, want: true},
+		{name: "none, lazy state, account off -> lazy", force: lazyForceNone, state: mgmProto.LazyState_LazyStateLazy, want: true},
+		{name: "none, eager state, account on -> active", force: lazyForceNone, remoteEnabled: true, state: mgmProto.LazyState_LazyStateEager, want: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := &ConnMgr{force: tt.force, remoteLazyEnabled: tt.remoteEnabled}
+			if got := e.PeerLazyDefault(tt.state); got != tt.want {
+				t.Fatalf("PeerLazyDefault(%v) = %v, want %v", tt.state, got, tt.want)
+			}
+		})
+	}
+}
+
 func durPtr(d time.Duration) *time.Duration { return &d }
+
+// TestToExcludedLazyPeers covers the per-peer lazy classification (proxy vs
+// normal, across the force/account-flag matrix). Forwarder-target exclusion is
+// covered by TestToExcludedLazyPeers_ForwardTarget.
+func TestToExcludedLazyPeers(t *testing.T) {
+	const (
+		normalKey = "normal"
+		lazyKey   = "lazy-state"
+		eagerKey  = "eager-state"
+	)
+
+	peers := []*mgmProto.RemotePeerConfig{
+		{WgPubKey: normalKey, AllowedIps: []string{"100.64.0.1/32"}},
+		{WgPubKey: lazyKey, AllowedIps: []string{"100.64.0.2/32"}, LazyState: mgmProto.LazyState_LazyStateLazy},
+		{WgPubKey: eagerKey, AllowedIps: []string{"100.64.0.3/32"}, LazyState: mgmProto.LazyState_LazyStateEager},
+	}
+
+	tests := []struct {
+		name          string
+		force         lazyForce
+		remoteEnabled bool
+		want          map[string]bool
+	}{
+		{
+			name:  "account off: lazy-state peer lazy, normal + eager active",
+			force: lazyForceNone, remoteEnabled: false,
+			want: map[string]bool{normalKey: true, eagerKey: true},
+		},
+		{
+			name:  "account on: only eager-state peer active",
+			force: lazyForceNone, remoteEnabled: true,
+			want: map[string]bool{eagerKey: true},
+		},
+		{
+			name:  "force off: everything active",
+			force: lazyForceOff, remoteEnabled: true,
+			want: map[string]bool{normalKey: true, lazyKey: true, eagerKey: true},
+		},
+		{
+			name:  "force on: nothing active",
+			force: lazyForceOn, remoteEnabled: false,
+			want: map[string]bool{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := &Engine{connMgr: &ConnMgr{force: tt.force, remoteLazyEnabled: tt.remoteEnabled}}
+			got := e.toExcludedLazyPeers(nil, peers)
+
+			if len(got) != len(tt.want) {
+				t.Fatalf("toExcludedLazyPeers() = %v, want %v", got, tt.want)
+			}
+			for k := range tt.want {
+				if !got[k] {
+					t.Fatalf("expected peer %s excluded, got %v", k, got)
+				}
+			}
+		})
+	}
+}

client/internal/engine.go

@@ -830,7 +830,7 @@ func (e *Engine) blockLanAccess() {
 
 // modifyPeers updates peers that have been modified (e.g. IP address has been changed).
 // It closes the existing connection, removes it from the peerConns map, and creates a new one.
-func (e *Engine) modifyPeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
+func (e *Engine) modifyPeers(peersUpdate []*mgmProto.RemotePeerConfig, forwardingRules []firewallManager.ForwardRule) error {
 
 	// first, check if peers have been modified
 	var modified []*mgmProto.RemotePeerConfig
@@ -869,7 +869,7 @@ func (e *Engine) modifyPeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
 	}
 	// third, add the peer connections again
 	for _, p := range modified {
-		err := e.addNewPeer(p)
+		err := e.addNewPeer(p, forwardingRules)
 		if err != nil {
 			return err
 		}
@@ -1570,15 +1570,14 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) error {
 	e.updateOfflinePeers(networkMap.GetOfflinePeers())
 	done()
 
-	remotePeers, err := e.reconcilePeers(networkMap)
+	remotePeers, err := e.reconcilePeers(networkMap, forwardingRules)
 	if err != nil {
 		return err
 	}
 
 	// must set the exclude list after the peers are added. Without it the manager can not figure out the peers parameters from the store
 	done = e.phase("lazy_exclude")
-	excludedLazyPeers := e.toExcludedLazyPeers(forwardingRules, remotePeers)
-	e.connMgr.SetExcludeList(e.ctx, excludedLazyPeers)
+	e.connMgr.SetExcludeList(e.ctx, e.toExcludedLazyPeers(forwardingRules, remotePeers))
 	done()
 
 	e.networkSerial = serial
@@ -1588,8 +1587,10 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) error {
 
 // reconcilePeers applies the remote peer list from the network map (removing,
 // modifying and adding peers, then updating SSH config) and returns the remote
-// peers with our own peer filtered out, for use by later sync steps.
-func (e *Engine) reconcilePeers(networkMap *mgmProto.NetworkMap) ([]*mgmProto.RemotePeerConfig, error) {
+// peers with our own peer filtered out, for use by later sync steps. The
+// forwarding rules are used to decide whether a newly added peer needs an
+// always-active connection.
+func (e *Engine) reconcilePeers(networkMap *mgmProto.NetworkMap, forwardingRules []firewallManager.ForwardRule) ([]*mgmProto.RemotePeerConfig, error) {
 	// Filter out own peer from the remote peers list
 	localPubKey := e.config.WgPrivateKey.PublicKey().String()
 	remotePeers := make([]*mgmProto.RemotePeerConfig, 0, len(networkMap.GetRemotePeers()))
@@ -1617,14 +1618,14 @@ func (e *Engine) reconcilePeers(networkMap *mgmProto.NetworkMap) ([]*mgmProto.Re
 	}
 
 	done = e.phase("modified_peers")
-	err = e.modifyPeers(remotePeers)
+	err = e.modifyPeers(remotePeers, forwardingRules)
 	done()
 	if err != nil {
 		return nil, err
 	}
 
 	done = e.phase("added_peers")
-	err = e.addNewPeers(remotePeers)
+	err = e.addNewPeers(remotePeers, forwardingRules)
 	done()
 	if err != nil {
 		return nil, err
@@ -1820,9 +1821,9 @@ func addrToString(addr netip.Addr) string {
 }
 
 // addNewPeers adds peers that were not know before but arrived from the Management service with the update
-func (e *Engine) addNewPeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
+func (e *Engine) addNewPeers(peersUpdate []*mgmProto.RemotePeerConfig, forwardingRules []firewallManager.ForwardRule) error {
 	for _, p := range peersUpdate {
-		err := e.addNewPeer(p)
+		err := e.addNewPeer(p, forwardingRules)
 		if err != nil {
 			return err
 		}
@@ -1830,8 +1831,9 @@ func (e *Engine) addNewPeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
 	return nil
 }
 
-// addNewPeer add peer if connection doesn't exist
-func (e *Engine) addNewPeer(peerConfig *mgmProto.RemotePeerConfig) error {
+// addNewPeer add peer if connection doesn't exist. A peer that is not lazy by
+// policy (or is a forwarder) gets an always-active connection instead.
+func (e *Engine) addNewPeer(peerConfig *mgmProto.RemotePeerConfig, forwardingRules []firewallManager.ForwardRule) error {
 	peerKey := peerConfig.GetWgPubKey()
 	peerIPs := make([]netip.Prefix, 0, len(peerConfig.GetAllowedIps()))
 	if _, ok := e.peerStore.PeerConn(peerKey); ok {
@@ -1865,7 +1867,7 @@ func (e *Engine) addNewPeer(peerConfig *mgmProto.RemotePeerConfig) error {
 		log.Warnf("error adding peer %s to status recorder, got error: %v", peerKey, err)
 	}
 
-	if exists := e.connMgr.AddPeerConn(e.ctx, peerKey, conn); exists {
+	if exists := e.connMgr.AddPeerConn(e.ctx, peerKey, conn, e.isPermanentPeer(peerConfig, forwardingRules)); exists {
 		conn.Close(false)
 		return fmt.Errorf("peer already exists: %s", peerKey)
 	}
@@ -2692,25 +2694,37 @@ func (e *Engine) updateForwardRules(rules []*mgmProto.ForwardingRule) ([]firewal
 	return forwardingRules, nberrors.FormatErrorOrNil(merr)
 }
 
+// toExcludedLazyPeers returns the peers that must have an always-active
+// connection, so the caller can reconcile the lazy manager's exclude list.
 func (e *Engine) toExcludedLazyPeers(rules []firewallManager.ForwardRule, peers []*mgmProto.RemotePeerConfig) map[string]bool {
 	excludedPeers := make(map[string]bool)
-
-	// Ingress forward targets: inbound forwarded traffic is initiated remotely and
-	// cannot wake a lazy connection, so the peer routing the target must stay
-	// permanently connected. AllowedIPs are already parsed on the peer conn, so
-	// reuse those typed prefixes instead of re-parsing the network map strings.
-	for _, r := range rules {
-		for _, p := range peers {
-			if e.peerRoutesAddr(p, r.TranslatedAddress) {
-				log.Infof("exclude forwarder peer from lazy connection: %s", p.GetWgPubKey())
-				excludedPeers[p.GetWgPubKey()] = true
-			}
+	for _, p := range peers {
+		if e.isPermanentPeer(p, rules) {
+			excludedPeers[p.GetWgPubKey()] = true
 		}
 	}
-
 	return excludedPeers
 }
 
+// isPermanentPeer reports whether a peer needs an always-active connection: it
+// is not lazy by policy (the per-peer lazy hint or account flag, subject to the
+// local override), or it is an ingress forward target. Inbound forwarded traffic
+// is initiated remotely and cannot wake a lazy connection, so the peer routing
+// the target must stay permanently connected.
+func (e *Engine) isPermanentPeer(p *mgmProto.RemotePeerConfig, rules []firewallManager.ForwardRule) bool {
+	if !e.connMgr.PeerLazyDefault(p.GetLazyState()) {
+		return true
+	}
+
+	for _, r := range rules {
+		if e.peerRoutesAddr(p, r.TranslatedAddress) {
+			log.Infof("exclude forwarder peer from lazy connection: %s", p.GetWgPubKey())
+			return true
+		}
+	}
+	return false
+}
+
 // peerRoutesAddr reports whether the peer is a router for addr, matched against
 // the peer's already-parsed AllowedIPs from the store (the same typed value the
 // lazy manager consumes) rather than re-parsing the network map strings.

client/internal/engine_lazy_exclude_test.go

@@ -49,7 +49,8 @@ func TestToExcludedLazyPeers_ForwardTarget(t *testing.T) {
 	store.AddPeerConn(targetPeerKey, newTestConn(t, targetPeerKey, "100.110.8.145/32"))
 	store.AddPeerConn(otherPeerKey, newTestConn(t, otherPeerKey, "100.110.9.10/32"))
 
-	e := &Engine{peerStore: store}
+	// Lazy on for normal peers, so the only exclusion under test is the forward target.
+	e := &Engine{peerStore: store, connMgr: &ConnMgr{force: lazyForceOn}}
 
 	peers := []*mgmProto.RemotePeerConfig{
 		{WgPubKey: targetPeerKey, AllowedIps: []string{"100.110.8.145/32"}},
@@ -67,7 +68,8 @@ func TestToExcludedLazyPeers_ForwardTarget(t *testing.T) {
 }
 
 func TestToExcludedLazyPeers_NoRules(t *testing.T) {
-	e := &Engine{peerStore: peerstore.NewConnStore()}
+	// Lazy on for normal peers and no forward rules, so nothing is excluded.
+	e := &Engine{peerStore: peerstore.NewConnStore(), connMgr: &ConnMgr{force: lazyForceOn}}
 
 	peers := []*mgmProto.RemotePeerConfig{
 		{WgPubKey: "peer-a", AllowedIps: []string{"100.110.8.145/32"}},