add support for sync message version negotiation

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

management/internals/controllers/network_map/controller/controller.go

@@ -57,11 +57,7 @@ type Controller struct {
 
 	integratedPeerValidator integrated_validator.IntegratedValidator
 
-	// componentsDisabled, when true, forces the controller to emit legacy
-	// proto.NetworkMap to every peer regardless of capability. Set once at
-	// construction and never written after — readers race-free without a
-	// mutex.
-	componentsDisabled bool
+	serverSupportedSyncMessageVersions []grpc.SyncMessageVersion
 }
 
 type bufferUpdate struct {
@@ -95,19 +91,13 @@ func NewController(ctx context.Context, store store.Store, metrics telemetry.App
 		settingsManager:         settingsManager,
 		dnsDomain:               dnsDomain,
 		config:                  config,
-		componentsDisabled:      parseBoolEnv("NB_NETWORK_MAP_COMPONENTS_DISABLE"),
 
-		proxyController:       proxyController,
-		EphemeralPeersManager: ephemeralPeersManager,
+		proxyController:                    proxyController,
+		EphemeralPeersManager:              ephemeralPeersManager,
+		serverSupportedSyncMessageVersions: grpc.SyncMessageVersionsFromString(config.SupportedSyncMessageVersions),
 	}
 }
 
-// PeerNeedsComponents reports whether the gRPC layer should emit the
-// component-based wire format for this peer.
-func (c *Controller) PeerNeedsComponents(p *nbpeer.Peer) bool {
-	return p != nil && p.SupportsComponentNetworkMap() && !c.componentsDisabled
-}
-
 // parseBoolEnv reads an env var via strconv.ParseBool so callers accept the
 // usual "1/t/T/TRUE/true/True" set instead of being strict about a single
 // literal.
@@ -243,26 +233,40 @@ func (c *Controller) sendUpdateAccountPeers(ctx context.Context, accountID strin
 			c.metrics.CountCalcPostureChecksDuration(time.Since(start))
 			start = time.Now()
 
-			result := account.GetPeerNetworkMapResult(ctx, p.ID, c.componentsDisabled, peersCustomZone, accountZones, approvedPeersMap, resourcePolicies, routers, c.accountManagerMetrics, groupIDToUserIDs)
-
-			c.metrics.CountCalcPeerNetworkMapDuration(time.Since(start))
-
-			proxyNetworkMap := proxyNetworkMaps[p.ID]
-			if result.NetworkMap != nil && proxyNetworkMap != nil {
-				result.NetworkMap.Merge(proxyNetworkMap)
-			}
-
 			peerGroups := account.GetPeerGroups(p.ID)
-			start = time.Now()
+			proxyNetworkMap := proxyNetworkMaps[p.ID]
 			var update *proto.SyncResponse
-			if result.IsComponents() {
+
+			commonSyncMessageVersions := grpc.CommonSyncMessageVersions(
+				c.serverSupportedSyncMessageVersions,
+				grpc.SyncMessageVersionsFromProtoEnums(peer.Meta.Capabilities))
+
+			if commonSyncMessageVersions[0] == grpc.ComponentNetworkMap {
+				components := account.GetPeerNetworkMapComponents(
+					ctx, p.ID, peersCustomZone, accountZones, approvedPeersMap, resourcePolicies, routers, groupIDToUserIDs)
+
+				c.metrics.CountCalcPeerNetworkMapDuration(time.Since(start))
+
+				start = time.Now()
 				// proxyNetworkMap rides the envelope as a ProxyPatch sidecar;
 				// the client merges it into Calculate()'s output the same
 				// way the legacy server did via NetworkMap.Merge.
-				update = grpc.ToComponentSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, p, nil, nil, result.Components, proxyNetworkMap, dnsDomain, postureChecks, account.Settings, extraSetting, maps.Keys(peerGroups), dnsFwdPort)
+				update = grpc.ToComponentSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, p, nil, nil, components, proxyNetworkMap, dnsDomain, postureChecks, account.Settings, extraSetting, maps.Keys(peerGroups), dnsFwdPort)
+
 			} else {
-				update = grpc.ToSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, p, nil, nil, result.NetworkMap, dnsDomain, postureChecks, dnsCache, account.Settings, extraSetting, maps.Keys(peerGroups), dnsFwdPort)
+				nmap := account.GetPeerNetworkMapFromComponents(
+					ctx, p.ID, peersCustomZone, accountZones, approvedPeersMap, resourcePolicies, routers, c.accountManagerMetrics, groupIDToUserIDs)
+
+				c.metrics.CountCalcPeerNetworkMapDuration(time.Since(start))
+
+				if proxyNetworkMap != nil {
+					nmap.Merge(proxyNetworkMap)
+				}
+
+				start = time.Now()
+				update = grpc.ToSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, p, nil, nil, nmap, dnsDomain, postureChecks, dnsCache, account.Settings, extraSetting, maps.Keys(peerGroups), dnsFwdPort)
 			}
+
 			c.metrics.CountToSyncResponseDuration(time.Since(start))
 
 			c.peersUpdateManager.SendUpdate(ctx, p.ID, &network_map.UpdateMessage{
@@ -381,27 +385,39 @@ func (c *Controller) sendUpdateForAffectedPeers(ctx context.Context, accountID s
 			c.metrics.CountCalcPostureChecksDuration(time.Since(start))
 			start = time.Now()
 
-			result := account.GetPeerNetworkMapResult(ctx, p.ID, c.componentsDisabled, peersCustomZone, accountZones, approvedPeersMap, resourcePolicies, routers, c.accountManagerMetrics, groupIDToUserIDs)
-
-			c.metrics.CountCalcPeerNetworkMapDuration(time.Since(start))
-
-			proxyNetworkMap := proxyNetworkMaps[p.ID]
-			if result.NetworkMap != nil && proxyNetworkMap != nil {
-				result.NetworkMap.Merge(proxyNetworkMap)
-			}
-
 			peerGroups := account.GetPeerGroups(p.ID)
-			start = time.Now()
+			proxyNetworkMap := proxyNetworkMaps[p.ID]
 			var update *proto.SyncResponse
-			if result.IsComponents() {
+
+			commonSyncMessageVersions := grpc.CommonSyncMessageVersions(
+				c.serverSupportedSyncMessageVersions,
+				grpc.SyncMessageVersionsFromProtoEnums(peer.Meta.Capabilities))
+
+			if commonSyncMessageVersions[0] == grpc.ComponentNetworkMap {
+				components := account.GetPeerNetworkMapComponents(
+					ctx, p.ID, peersCustomZone, accountZones, approvedPeersMap, resourcePolicies, routers, groupIDToUserIDs)
+
+				c.metrics.CountCalcPeerNetworkMapDuration(time.Since(start))
+
+				start = time.Now()
 				// proxyNetworkMap rides the envelope as a ProxyPatch sidecar;
 				// the client merges it into Calculate()'s output the same
 				// way the legacy server did via NetworkMap.Merge.
-				update = grpc.ToComponentSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, p, nil, nil, result.Components, proxyNetworkMap, dnsDomain, postureChecks, account.Settings, extraSetting, maps.Keys(peerGroups), dnsFwdPort)
+				update = grpc.ToComponentSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, p, nil, nil, components, proxyNetworkMap, dnsDomain, postureChecks, account.Settings, extraSetting, maps.Keys(peerGroups), dnsFwdPort)
+
 			} else {
-				update = grpc.ToSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, p, nil, nil, result.NetworkMap, dnsDomain, postureChecks, dnsCache, account.Settings, extraSetting, maps.Keys(peerGroups), dnsFwdPort)
+				nmap := account.GetPeerNetworkMapFromComponents(
+					ctx, p.ID, peersCustomZone, accountZones, approvedPeersMap, resourcePolicies, routers, c.accountManagerMetrics, groupIDToUserIDs)
+
+				c.metrics.CountCalcPeerNetworkMapDuration(time.Since(start))
+
+				if proxyNetworkMap != nil {
+					nmap.Merge(proxyNetworkMap)
+				}
+
+				start = time.Now()
+				update = grpc.ToSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, p, nil, nil, nmap, dnsDomain, postureChecks, dnsCache, account.Settings, extraSetting, maps.Keys(peerGroups), dnsFwdPort)
 			}
-			c.metrics.CountToSyncResponseDuration(time.Since(start))
 
 			c.peersUpdateManager.SendUpdate(ctx, p.ID, &network_map.UpdateMessage{
 				Update:      update,
@@ -488,13 +504,7 @@ func (c *Controller) UpdateAccountPeer(ctx context.Context, accountId string, pe
 		return err
 	}
 
-	result := account.GetPeerNetworkMapResult(ctx, peerId, c.componentsDisabled, peersCustomZone, accountZones, approvedPeersMap, resourcePolicies, routers, c.accountManagerMetrics, groupIDToUserIDs)
-
 	proxyNetworkMap := proxyNetworkMaps[peer.ID]
-	if result.NetworkMap != nil && proxyNetworkMap != nil {
-		result.NetworkMap.Merge(proxyNetworkMap)
-	}
-
 	extraSettings, err := c.settingsManager.GetExtraSettings(ctx, peer.AccountID)
 	if err != nil {
 		return fmt.Errorf("failed to get extra settings: %v", err)
@@ -504,11 +514,30 @@ func (c *Controller) UpdateAccountPeer(ctx context.Context, accountId string, pe
 	dnsFwdPort := computeForwarderPort(maps.Values(account.Peers), network_map.DnsForwarderPortMinVersion)
 
 	var update *proto.SyncResponse
-	if result.IsComponents() {
-		update = grpc.ToComponentSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, peer, nil, nil, result.Components, proxyNetworkMap, dnsDomain, postureChecks, account.Settings, extraSettings, maps.Keys(peerGroups), dnsFwdPort)
+
+	commonSyncMessageVersions := grpc.CommonSyncMessageVersions(
+		c.serverSupportedSyncMessageVersions,
+		grpc.SyncMessageVersionsFromProtoEnums(peer.Meta.Capabilities))
+
+	if commonSyncMessageVersions[0] == grpc.ComponentNetworkMap {
+		components := account.GetPeerNetworkMapComponents(
+			ctx, peer.ID, peersCustomZone, accountZones, approvedPeersMap, resourcePolicies, routers, groupIDToUserIDs)
+
+		// proxyNetworkMap rides the envelope as a ProxyPatch sidecar;
+		// the client merges it into Calculate()'s output the same
+		// way the legacy server did via NetworkMap.Merge.
+		update = grpc.ToComponentSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, peer, nil, nil, components, proxyNetworkMap, dnsDomain, postureChecks, account.Settings, extraSettings, maps.Keys(peerGroups), dnsFwdPort)
 	} else {
-		update = grpc.ToSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, peer, nil, nil, result.NetworkMap, dnsDomain, postureChecks, dnsCache, account.Settings, extraSettings, maps.Keys(peerGroups), dnsFwdPort)
+		nmap := account.GetPeerNetworkMapFromComponents(
+			ctx, peer.ID, peersCustomZone, accountZones, approvedPeersMap, resourcePolicies, routers, c.accountManagerMetrics, groupIDToUserIDs)
+
+		if proxyNetworkMap != nil {
+			nmap.Merge(proxyNetworkMap)
+		}
+
+		update = grpc.ToSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, peer, nil, nil, nmap, dnsDomain, postureChecks, dnsCache, account.Settings, extraSettings, maps.Keys(peerGroups), dnsFwdPort)
 	}
+
 	c.peersUpdateManager.SendUpdate(ctx, peer.ID, &network_map.UpdateMessage{
 		Update:      update,
 		MessageType: network_map.MessageTypeNetworkMap,

management/internals/controllers/network_map/interface.go

@@ -25,9 +25,6 @@ type Controller interface {
 	BufferUpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error
 	GetValidatedPeerWithMap(ctx context.Context, isRequiresApproval bool, accountID string, peerID string) (*types.NetworkMap, []*posture.Checks, int64, error)
 	GetValidatedPeerWithComponents(ctx context.Context, isRequiresApproval bool, accountID string, p *nbpeer.Peer) (*nbpeer.Peer, *types.NetworkMapComponents, *types.NetworkMap, []*posture.Checks, int64, error)
-	// PeerNeedsComponents combines the peer's advertised capability with the
-	// kill-switch flag — the only public predicate gRPC layers should ask.
-	PeerNeedsComponents(p *nbpeer.Peer) bool
 	GetDNSDomain(settings *types.Settings) string
 	StartWarmup(context.Context)
 	GetNetworkMap(ctx context.Context, peerID string) (*types.NetworkMap, error)

management/internals/controllers/network_map/interface_mock.go

@@ -1,9 +1,9 @@
 // Code generated by MockGen. DO NOT EDIT.
-// Source: management/internals/controllers/network_map/interface.go
+// Source: ./interface.go
 //
 // Generated by this command:
 //
-//	mockgen -package network_map -destination=management/internals/controllers/network_map/interface_mock.go -source=management/internals/controllers/network_map/interface.go -build_flags=-mod=mod
+//	mockgen -package network_map -destination=interface_mock.go -source=./interface.go -build_flags=-mod=mod
 //
 
 // Package network_map is a generated GoMock package.
@@ -126,23 +126,6 @@ func (mr *MockControllerMockRecorder) GetNetworkMap(ctx, peerID any) *gomock.Cal
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNetworkMap", reflect.TypeOf((*MockController)(nil).GetNetworkMap), ctx, peerID)
 }
 
-// GetValidatedPeerWithMap mocks base method.
-func (m *MockController) GetValidatedPeerWithMap(ctx context.Context, isRequiresApproval bool, accountID string, peerID string) (*types.NetworkMap, []*posture.Checks, int64, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetValidatedPeerWithMap", ctx, isRequiresApproval, accountID, peerID)
-	ret0, _ := ret[0].(*types.NetworkMap)
-	ret1, _ := ret[1].([]*posture.Checks)
-	ret2, _ := ret[2].(int64)
-	ret3, _ := ret[3].(error)
-	return ret0, ret1, ret2, ret3
-}
-
-// GetValidatedPeerWithMap indicates an expected call of GetValidatedPeerWithMap.
-func (mr *MockControllerMockRecorder) GetValidatedPeerWithMap(ctx, isRequiresApproval, accountID, peerID any) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetValidatedPeerWithMap", reflect.TypeOf((*MockController)(nil).GetValidatedPeerWithMap), ctx, isRequiresApproval, accountID, peerID)
-}
-
 // GetValidatedPeerWithComponents mocks base method.
 func (m *MockController) GetValidatedPeerWithComponents(ctx context.Context, isRequiresApproval bool, accountID string, p *peer.Peer) (*peer.Peer, *types.NetworkMapComponents, *types.NetworkMap, []*posture.Checks, int64, error) {
 	m.ctrl.T.Helper()
@@ -162,18 +145,21 @@ func (mr *MockControllerMockRecorder) GetValidatedPeerWithComponents(ctx, isRequ
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetValidatedPeerWithComponents", reflect.TypeOf((*MockController)(nil).GetValidatedPeerWithComponents), ctx, isRequiresApproval, accountID, p)
 }
 
-// PeerNeedsComponents mocks base method.
-func (m *MockController) PeerNeedsComponents(p *peer.Peer) bool {
+// GetValidatedPeerWithMap mocks base method.
+func (m *MockController) GetValidatedPeerWithMap(ctx context.Context, isRequiresApproval bool, accountID, peerID string) (*types.NetworkMap, []*posture.Checks, int64, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "PeerNeedsComponents", p)
-	ret0, _ := ret[0].(bool)
-	return ret0
+	ret := m.ctrl.Call(m, "GetValidatedPeerWithMap", ctx, isRequiresApproval, accountID, peerID)
+	ret0, _ := ret[0].(*types.NetworkMap)
+	ret1, _ := ret[1].([]*posture.Checks)
+	ret2, _ := ret[2].(int64)
+	ret3, _ := ret[3].(error)
+	return ret0, ret1, ret2, ret3
 }
 
-// PeerNeedsComponents indicates an expected call of PeerNeedsComponents.
-func (mr *MockControllerMockRecorder) PeerNeedsComponents(p any) *gomock.Call {
+// GetValidatedPeerWithMap indicates an expected call of GetValidatedPeerWithMap.
+func (mr *MockControllerMockRecorder) GetValidatedPeerWithMap(ctx, isRequiresApproval, accountID, peerID any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PeerNeedsComponents", reflect.TypeOf((*MockController)(nil).PeerNeedsComponents), p)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetValidatedPeerWithMap", reflect.TypeOf((*MockController)(nil).GetValidatedPeerWithMap), ctx, isRequiresApproval, accountID, peerID)
 }
 
 // OnPeerConnected mocks base method.
@@ -204,7 +190,7 @@ func (mr *MockControllerMockRecorder) OnPeerDisconnected(ctx, accountID, peerID
 }
 
 // OnPeersAdded mocks base method.
-func (m *MockController) OnPeersAdded(ctx context.Context, accountID string, peerIDs []string, affectedPeerIDs []string) error {
+func (m *MockController) OnPeersAdded(ctx context.Context, accountID string, peerIDs, affectedPeerIDs []string) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "OnPeersAdded", ctx, accountID, peerIDs, affectedPeerIDs)
 	ret0, _ := ret[0].(error)
@@ -218,7 +204,7 @@ func (mr *MockControllerMockRecorder) OnPeersAdded(ctx, accountID, peerIDs, affe
 }
 
 // OnPeersDeleted mocks base method.
-func (m *MockController) OnPeersDeleted(ctx context.Context, accountID string, peerIDs []string, affectedPeerIDs []string) error {
+func (m *MockController) OnPeersDeleted(ctx context.Context, accountID string, peerIDs, affectedPeerIDs []string) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "OnPeersDeleted", ctx, accountID, peerIDs, affectedPeerIDs)
 	ret0, _ := ret[0].(error)
@@ -232,7 +218,7 @@ func (mr *MockControllerMockRecorder) OnPeersDeleted(ctx, accountID, peerIDs, af
 }
 
 // OnPeersUpdated mocks base method.
-func (m *MockController) OnPeersUpdated(ctx context.Context, accountId string, peerIDs []string, affectedPeerIDs []string) error {
+func (m *MockController) OnPeersUpdated(ctx context.Context, accountId string, peerIDs, affectedPeerIDs []string) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "OnPeersUpdated", ctx, accountId, peerIDs, affectedPeerIDs)
 	ret0, _ := ret[0].(error)