Merge branch 'main' into refactor/mgmt-bootstrap

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

@@ -45,7 +45,7 @@ type Controller struct {
 	EphemeralPeersManager ephemeral.Manager
 
 	accountUpdateLocks               sync.Map
-	sendAccountUpdateLocks           sync.Map
+	affectedPeerUpdateLocks          sync.Map
 	updateAccountPeersBufferInterval atomic.Int64
 	// dnsDomain is used for peer resolution. This is appended to the peer's name
 	dnsDomain string
@@ -64,6 +64,13 @@ type bufferUpdate struct {
 	update atomic.Bool
 }
 
+type bufferAffectedUpdate struct {
+	sendMu  sync.Mutex
+	dataMu  sync.Mutex
+	next    *time.Timer
+	peerIDs map[string]struct{}
+}
+
 var _ network_map.Controller = (*Controller)(nil)
 
 func NewController(ctx context.Context, store store.Store, metrics telemetry.AppMetrics, peersUpdateManager network_map.PeersUpdateManager, requestBuffer account.RequestBuffer, integratedPeerValidator integrated_validator.IntegratedValidator, settingsManager settings.Manager, dnsDomain string, proxyController port_forwarding.Controller, ephemeralPeersManager ephemeral.Manager, config *config.Config) *Controller {
@@ -201,7 +208,7 @@ func (c *Controller) sendUpdateAccountPeers(ctx context.Context, accountID strin
 
 			c.metrics.CountCalcPeerNetworkMapDuration(time.Since(start))
 
-			proxyNetworkMap, ok := proxyNetworkMaps[peer.ID]
+			proxyNetworkMap, ok := proxyNetworkMaps[p.ID]
 			if ok {
 				remotePeerNetworkMap.Merge(proxyNetworkMap)
 			}
@@ -226,44 +233,6 @@ func (c *Controller) sendUpdateAccountPeers(ctx context.Context, accountID strin
 	return nil
 }
 
-func (c *Controller) bufferSendUpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error {
-	log.WithContext(ctx).Tracef("buffer sending update peers for account %s from %s", accountID, util.GetCallerName())
-
-	if c.accountManagerMetrics != nil {
-		c.accountManagerMetrics.CountUpdateAccountPeersTriggered(string(reason.Resource), string(reason.Operation))
-	}
-
-	bufUpd, _ := c.sendAccountUpdateLocks.LoadOrStore(accountID, &bufferUpdate{})
-	b := bufUpd.(*bufferUpdate)
-
-	if !b.mu.TryLock() {
-		b.update.Store(true)
-		return nil
-	}
-
-	if b.next != nil {
-		b.next.Stop()
-	}
-
-	go func() {
-		defer b.mu.Unlock()
-		_ = c.sendUpdateAccountPeers(ctx, accountID, reason)
-		if !b.update.Load() {
-			return
-		}
-		b.update.Store(false)
-		if b.next == nil {
-			b.next = time.AfterFunc(time.Duration(c.updateAccountPeersBufferInterval.Load()), func() {
-				_ = c.sendUpdateAccountPeers(ctx, accountID, reason)
-			})
-			return
-		}
-		b.next.Reset(time.Duration(c.updateAccountPeersBufferInterval.Load()))
-	}()
-
-	return nil
-}
-
 // UpdatePeers updates all peers that belong to an account.
 // Should be called when changes have to be synced to peers.
 func (c *Controller) UpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error {
@@ -273,6 +242,143 @@ func (c *Controller) UpdateAccountPeers(ctx context.Context, accountID string, r
 	return c.sendUpdateAccountPeers(ctx, accountID, reason)
 }
 
+// UpdateAffectedPeers updates only the specified peers that belong to an account.
+func (c *Controller) UpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string) error {
+	if len(peerIDs) == 0 {
+		return nil
+	}
+	return c.sendUpdateForAffectedPeers(ctx, accountID, peerIDs)
+}
+
+func (c *Controller) sendUpdateForAffectedPeers(ctx context.Context, accountID string, peerIDs []string) error {
+	log.WithContext(ctx).Tracef("sendUpdateForAffectedPeers: account %s, %d affected peers: %v (caller: %s)", accountID, len(peerIDs), peerIDs, util.GetCallerName())
+
+	if !c.hasConnectedPeers(peerIDs) {
+		log.WithContext(ctx).Tracef("sendUpdateForAffectedPeers: no connected peers among %v, skipping", peerIDs)
+		return nil
+	}
+
+	account, err := c.requestBuffer.GetAccountWithBackpressure(ctx, accountID)
+	if err != nil {
+		return fmt.Errorf("failed to get account: %v", err)
+	}
+
+	globalStart := time.Now()
+
+	peersToUpdate := c.filterConnectedAffectedPeers(account, peerIDs)
+	if len(peersToUpdate) == 0 {
+		log.WithContext(ctx).Tracef("sendUpdateForAffectedPeers: no peers to update (affected peers not found in account or no channels)")
+		return nil
+	}
+
+	log.WithContext(ctx).Tracef("sendUpdateForAffectedPeers: sending network map to %d connected peers", len(peersToUpdate))
+
+	approvedPeersMap, err := c.integratedPeerValidator.GetValidatedPeers(ctx, account.Id, maps.Values(account.Groups), maps.Values(account.Peers), account.Settings.Extra)
+	if err != nil {
+		return fmt.Errorf("failed to get validate peers: %v", err)
+	}
+
+	var wg sync.WaitGroup
+	semaphore := make(chan struct{}, 10)
+
+	account.InjectProxyPolicies(ctx)
+	dnsCache := &cache.DNSConfigCache{}
+	dnsDomain := c.GetDNSDomain(account.Settings)
+	peersCustomZone := account.GetPeersCustomZone(ctx, dnsDomain)
+	resourcePolicies := account.GetResourcePoliciesMap()
+	routers := account.GetResourceRoutersMap()
+	groupIDToUserIDs := account.GetActiveGroupUsers()
+
+	proxyNetworkMaps, err := c.proxyController.GetProxyNetworkMapsAll(ctx, accountID, account.Peers)
+	if err != nil {
+		log.WithContext(ctx).Errorf("failed to get proxy network maps: %v", err)
+		return fmt.Errorf("failed to get proxy network maps: %v", err)
+	}
+
+	extraSetting, err := c.settingsManager.GetExtraSettings(ctx, accountID)
+	if err != nil {
+		return fmt.Errorf("failed to get flow enabled status: %v", err)
+	}
+
+	dnsFwdPort := computeForwarderPort(maps.Values(account.Peers), network_map.DnsForwarderPortMinVersion)
+
+	accountZones, err := c.repo.GetAccountZones(ctx, account.Id)
+	if err != nil {
+		log.WithContext(ctx).Errorf("failed to get account zones: %v", err)
+		return fmt.Errorf("failed to get account zones: %v", err)
+	}
+
+	for _, peer := range peersToUpdate {
+		wg.Add(1)
+		semaphore <- struct{}{}
+		go func(p *nbpeer.Peer) {
+			defer wg.Done()
+			defer func() { <-semaphore }()
+
+			start := time.Now()
+
+			postureChecks, err := c.getPeerPostureChecks(account, p.ID)
+			if err != nil {
+				log.WithContext(ctx).Debugf("failed to get posture checks for peer %s: %v", p.ID, err)
+				return
+			}
+
+			c.metrics.CountCalcPostureChecksDuration(time.Since(start))
+			start = time.Now()
+
+			remotePeerNetworkMap := account.GetPeerNetworkMapFromComponents(ctx, p.ID, peersCustomZone, accountZones, approvedPeersMap, resourcePolicies, routers, c.accountManagerMetrics, groupIDToUserIDs)
+
+			c.metrics.CountCalcPeerNetworkMapDuration(time.Since(start))
+
+			proxyNetworkMap, ok := proxyNetworkMaps[p.ID]
+			if ok {
+				remotePeerNetworkMap.Merge(proxyNetworkMap)
+			}
+
+			peerGroups := account.GetPeerGroups(p.ID)
+			start = time.Now()
+			update := grpc.ToSyncResponse(ctx, nil, c.config.HttpConfig, c.config.DeviceAuthorizationFlow, p, nil, nil, remotePeerNetworkMap, 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,
+				MessageType: network_map.MessageTypeNetworkMap,
+			})
+		}(peer)
+	}
+
+	wg.Wait()
+	if c.accountManagerMetrics != nil {
+		c.accountManagerMetrics.CountUpdateAccountPeersDuration(time.Since(globalStart))
+	}
+
+	return nil
+}
+
+func (c *Controller) hasConnectedPeers(peerIDs []string) bool {
+	for _, id := range peerIDs {
+		if c.peersUpdateManager.HasChannel(id) {
+			return true
+		}
+	}
+	return false
+}
+
+func (c *Controller) filterConnectedAffectedPeers(account *types.Account, peerIDs []string) []*nbpeer.Peer {
+	affected := make(map[string]struct{}, len(peerIDs))
+	for _, id := range peerIDs {
+		affected[id] = struct{}{}
+	}
+
+	var result []*nbpeer.Peer
+	for _, peer := range account.Peers {
+		if _, ok := affected[peer.ID]; ok && c.peersUpdateManager.HasChannel(peer.ID) {
+			result = append(result, peer)
+		}
+	}
+	return result
+}
+
 func (c *Controller) UpdateAccountPeer(ctx context.Context, accountId string, peerId string) error {
 	if !c.peersUpdateManager.HasChannel(peerId) {
 		return fmt.Errorf("peer %s doesn't have a channel, skipping network map update", peerId)
@@ -381,6 +487,104 @@ func (c *Controller) BufferUpdateAccountPeers(ctx context.Context, accountID str
 	return nil
 }
 
+// BufferUpdateAffectedPeers accumulates peer IDs and flushes them after the buffer interval.
+func (c *Controller) BufferUpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string, reason types.UpdateReason) error {
+	if len(peerIDs) == 0 {
+		return nil
+	}
+
+	if c.accountManagerMetrics != nil {
+		c.accountManagerMetrics.CountUpdateAccountPeersTriggered(string(reason.Resource), string(reason.Operation))
+	}
+
+	log.WithContext(ctx).Tracef("buffer updating %d affected peers for account %s from %s", len(peerIDs), accountID, util.GetCallerName())
+
+	bufUpd, _ := c.affectedPeerUpdateLocks.LoadOrStore(accountID, &bufferAffectedUpdate{
+		peerIDs: make(map[string]struct{}),
+	})
+	b := bufUpd.(*bufferAffectedUpdate)
+
+	b.addPeerIDs(peerIDs)
+
+	if !b.sendMu.TryLock() {
+		// Another goroutine is already sending; it will pick up our IDs on its next drain.
+		return nil
+	}
+
+	b.stopTimer()
+
+	// The send and the debounced timer outlive the calling request, so detach from
+	// its context to avoid sending with a cancelled context once the handler returns.
+	bgCtx := context.WithoutCancel(ctx)
+
+	collected := b.drainPeerIDs()
+	go func() {
+		defer b.sendMu.Unlock()
+		_ = c.sendUpdateForAffectedPeers(bgCtx, accountID, collected)
+
+		// Check if more peer IDs accumulated while we were sending.
+		if !b.hasPending() {
+			return
+		}
+
+		// Schedule a debounced flush for the newly accumulated IDs.
+		b.setTimer(time.Duration(c.updateAccountPeersBufferInterval.Load()), func() {
+			ids := b.drainPeerIDs()
+			if len(ids) > 0 {
+				_ = c.sendUpdateForAffectedPeers(bgCtx, accountID, ids)
+			}
+		})
+	}()
+
+	return nil
+}
+
+func (b *bufferAffectedUpdate) addPeerIDs(ids []string) {
+	b.dataMu.Lock()
+	for _, id := range ids {
+		b.peerIDs[id] = struct{}{}
+	}
+	b.dataMu.Unlock()
+}
+
+func (b *bufferAffectedUpdate) drainPeerIDs() []string {
+	b.dataMu.Lock()
+	defer b.dataMu.Unlock()
+	if len(b.peerIDs) == 0 {
+		return nil
+	}
+	ids := make([]string, 0, len(b.peerIDs))
+	for id := range b.peerIDs {
+		ids = append(ids, id)
+	}
+	b.peerIDs = make(map[string]struct{})
+	return ids
+}
+
+func (b *bufferAffectedUpdate) hasPending() bool {
+	b.dataMu.Lock()
+	defer b.dataMu.Unlock()
+	return len(b.peerIDs) > 0
+}
+
+func (b *bufferAffectedUpdate) stopTimer() {
+	b.dataMu.Lock()
+	defer b.dataMu.Unlock()
+	if b.next != nil {
+		b.next.Stop()
+	}
+}
+
+func (b *bufferAffectedUpdate) setTimer(d time.Duration, f func()) {
+	b.dataMu.Lock()
+	defer b.dataMu.Unlock()
+	if b.next == nil {
+		b.next = time.AfterFunc(d, f)
+		return
+	}
+	b.next.Reset(d)
+}
+
 func (c *Controller) GetValidatedPeerWithMap(ctx context.Context, isRequiresApproval bool, accountID string, peer *nbpeer.Peer) (*nbpeer.Peer, *types.NetworkMap, []*posture.Checks, int64, error) {
 	if isRequiresApproval {
 		network, err := c.repo.GetAccountNetwork(ctx, accountID)
@@ -578,21 +782,24 @@ func isPeerInPolicySourceGroups(account *types.Account, peerID string, policy *t
 	return false, nil
 }
 
-func (c *Controller) OnPeersUpdated(ctx context.Context, accountID string, peerIDs []string) error {
-	err := c.bufferSendUpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationUpdate})
-	if err != nil {
-		log.WithContext(ctx).Errorf("failed to buffer update account peers for peer update in account %s: %v", accountID, err)
+func (c *Controller) OnPeersUpdated(ctx context.Context, accountID string, peerIDs []string, affectedPeerIDs []string) error {
+	if len(affectedPeerIDs) == 0 {
+		log.WithContext(ctx).Tracef("no affected peers for peer update in account %s, skipping", accountID)
+		return nil
 	}
-
-	return nil
+	return c.BufferUpdateAffectedPeers(ctx, accountID, affectedPeerIDs, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationUpdate})
 }
 
-func (c *Controller) OnPeersAdded(ctx context.Context, accountID string, peerIDs []string) error {
+func (c *Controller) OnPeersAdded(ctx context.Context, accountID string, peerIDs []string, affectedPeerIDs []string) error {
 	log.WithContext(ctx).Debugf("OnPeersAdded call to add peers: %v", peerIDs)
-	return c.bufferSendUpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationCreate})
+	if len(affectedPeerIDs) == 0 {
+		log.WithContext(ctx).Tracef("no affected peers for peer add in account %s, skipping", accountID)
+		return nil
+	}
+	return c.BufferUpdateAffectedPeers(ctx, accountID, affectedPeerIDs, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationCreate})
 }
 
-func (c *Controller) OnPeersDeleted(ctx context.Context, accountID string, peerIDs []string) error {
+func (c *Controller) OnPeersDeleted(ctx context.Context, accountID string, peerIDs []string, affectedPeerIDs []string) error {
 	network, err := c.repo.GetAccountNetwork(ctx, accountID)
 	if err != nil {
 		return err
@@ -625,7 +832,11 @@ func (c *Controller) OnPeersDeleted(ctx context.Context, accountID string, peerI
 		c.peersUpdateManager.CloseChannel(ctx, peerID)
 	}
 
-	return c.bufferSendUpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationDelete})
+	if len(affectedPeerIDs) == 0 {
+		log.WithContext(ctx).Tracef("no affected peers for peer delete in account %s, skipping", accountID)
+		return nil
+	}
+	return c.BufferUpdateAffectedPeers(ctx, accountID, affectedPeerIDs, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationDelete})
 }
 
 // GetNetworkMap returns Network map for a given peer (omits original peer from the Peers result)

management/internals/controllers/network_map/interface.go

@@ -19,6 +19,8 @@ const (
 
 type Controller interface {
 	UpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error
+	UpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string) error
+	BufferUpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string, reason types.UpdateReason) error
 	UpdateAccountPeer(ctx context.Context, accountId string, peerId string) error
 	BufferUpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error
 	GetValidatedPeerWithMap(ctx context.Context, isRequiresApproval bool, accountID string, p *nbpeer.Peer) (*nbpeer.Peer, *types.NetworkMap, []*posture.Checks, int64, error)
@@ -27,9 +29,9 @@ type Controller interface {
 	GetNetworkMap(ctx context.Context, peerID string) (*types.NetworkMap, error)
 	CountStreams() int
 
-	OnPeersUpdated(ctx context.Context, accountId string, peerIDs []string) error
-	OnPeersAdded(ctx context.Context, accountID string, peerIDs []string) error
-	OnPeersDeleted(ctx context.Context, accountID string, peerIDs []string) error
+	OnPeersUpdated(ctx context.Context, accountId string, peerIDs []string, affectedPeerIDs []string) error
+	OnPeersAdded(ctx context.Context, accountID string, peerIDs []string, affectedPeerIDs []string) error
+	OnPeersDeleted(ctx context.Context, accountID string, peerIDs []string, affectedPeerIDs []string) error
 	DisconnectPeers(ctx context.Context, accountId string, peerIDs []string)
 	OnPeerConnected(ctx context.Context, accountID string, peerID string) (chan *UpdateMessage, error)
 	OnPeerDisconnected(ctx context.Context, accountID string, peerID string)

management/internals/controllers/network_map/interface_mock.go

@@ -57,6 +57,20 @@ func (mr *MockControllerMockRecorder) BufferUpdateAccountPeers(ctx, accountID, r
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BufferUpdateAccountPeers", reflect.TypeOf((*MockController)(nil).BufferUpdateAccountPeers), ctx, accountID, reason)
 }
 
+// BufferUpdateAffectedPeers mocks base method.
+func (m *MockController) BufferUpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string, reason types.UpdateReason) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "BufferUpdateAffectedPeers", ctx, accountID, peerIDs, reason)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// BufferUpdateAffectedPeers indicates an expected call of BufferUpdateAffectedPeers.
+func (mr *MockControllerMockRecorder) BufferUpdateAffectedPeers(ctx, accountID, peerIDs, reason any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BufferUpdateAffectedPeers", reflect.TypeOf((*MockController)(nil).BufferUpdateAffectedPeers), ctx, accountID, peerIDs, reason)
+}
+
 // CountStreams mocks base method.
 func (m *MockController) CountStreams() int {
 	m.ctrl.T.Helper()
@@ -158,45 +172,45 @@ func (mr *MockControllerMockRecorder) OnPeerDisconnected(ctx, accountID, peerID
 }
 
 // OnPeersAdded mocks base method.
-func (m *MockController) OnPeersAdded(ctx context.Context, accountID string, peerIDs []string) error {
+func (m *MockController) OnPeersAdded(ctx context.Context, accountID string, peerIDs []string, affectedPeerIDs []string) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "OnPeersAdded", ctx, accountID, peerIDs)
+	ret := m.ctrl.Call(m, "OnPeersAdded", ctx, accountID, peerIDs, affectedPeerIDs)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // OnPeersAdded indicates an expected call of OnPeersAdded.
-func (mr *MockControllerMockRecorder) OnPeersAdded(ctx, accountID, peerIDs any) *gomock.Call {
+func (mr *MockControllerMockRecorder) OnPeersAdded(ctx, accountID, peerIDs, affectedPeerIDs any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnPeersAdded", reflect.TypeOf((*MockController)(nil).OnPeersAdded), ctx, accountID, peerIDs)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnPeersAdded", reflect.TypeOf((*MockController)(nil).OnPeersAdded), ctx, accountID, peerIDs, affectedPeerIDs)
 }
 
 // OnPeersDeleted mocks base method.
-func (m *MockController) OnPeersDeleted(ctx context.Context, accountID string, peerIDs []string) error {
+func (m *MockController) OnPeersDeleted(ctx context.Context, accountID string, peerIDs []string, affectedPeerIDs []string) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "OnPeersDeleted", ctx, accountID, peerIDs)
+	ret := m.ctrl.Call(m, "OnPeersDeleted", ctx, accountID, peerIDs, affectedPeerIDs)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // OnPeersDeleted indicates an expected call of OnPeersDeleted.
-func (mr *MockControllerMockRecorder) OnPeersDeleted(ctx, accountID, peerIDs any) *gomock.Call {
+func (mr *MockControllerMockRecorder) OnPeersDeleted(ctx, accountID, peerIDs, affectedPeerIDs any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnPeersDeleted", reflect.TypeOf((*MockController)(nil).OnPeersDeleted), ctx, accountID, peerIDs)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnPeersDeleted", reflect.TypeOf((*MockController)(nil).OnPeersDeleted), ctx, accountID, peerIDs, affectedPeerIDs)
 }
 
 // OnPeersUpdated mocks base method.
-func (m *MockController) OnPeersUpdated(ctx context.Context, accountId string, peerIDs []string) error {
+func (m *MockController) OnPeersUpdated(ctx context.Context, accountId string, peerIDs []string, affectedPeerIDs []string) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "OnPeersUpdated", ctx, accountId, peerIDs)
+	ret := m.ctrl.Call(m, "OnPeersUpdated", ctx, accountId, peerIDs, affectedPeerIDs)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // OnPeersUpdated indicates an expected call of OnPeersUpdated.
-func (mr *MockControllerMockRecorder) OnPeersUpdated(ctx, accountId, peerIDs any) *gomock.Call {
+func (mr *MockControllerMockRecorder) OnPeersUpdated(ctx, accountId, peerIDs, affectedPeerIDs any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnPeersUpdated", reflect.TypeOf((*MockController)(nil).OnPeersUpdated), ctx, accountId, peerIDs)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnPeersUpdated", reflect.TypeOf((*MockController)(nil).OnPeersUpdated), ctx, accountId, peerIDs, affectedPeerIDs)
 }
 
 // StartWarmup mocks base method.
@@ -250,3 +264,17 @@ func (mr *MockControllerMockRecorder) UpdateAccountPeers(ctx, accountID, reason
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateAccountPeers", reflect.TypeOf((*MockController)(nil).UpdateAccountPeers), ctx, accountID, reason)
 }
+
+// UpdateAffectedPeers mocks base method.
+func (m *MockController) UpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "UpdateAffectedPeers", ctx, accountID, peerIDs)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// UpdateAffectedPeers indicates an expected call of UpdateAffectedPeers.
+func (mr *MockControllerMockRecorder) UpdateAffectedPeers(ctx, accountID, peerIDs any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateAffectedPeers", reflect.TypeOf((*MockController)(nil).UpdateAffectedPeers), ctx, accountID, peerIDs)
+}

management/internals/modules/reverseproxy/service/manager/l4_port_test.go

@@ -488,6 +488,195 @@ func TestUpdate_AllowsPortChange(t *testing.T) {
 	assert.Equal(t, uint16(54321), updated.ListenPort, "explicit port change should be applied")
 }
 
+func TestUpdate_PreservesPortWhenCustomPortsNotSupported(t *testing.T) {
+	mgr, testStore, _ := setupL4Test(t, boolPtr(false))
+	ctx := context.Background()
+
+	existing := seedService(t, testStore, "tcp-svc", "tcp", testCluster, testCluster, 12345)
+
+	updated := &rpservice.Service{
+		ID:           existing.ID,
+		AccountID:    testAccountID,
+		Name:         "tcp-svc-renamed",
+		Mode:         "tcp",
+		Domain:       testCluster,
+		ProxyCluster: testCluster,
+		ListenPort:   0,
+		Enabled:      true,
+		Targets: []*rpservice.Target{
+			{AccountID: testAccountID, TargetId: testPeerID, TargetType: rpservice.TargetTypePeer, Protocol: "tcp", Port: 9090, Enabled: true},
+		},
+	}
+
+	_, err := mgr.persistServiceUpdate(ctx, testAccountID, updated)
+	require.NoError(t, err, "update must not be rejected by the custom-port capability check")
+	assert.Equal(t, uint16(12345), updated.ListenPort, "existing listen port should be preserved on unsupported cluster")
+}
+
+func TestUpdate_PreservesPortWhenCustomPortsUnknown(t *testing.T) {
+	mgr, testStore, _ := setupL4Test(t, nil)
+	ctx := context.Background()
+
+	existing := seedService(t, testStore, "tcp-svc", "tcp", testCluster, testCluster, 12345)
+
+	updated := &rpservice.Service{
+		ID:           existing.ID,
+		AccountID:    testAccountID,
+		Name:         "tcp-svc-renamed",
+		Mode:         "tcp",
+		Domain:       testCluster,
+		ProxyCluster: testCluster,
+		ListenPort:   0,
+		Enabled:      true,
+		Targets: []*rpservice.Target{
+			{AccountID: testAccountID, TargetId: testPeerID, TargetType: rpservice.TargetTypePeer, Protocol: "tcp", Port: 9090, Enabled: true},
+		},
+	}
+
+	_, err := mgr.persistServiceUpdate(ctx, testAccountID, updated)
+	require.NoError(t, err, "update must not be rejected when cluster capability is unknown")
+	assert.Equal(t, uint16(12345), updated.ListenPort, "existing listen port should be preserved when capability is unknown")
+}
+
+func TestUpdate_RejectsPortChangeWhenCustomPortsNotSupported(t *testing.T) {
+	mgr, testStore, _ := setupL4Test(t, boolPtr(false))
+	ctx := context.Background()
+
+	existing := seedService(t, testStore, "tcp-svc", "tcp", testCluster, testCluster, 12345)
+
+	updated := &rpservice.Service{
+		ID:           existing.ID,
+		AccountID:    testAccountID,
+		Name:         "tcp-svc",
+		Mode:         "tcp",
+		Domain:       testCluster,
+		ProxyCluster: testCluster,
+		ListenPort:   54321,
+		Enabled:      true,
+		Targets: []*rpservice.Target{
+			{AccountID: testAccountID, TargetId: testPeerID, TargetType: rpservice.TargetTypePeer, Protocol: "tcp", Port: 9090, Enabled: true},
+		},
+	}
+
+	_, err := mgr.persistServiceUpdate(ctx, testAccountID, updated)
+	require.Error(t, err, "explicit port change on update must be rejected on unsupported clusters")
+	assert.Contains(t, err.Error(), "custom ports not supported on target cluster")
+}
+
+func TestUpdate_TLSPortChangeAllowedWhenNotSupported(t *testing.T) {
+	mgr, testStore, _ := setupL4Test(t, boolPtr(false))
+	ctx := context.Background()
+
+	existing := seedService(t, testStore, "tls-svc", "tls", "app.example.com", testCluster, 443)
+
+	updated := &rpservice.Service{
+		ID:           existing.ID,
+		AccountID:    testAccountID,
+		Name:         "tls-svc",
+		Mode:         "tls",
+		Domain:       "app.example.com",
+		ProxyCluster: testCluster,
+		ListenPort:   9999,
+		Enabled:      true,
+		Targets: []*rpservice.Target{
+			{AccountID: testAccountID, TargetId: testPeerID, TargetType: rpservice.TargetTypePeer, Protocol: "tcp", Port: 8443, Enabled: true},
+		},
+	}
+
+	_, err := mgr.persistServiceUpdate(ctx, testAccountID, updated)
+	require.NoError(t, err, "TLS port change uses SNI routing and is exempt from the custom-port check")
+	assert.Equal(t, uint16(9999), updated.ListenPort, "TLS port change should be applied")
+}
+
+func TestValidateL4PortDiffOnClusterDiff(t *testing.T) {
+	tests := []struct {
+		name        string
+		mode        string
+		customPorts *bool
+		newPort     uint16
+		oldPort     uint16
+		wantErr     bool
+	}{
+		{"tcp port change unsupported", "tcp", boolPtr(false), 54321, 12345, true},
+		{"tcp port change unknown capability", "tcp", nil, 54321, 12345, true},
+		{"udp port change unsupported", "udp", boolPtr(false), 54321, 12345, true},
+		{"tcp first port assignment unsupported", "tcp", boolPtr(false), 54321, 0, true},
+		{"tcp port change supported", "tcp", boolPtr(true), 54321, 12345, false},
+		{"tcp port unchanged unsupported", "tcp", boolPtr(false), 12345, 12345, false},
+		{"tcp zero port unsupported", "tcp", boolPtr(false), 0, 12345, false},
+		{"tls port change unsupported", "tls", boolPtr(false), 9999, 443, false},
+		{"http mode ignored", "http", boolPtr(false), 54321, 12345, false},
+		{"empty mode ignored", "", boolPtr(false), 54321, 12345, false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			newSvc := &rpservice.Service{Mode: tc.mode, ListenPort: tc.newPort, ProxyCluster: testCluster}
+			oldSvc := &rpservice.Service{Mode: tc.mode, ListenPort: tc.oldPort, ProxyCluster: testCluster}
+
+			err := validateL4PortDiffOnClusterDiff(tc.customPorts, newSvc, oldSvc)
+			if tc.wantErr {
+				assert.Error(t, err, "port diff should be rejected for %s", tc.name)
+			} else {
+				assert.NoError(t, err, "port diff should be allowed for %s", tc.name)
+			}
+		})
+	}
+}
+
+func TestUpdate_PortConflictRejected(t *testing.T) {
+	mgr, testStore, _ := setupL4Test(t, boolPtr(true))
+	ctx := context.Background()
+
+	seedService(t, testStore, "tcp-a", "tcp", "tcp-a."+testCluster, testCluster, 5432)
+	svcB := seedService(t, testStore, "tcp-b", "tcp", "tcp-b."+testCluster, testCluster, 6543)
+
+	updated := &rpservice.Service{
+		ID:           svcB.ID,
+		AccountID:    testAccountID,
+		Name:         "tcp-b",
+		Mode:         "tcp",
+		Domain:       "tcp-b." + testCluster,
+		ProxyCluster: testCluster,
+		ListenPort:   5432,
+		Enabled:      true,
+		Targets: []*rpservice.Target{
+			{AccountID: testAccountID, TargetId: testPeerID, TargetType: rpservice.TargetTypePeer, Protocol: "tcp", Port: 9090, Enabled: true},
+		},
+	}
+
+	_, err := mgr.persistServiceUpdate(ctx, testAccountID, updated)
+	require.Error(t, err, "updating to a port held by another service should be rejected")
+	assert.Contains(t, err.Error(), "already in use")
+}
+
+func TestUpdate_AutoAssignsWhenNoPort(t *testing.T) {
+	mgr, testStore, _ := setupL4Test(t, boolPtr(false))
+	ctx := context.Background()
+
+	existing := seedService(t, testStore, "tcp-svc", "tcp", testCluster, testCluster, 0)
+
+	updated := &rpservice.Service{
+		ID:           existing.ID,
+		AccountID:    testAccountID,
+		Name:         "tcp-svc",
+		Mode:         "tcp",
+		Domain:       testCluster,
+		ProxyCluster: testCluster,
+		ListenPort:   0,
+		Enabled:      true,
+		Targets: []*rpservice.Target{
+			{AccountID: testAccountID, TargetId: testPeerID, TargetType: rpservice.TargetTypePeer, Protocol: "tcp", Port: 9090, Enabled: true},
+		},
+	}
+
+	_, err := mgr.persistServiceUpdate(ctx, testAccountID, updated)
+	require.NoError(t, err)
+	assert.True(t, updated.ListenPort >= autoAssignPortMin && updated.ListenPort <= autoAssignPortMax,
+		"auto-assigned port %d should be in range [%d, %d]", updated.ListenPort, autoAssignPortMin, autoAssignPortMax)
+	assert.True(t, updated.PortAutoAssigned, "PortAutoAssigned should be set when update triggers auto-assignment")
+}
+
 func TestCreateServiceFromPeer_TCP(t *testing.T) {
 	mgr, _, _ := setupL4Test(t, boolPtr(false))
 	ctx := context.Background()

management/internals/modules/reverseproxy/service/manager/manager.go

@@ -338,7 +338,7 @@ func (m *Manager) persistNewService(ctx context.Context, accountID string, svc *
 			}
 		}
 
-		if err := m.ensureL4Port(ctx, transaction, svc, customPorts); err != nil {
+		if err := m.ensureL4Port(ctx, transaction, svc, customPorts, false); err != nil {
 			return err
 		}
 
@@ -367,11 +367,11 @@ func (m *Manager) clusterCustomPorts(ctx context.Context, svc *service.Service)
 
 // ensureL4Port auto-assigns a listen port when needed and validates cluster support.
 // customPorts must be pre-computed via clusterCustomPorts before entering a transaction.
-func (m *Manager) ensureL4Port(ctx context.Context, tx store.Store, svc *service.Service, customPorts *bool) error {
+func (m *Manager) ensureL4Port(ctx context.Context, tx store.Store, svc *service.Service, customPorts *bool, serviceUpdate bool) error {
 	if !service.IsL4Protocol(svc.Mode) {
 		return nil
 	}
-	if service.IsPortBasedProtocol(svc.Mode) && svc.ListenPort > 0 && (customPorts == nil || !*customPorts) {
+	if service.IsPortBasedProtocol(svc.Mode) && svc.ListenPort > 0 && !serviceUpdate && (customPorts == nil || !*customPorts) {
 		if svc.Source != service.SourceEphemeral {
 			return status.Errorf(status.InvalidArgument, "custom ports not supported on cluster %s", svc.ProxyCluster)
 		}
@@ -465,7 +465,7 @@ func (m *Manager) persistNewEphemeralService(ctx context.Context, accountID, pee
 			return err
 		}
 
-		if err := m.ensureL4Port(ctx, transaction, svc, customPorts); err != nil {
+		if err := m.ensureL4Port(ctx, transaction, svc, customPorts, false); err != nil {
 			return err
 		}
 
@@ -651,12 +651,22 @@ func (m *Manager) executeServiceUpdate(ctx context.Context, transaction store.St
 	m.preserveListenPort(service, existingService)
 	updateInfo.serviceEnabledChanged = existingService.Enabled != service.Enabled
 
-	if err := m.ensureL4Port(ctx, transaction, service, customPorts); err != nil {
+	// if the service is being updated, and we decide in the future to allow mode update,
+	// we should reconsider the currently assigned port if not 0 for clusters that don't support custom ports
+	if err := validateL4PortDiffOnClusterDiff(customPorts, service, existingService); err != nil {
 		return err
 	}
+
+	if err := m.ensureL4Port(ctx, transaction, service, customPorts, true); err != nil {
+		return err
+	}
+
+	// we can try carrying the previous service port into a new cluster, if this becomes a problem for multiple users,
+	// we should reconsider adding another check
 	if err := m.checkPortConflict(ctx, transaction, service); err != nil {
 		return err
 	}
+
 	if err := transaction.UpdateService(ctx, service); err != nil {
 		return fmt.Errorf("update service: %w", err)
 	}
@@ -664,6 +674,21 @@ func (m *Manager) executeServiceUpdate(ctx context.Context, transaction store.St
 	return nil
 }
 
+// validateL4PortDiffOnClusterDiff checks if custom L4 ports are configured and validates port changes across clusters.
+// It ensures no port changes if custom ports are unsupported for a given cluster and protocol mode.
+// Returns an error if validation fails, otherwise returns nil.
+func validateL4PortDiffOnClusterDiff(customPorts *bool, newSVC, oldSVC *service.Service) error {
+	if !service.IsPortBasedProtocol(newSVC.Mode) || (customPorts != nil && *customPorts) {
+		return nil
+	}
+
+	if newSVC.ListenPort != 0 && newSVC.ListenPort != oldSVC.ListenPort {
+		return status.Errorf(status.InvalidArgument, "custom ports not supported on target cluster %s", newSVC.ProxyCluster)
+	}
+
+	return nil
+}
+
 // handleDomainChange validates the new domain is free inside the transaction
 // and applies the pre-resolved cluster (computed outside the tx by
 // resolveEffectiveCluster). It must NOT call clusterDeriver here: that talks
@@ -893,6 +918,10 @@ func (m *Manager) DeleteAllServices(ctx context.Context, accountID, userID strin
 		}
 
 		for _, svc := range services {
+			if err = transaction.DeleteServiceTargets(ctx, accountID, svc.ID); err != nil {
+				return fmt.Errorf("failed to delete service targets: %w", err)
+			}
+
 			if err = transaction.DeleteService(ctx, accountID, svc.ID); err != nil {
 				return fmt.Errorf("failed to delete service: %w", err)
 			}
@@ -1245,6 +1274,10 @@ func (m *Manager) deletePeerService(ctx context.Context, accountID, peerID, serv
 			return status.Errorf(status.PermissionDenied, "cannot delete service exposed by another peer")
 		}
 
+		if err = transaction.DeleteServiceTargets(ctx, accountID, serviceID); err != nil {
+			return fmt.Errorf("delete service targets: %w", err)
+		}
+
 		if err = transaction.DeleteService(ctx, accountID, serviceID); err != nil {
 			return fmt.Errorf("delete service: %w", err)
 		}
@@ -1294,6 +1327,10 @@ func (m *Manager) deleteExpiredPeerService(ctx context.Context, accountID, peerI
 			return nil
 		}
 
+		if err = transaction.DeleteServiceTargets(ctx, accountID, serviceID); err != nil {
+			return fmt.Errorf("delete service targets: %w", err)
+		}
+
 		if err = transaction.DeleteService(ctx, accountID, serviceID); err != nil {
 			return fmt.Errorf("delete service: %w", err)
 		}

management/internals/modules/reverseproxy/service/manager/manager_test.go

@@ -458,6 +458,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) {
 				txMock.EXPECT().
 					GetServiceByID(ctx, store.LockingStrengthUpdate, accountID, serviceID).
 					Return(newEphemeralService(), nil)
+				txMock.EXPECT().
+					DeleteServiceTargets(ctx, accountID, serviceID).
+					Return(nil)
 				txMock.EXPECT().
 					DeleteService(ctx, accountID, serviceID).
 					Return(nil)
@@ -560,6 +563,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) {
 				txMock.EXPECT().
 					GetServiceByID(ctx, store.LockingStrengthUpdate, accountID, serviceID).
 					Return(newEphemeralService(), nil)
+				txMock.EXPECT().
+					DeleteServiceTargets(ctx, accountID, serviceID).
+					Return(nil)
 				txMock.EXPECT().
 					DeleteService(ctx, accountID, serviceID).
 					Return(nil)
@@ -604,6 +610,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) {
 				txMock.EXPECT().
 					GetServiceByID(ctx, store.LockingStrengthUpdate, accountID, serviceID).
 					Return(newEphemeralService(), nil)
+				txMock.EXPECT().
+					DeleteServiceTargets(ctx, accountID, serviceID).
+					Return(nil)
 				txMock.EXPECT().
 					DeleteService(ctx, accountID, serviceID).
 					Return(nil)
@@ -1192,6 +1201,67 @@ func TestDeleteService_DeletesTargets(t *testing.T) {
 	assert.Len(t, targets, 0, "All targets should be deleted when service is deleted")
 }
 
+func TestDeleteExpiredPeerService_DeletesTargets(t *testing.T) {
+	ctx := context.Background()
+	mgr, testStore := setupIntegrationTest(t)
+
+	resp, err := mgr.CreateServiceFromPeer(ctx, testAccountID, testPeerID, &rpservice.ExposeServiceRequest{
+		Port: 8080,
+		Mode: "http",
+	})
+	require.NoError(t, err)
+
+	svcID := resolveServiceIDByDomain(t, testStore, resp.Domain)
+
+	targets, err := testStore.GetTargetsByServiceID(ctx, store.LockingStrengthNone, testAccountID, svcID)
+	require.NoError(t, err)
+	require.Len(t, targets, 1, "ephemeral peer-exposed service should have exactly one persisted target before reaping")
+
+	expireEphemeralService(t, testStore, testAccountID, resp.Domain)
+	err = mgr.deleteExpiredPeerService(ctx, testAccountID, testPeerID, svcID)
+	require.NoError(t, err)
+
+	_, err = testStore.GetServiceByDomain(ctx, resp.Domain)
+	require.Error(t, err, "expired peer-exposed service should be deleted")
+	s, ok := status.FromError(err)
+	require.True(t, ok)
+	assert.Equal(t, status.NotFound, s.Type())
+
+	targets, err = testStore.GetTargetsByServiceID(ctx, store.LockingStrengthNone, testAccountID, svcID)
+	require.NoError(t, err)
+	assert.Len(t, targets, 0, "orphaned target rows must be deleted when an expired peer-exposed service is reaped")
+}
+
+func TestDeleteServiceFromPeer_DeletesTargets(t *testing.T) {
+	ctx := context.Background()
+	mgr, testStore := setupIntegrationTest(t)
+
+	resp, err := mgr.CreateServiceFromPeer(ctx, testAccountID, testPeerID, &rpservice.ExposeServiceRequest{
+		Port: 8080,
+		Mode: "http",
+	})
+	require.NoError(t, err)
+
+	svcID := resolveServiceIDByDomain(t, testStore, resp.Domain)
+
+	targets, err := testStore.GetTargetsByServiceID(ctx, store.LockingStrengthNone, testAccountID, svcID)
+	require.NoError(t, err)
+	require.Len(t, targets, 1, "ephemeral peer-exposed service should have exactly one persisted target before stopping")
+
+	err = mgr.StopServiceFromPeer(ctx, testAccountID, testPeerID, svcID)
+	require.NoError(t, err)
+
+	_, err = testStore.GetServiceByDomain(ctx, resp.Domain)
+	require.Error(t, err, "stopped peer-exposed service should be deleted")
+	s, ok := status.FromError(err)
+	require.True(t, ok)
+	assert.Equal(t, status.NotFound, s.Type())
+
+	targets, err = testStore.GetTargetsByServiceID(ctx, store.LockingStrengthNone, testAccountID, svcID)
+	require.NoError(t, err)
+	assert.Len(t, targets, 0, "orphaned target rows must be deleted when a peer stops its exposed service")
+}
+
 func TestValidateProtocolChange(t *testing.T) {
 	tests := []struct {
 		name    string

management/internals/shared/grpc/conversion.go

@@ -8,6 +8,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/hashicorp/go-version"
+	nbversion "github.com/netbirdio/netbird/version"
 	log "github.com/sirupsen/logrus"
 	goproto "google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -26,6 +28,23 @@ import (
 	"github.com/netbirdio/netbird/shared/sshauth"
 )
 
+const (
+	// deprecatedRemotePeersVersion is the version of Netbird that introduced the NetworkMap.RemotePeers field, deprecated in favor of RemotePeers.
+	deprecatedRemotePeersVersion = "0.29.3"
+)
+
+// precomputedDeprecatedRemotePeersConstraint is the parsed ">= 0.29.3" constraint,
+// built once at init since the bound is a compile-time constant.
+var precomputedDeprecatedRemotePeersConstraint version.Constraints
+
+func init() {
+	constraint, err := version.NewConstraint(">= " + deprecatedRemotePeersVersion)
+	if err != nil {
+		panic("parse deprecated remote peers version constraint: " + err.Error())
+	}
+	precomputedDeprecatedRemotePeersConstraint = constraint
+}
+
 func toNetbirdConfig(config *nbconfig.Config, turnCredentials *Token, relayToken *Token, extraSettings *types.ExtraSettings) *proto.NetbirdConfig {
 	if config == nil {
 		return nil
@@ -151,7 +170,11 @@ func ToSyncResponse(ctx context.Context, config *nbconfig.Config, httpConfig *nb
 
 	remotePeers := make([]*proto.RemotePeerConfig, 0, len(networkMap.Peers)+len(networkMap.OfflinePeers))
 	remotePeers = appendRemotePeerConfig(remotePeers, networkMap.Peers, dnsName, includeIPv6)
-	response.RemotePeers = remotePeers
+
+	if !shouldSkipSendingDeprecatedRemotePeers(peer.Meta.WtVersion) {
+		response.RemotePeers = remotePeers
+	}
+
 	response.NetworkMap.RemotePeers = remotePeers
 	response.RemotePeersIsEmpty = len(remotePeers) == 0
 	response.NetworkMap.RemotePeersIsEmpty = response.RemotePeersIsEmpty
@@ -242,6 +265,19 @@ func buildAuthorizedUsersProto(ctx context.Context, authorizedUsers map[string]m
 	return hashedUsers, machineUsers
 }
 
+func shouldSkipSendingDeprecatedRemotePeers(peerVersion string) bool {
+	if nbversion.IsDevelopmentVersion(peerVersion) {
+		return true
+	}
+
+	peerNBVersion, err := version.NewVersion(peerVersion)
+	if err != nil {
+		return false
+	}
+
+	return precomputedDeprecatedRemotePeersConstraint.Check(peerNBVersion)
+}
+
 func appendRemotePeerConfig(dst []*proto.RemotePeerConfig, peers []*nbpeer.Peer, dnsName string, includeIPv6 bool) []*proto.RemotePeerConfig {
 	for _, rPeer := range peers {
 		allowedIPs := []string{rPeer.IP.String() + "/32"}

management/internals/shared/grpc/conversion_test.go

@@ -202,6 +202,42 @@ func TestBuildJWTConfig_Audiences(t *testing.T) {
 	}
 }
 
+// TestShouldSkipSendingDeprecatedRemotePeers covers the version gate that
+// stops populating the deprecated top-level SyncResponse.RemotePeers field for
+// peers new enough to read RemotePeers off the NetworkMap. Development builds
+// are treated as latest and skip the field. The gate otherwise fails safe: a
+// release version older than the boundary, or one that can't be parsed (empty,
+// garbage, prereleases of the boundary) still receives the deprecated field so
+// older/unknown clients keep working.
+func TestShouldSkipSendingDeprecatedRemotePeers(t *testing.T) {
+	tests := []struct {
+		name        string
+		peerVersion string
+		wantSkip    bool
+	}{
+		{"exact boundary skips", "0.29.3", true},
+		{"newer patch skips", "0.29.4", true},
+		{"newer minor skips", "0.30.0", true},
+		{"newer major skips", "1.0.0", true},
+		{"v-prefixed newer skips", "v0.30.0", true},
+		{"development build skips", "development", true},
+		{"development build with commit skips", "development-abc123def456-dirty", true},
+		{"older patch keeps field", "0.29.2", false},
+		{"older minor keeps field", "0.28.0", false},
+		{"prerelease of boundary keeps field", "0.29.3-SNAPSHOT", false},
+		{"tagged dev prerelease keeps field", "v0.31.1-dev", false},
+		{"empty version keeps field", "", false},
+		{"garbage version keeps field", "not-a-version", false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got := shouldSkipSendingDeprecatedRemotePeers(tc.peerVersion)
+			assert.Equal(t, tc.wantSkip, got, "skip decision for peer version %q", tc.peerVersion)
+		})
+	}
+}
+
 // TestEncodeSessionExpiresAt pins the wire encoding the client's
 // applySessionDeadline depends on:
 //

management/internals/shared/grpc/proxy.go

@@ -666,8 +666,10 @@ func (s *ProxyServiceServer) sender(conn *proxyConnection, errChan chan<- error)
 		case resp := <-conn.sendChan:
 			if err := conn.sendResponse(resp); err != nil {
 				errChan <- err
+				log.WithContext(conn.ctx).Tracef("Failed to send response to proxy %s: %v", conn.proxyID, err)
 				return
 			}
+			log.WithContext(conn.ctx).Tracef("Send response to proxy %s", conn.proxyID)
 		case <-conn.ctx.Done():
 			return
 		}
@@ -978,6 +980,7 @@ func shallowCloneMapping(m *proto.ProxyMapping) *proto.ProxyMapping {
 		Mode:               m.Mode,
 		ListenPort:         m.ListenPort,
 		AccessRestrictions: m.AccessRestrictions,
+		Private:            m.Private,
 	}
 }
 

management/internals/shared/grpc/proxy_clone_test.go

@@ -0,0 +1,88 @@
+package grpc
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/netbirdio/netbird/shared/management/proto"
+)
+
+// authTokenField is the only per-proxy field that shallowCloneMapping must NOT
+// copy from the source, since callers assign it individually after cloning.
+const authTokenField = "AuthToken"
+
+// TestShallowCloneMapping_ClonesAllFields populates every exported field of
+// ProxyMapping with a non-zero value and verifies the clone carries each one
+// (except AuthToken). It uses reflection so adding a new field to ProxyMapping
+// without updating shallowCloneMapping fails this test.
+func TestShallowCloneMapping_ClonesAllFields(t *testing.T) {
+	src := &proto.ProxyMapping{}
+	populated := populateExportedFields(t, reflect.ValueOf(src).Elem())
+	require.NotEmpty(t, populated, "ProxyMapping should expose fields to populate")
+
+	clone := shallowCloneMapping(src)
+	require.NotNil(t, clone, "clone must not be nil")
+
+	srcVal := reflect.ValueOf(src).Elem()
+	cloneVal := reflect.ValueOf(clone).Elem()
+
+	for _, name := range populated {
+		srcField := srcVal.FieldByName(name).Interface()
+		cloneField := cloneVal.FieldByName(name).Interface()
+
+		if name == authTokenField {
+			assert.Zero(t, cloneField, "AuthToken must not be cloned; it is set per proxy after cloning")
+			continue
+		}
+
+		assert.Equal(t, srcField, cloneField, "field %s must be carried over by shallowCloneMapping", name)
+	}
+}
+
+// populateExportedFields sets a non-zero value on every settable exported field
+// of the struct and returns their names.
+func populateExportedFields(t *testing.T, v reflect.Value) []string {
+	t.Helper()
+
+	var names []string
+	typ := v.Type()
+	for i := 0; i < v.NumField(); i++ {
+		field := v.Field(i)
+		structField := typ.Field(i)
+
+		if structField.PkgPath != "" || !field.CanSet() {
+			continue
+		}
+
+		setNonZero(t, field, structField.Name)
+		names = append(names, structField.Name)
+	}
+	return names
+}
+
+// setNonZero assigns a deterministic non-zero value based on the field kind.
+func setNonZero(t *testing.T, field reflect.Value, name string) {
+	t.Helper()
+
+	switch field.Kind() {
+	case reflect.String:
+		field.SetString("non-zero-" + name)
+	case reflect.Bool:
+		field.SetBool(true)
+	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
+		field.SetInt(7)
+	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
+		field.SetUint(7)
+	case reflect.Ptr:
+		field.Set(reflect.New(field.Type().Elem()))
+	case reflect.Slice:
+		field.Set(reflect.MakeSlice(field.Type(), 1, 1))
+	case reflect.Map:
+		field.Set(reflect.MakeMapWithSize(field.Type(), 0))
+	default:
+		t.Fatalf("unhandled field kind %s for field %s; extend setNonZero", field.Kind(), name)
+	}
+}