Advance the network serial when a peer sync or login changes map content

management/internals/modules/peers/ephemeral/manager/ephemeral.go

@@ -216,52 +216,13 @@ func (e *EphemeralManager) cleanup(ctx context.Context) {
 
 	for accountID, peerIDs := range peerIDsPerAccount {
 		log.WithContext(ctx).Debugf("cleanup: deleting %d ephemeral peers for account %s: %s", len(peerIDs), accountID, peerIDs)
-		skipped, err := e.peersManager.DeletePeers(ctx, accountID, peerIDs, activity.SystemInitiator, true)
+		err := e.peersManager.DeletePeers(ctx, accountID, peerIDs, activity.SystemInitiator, true)
 		if err != nil {
 			log.WithContext(ctx).Errorf("failed to delete ephemeral peers: %s", err)
 			e.metrics.CountCleanupError()
 			continue
 		}
-		if len(skipped) > 0 {
-			// A skipped peer could not be deleted yet (still connected in the
-			// store, or seen too recently), which says nothing about whether a
-			// disconnect will ever be observed for it again. Schedule another
-			// attempt instead of dropping it, or it is never collected.
-			log.WithContext(ctx).Debugf("cleanup: requeueing %d skipped ephemeral peers for account %s: %s", len(skipped), accountID, skipped)
-			e.requeuePeers(ctx, accountID, skipped)
-		}
-		e.metrics.CountPeersCleaned(int64(len(peerIDs) - len(skipped)))
-	}
-}
-
-// requeuePeers puts peers whose deletion was skipped back on the list with a
-// fresh deadline. A peer that reconnected and disconnected in the meantime is
-// already listed again and keeps its existing entry.
-func (e *EphemeralManager) requeuePeers(ctx context.Context, accountID string, peerIDs []string) {
-	e.peersLock.Lock()
-	defer e.peersLock.Unlock()
-
-	added := 0
-	for _, id := range peerIDs {
-		if e.isPeerOnList(id) {
-			continue
-		}
-		e.addPeer(accountID, id, e.newDeadLine())
-		added++
-	}
-	if added == 0 {
-		return
-	}
-	e.metrics.AddPending(int64(added))
-
-	if e.timer == nil {
-		delay := e.headPeer.deadline.Sub(timeNow()) + e.cleanupWindow
-		if delay < 0 {
-			delay = 0
-		}
-		e.timer = time.AfterFunc(delay, func() {
-			e.cleanup(ctx)
-		})
+		e.metrics.CountPeersCleaned(int64(len(peerIDs)))
 	}
 }
 

management/internals/modules/peers/ephemeral/manager/ephemeral_test.go

@@ -7,9 +7,9 @@ import (
 	"testing"
 	"time"
 
+	"go.uber.org/mock/gomock"
 	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
-	"go.uber.org/mock/gomock"
 
 	nbdns "github.com/netbirdio/netbird/dns"
 	"github.com/netbirdio/netbird/management/internals/modules/peers"
@@ -104,11 +104,11 @@ func TestNewManager(t *testing.T) {
 	// Expect DeletePeers to be called for ephemeral peers
 	peersManager.EXPECT().
 		DeletePeers(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), true).
-		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
+		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) error {
 			for _, peerID := range peerIDs {
 				delete(store.account.Peers, peerID)
 			}
-			return nil, nil
+			return nil
 		}).
 		AnyTimes()
 
@@ -142,11 +142,11 @@ func TestNewManagerPeerConnected(t *testing.T) {
 	// Expect DeletePeers to be called for ephemeral peers (except the connected one)
 	peersManager.EXPECT().
 		DeletePeers(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), true).
-		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
+		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) error {
 			for _, peerID := range peerIDs {
 				delete(store.account.Peers, peerID)
 			}
-			return nil, nil
+			return nil
 		}).
 		AnyTimes()
 
@@ -183,11 +183,11 @@ func TestNewManagerPeerDisconnected(t *testing.T) {
 	// Expect DeletePeers to be called for the one disconnected peer
 	peersManager.EXPECT().
 		DeletePeers(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), true).
-		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
+		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) error {
 			for _, peerID := range peerIDs {
 				delete(store.account.Peers, peerID)
 			}
-			return nil, nil
+			return nil
 		}).
 		AnyTimes()
 
@@ -240,16 +240,16 @@ func TestCleanupSchedulingBehaviorIsBatched(t *testing.T) {
 	// Set up expectation that DeletePeers will be called once with all peer IDs
 	peersManager.EXPECT().
 		DeletePeers(gomock.Any(), account.Id, gomock.Any(), gomock.Any(), true).
-		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
+		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) error {
 			// Simulate the actual deletion behavior
 			for _, peerID := range peerIDs {
 				err := mockAM.DeletePeer(ctx, accountID, peerID, userID)
 				if err != nil {
-					return nil, err
+					return err
 				}
 			}
 			mockAM.BufferUpdateAccountPeers(ctx, accountID, types.UpdateReason{})
-			return nil, nil
+			return nil
 		}).
 		Times(1)
 
@@ -276,50 +276,6 @@ func TestCleanupSchedulingBehaviorIsBatched(t *testing.T) {
 	assert.Equal(t, ephemeralPeers, mockAM.GetDeletePeerCalls(), "should have deleted all peers")
 }
 
-// TestCleanupRequeuesVetoedPeers covers a peer whose deletion is vetoed (the
-// store still reports it connected, or it was seen too recently): it must be
-// scheduled for another attempt rather than dropped from the list, or it is
-// never collected once the veto clears.
-func TestCleanupRequeuesVetoedPeers(t *testing.T) {
-	t.Cleanup(func() {
-		timeNow = time.Now
-	})
-	startTime := time.Now()
-	timeNow = func() time.Time {
-		return startTime
-	}
-
-	mockStore := &MockStore{}
-	seedPeers(mockStore, 0, 1)
-	ctrl := gomock.NewController(t)
-	peersManager := peers.NewMockManager(ctrl)
-
-	// The first attempt vetoes the peer, the second deletes it.
-	first := peersManager.EXPECT().
-		DeletePeers(gomock.Any(), gomock.Any(), []string{"ephemeral_peer_0"}, gomock.Any(), true).
-		Return([]string{"ephemeral_peer_0"}, nil)
-	peersManager.EXPECT().
-		DeletePeers(gomock.Any(), gomock.Any(), []string{"ephemeral_peer_0"}, gomock.Any(), true).
-		After(first).
-		DoAndReturn(func(_ context.Context, _ string, peerIDs []string, _ string, _ bool) ([]string, error) {
-			for _, peerID := range peerIDs {
-				delete(mockStore.account.Peers, peerID)
-			}
-			return nil, nil
-		})
-
-	mgr := NewEphemeralManager(mockStore, peersManager)
-	mgr.loadEphemeralPeers(context.Background())
-
-	startTime = startTime.Add(ephemeral.EphemeralLifeTime + time.Second)
-	mgr.cleanup(context.Background())
-	assert.Len(t, mockStore.account.Peers, 1, "vetoed peer must not be deleted yet")
-
-	startTime = startTime.Add(ephemeral.EphemeralLifeTime + time.Second)
-	mgr.cleanup(context.Background())
-	assert.Len(t, mockStore.account.Peers, 0, "vetoed peer should be retried and deleted once the veto clears")
-}
-
 func seedPeers(store *MockStore, numberOfPeers int, numberOfEphemeralPeers int) {
 	store.account = newAccountWithId(context.Background(), "my account", "", "", false)
 

management/internals/modules/peers/manager.go

@@ -30,11 +30,7 @@ type Manager interface {
 	GetPeerAccountID(ctx context.Context, peerID string) (string, error)
 	GetAllPeers(ctx context.Context, accountID, userID string) ([]*peer.Peer, error)
 	GetPeersByGroupIDs(ctx context.Context, accountID string, groupsIDs []string) ([]*peer.Peer, error)
-	// DeletePeers removes the given peers along with their group memberships and
-	// policies. With checkConnected, a peer that is still connected or was seen
-	// too recently is left in place and returned in skipped, so the caller can
-	// retry it later.
-	DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) (skipped []string, err error)
+	DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) error
 	SetNetworkMapController(networkMapController network_map.Controller)
 	SetIntegratedPeerValidator(integratedPeerValidator integrated_validator.IntegratedValidator)
 	SetAccountManager(accountManager account.Manager)
@@ -132,22 +128,16 @@ func (m *managerImpl) GetPeerWithGroups(ctx context.Context, accountID, peerID s
 	return p, groups, nil
 }
 
-func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
+func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) error {
 	settings, err := m.store.GetAccountSettings(ctx, store.LockingStrengthNone, accountID)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	dnsDomain := m.networkMapController.GetDNSDomain(settings)
 
-	var skipped []string
-	deletedAny := false
 	for _, peerID := range peerIDs {
 		var eventsToStore []func()
-		vetoed := false
-		deleted := false
 		err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
-			vetoed = false
-			deleted = false
 			peer, err := transaction.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peerID)
 			if err != nil {
 				if e, ok := status.FromError(err); ok && e.Type() == status.NotFound {
@@ -163,7 +153,6 @@ func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs
 					peer.Status.LastSeen.Format(time.RFC3339),
 					time.Now().Add(-(ephemeral.EphemeralLifeTime - 10*time.Second)).Format(time.RFC3339),
 					peer.Ephemeral)
-				vetoed = true
 				return nil
 			}
 
@@ -194,7 +183,6 @@ func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs
 			if err = transaction.DeletePeer(ctx, accountID, peerID); err != nil {
 				return err
 			}
-			deleted = true
 
 			log.WithContext(ctx).Debugf("DeletePeers: deleted peer %s", peerID)
 
@@ -211,16 +199,6 @@ func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs
 			continue
 		}
 
-		if vetoed {
-			skipped = append(skipped, peerID)
-			continue
-		}
-
-		if !deleted {
-			continue
-		}
-		deletedAny = true
-
 		if m.integratedPeerValidator != nil {
 			if err = m.integratedPeerValidator.PeerDeleted(ctx, accountID, peerID, settings.Extra); err != nil {
 				log.WithContext(ctx).Errorf("failed to delete peer %s from integrated validator: %v", peerID, err)
@@ -232,13 +210,9 @@ func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs
 		}
 	}
 
-	// Skipped or missing peers changed nothing, so an update would push an
-	// identical map to every peer in the account.
-	if deletedAny {
-		m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationDelete})
-	}
+	m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationDelete})
 
-	return skipped, nil
+	return nil
 }
 
 func (m *managerImpl) GetPeerID(ctx context.Context, peerKey string) (string, error) {

management/internals/modules/peers/manager_mock.go

@@ -61,12 +61,11 @@ func (mr *MockManagerMockRecorder) CreateProxyPeer(ctx, accountID, peerKey, clus
 }
 
 // DeletePeers mocks base method.
-func (m *MockManager) DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
+func (m *MockManager) DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "DeletePeers", ctx, accountID, peerIDs, userID, checkConnected)
-	ret0, _ := ret[0].([]string)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
+	ret0, _ := ret[0].(error)
+	return ret0
 }
 
 // DeletePeers indicates an expected call of DeletePeers.

management/server/peer.go

@@ -1066,6 +1066,12 @@ func (am *DefaultAccountManager) SyncPeer(ctx context.Context, sync types.PeerSy
 
 	metaDiffAffectsPosture := posture.AffectsPosture(ctx, &metaDiff, resPostureChecks)
 	if requiresPeerUpdate(ctx, isStatusChanged, sync.UpdateAccountPeers, ipv6CapabilityChanged, metaDiffAffectsPosture, metaDiff.VersionChanged(), metaDiff.HostnameChanged()) {
+		// The maps pushed below carry changed content (the peer's version,
+		// hostname, capabilities, or validation state). The serial versions the
+		// distributed map, so it must advance with the content.
+		if err = am.Store.IncrementNetworkSerial(ctx, accountID); err != nil {
+			return nil, nil, nil, 0, fmt.Errorf("increment network serial: %w", err)
+		}
 		changedPeerIDs := []string{peer.ID}
 		affectedPeerIDs := am.syncPeerAffectedPeers(ctx, accountID, peer.ID, nmap, peerNotValid, metaDiffAffectsPosture)
 		if err = am.networkMapController.OnPeersUpdated(ctx, accountID, changedPeerIDs, affectedPeerIDs); err != nil {
@@ -1236,6 +1242,11 @@ func (am *DefaultAccountManager) LoginPeer(ctx context.Context, login types.Peer
 	}
 
 	if shouldUpdatePeers {
+		// The maps pushed below carry changed peer content. The serial versions
+		// the distributed map, so it must advance with the content.
+		if err = am.Store.IncrementNetworkSerial(ctx, accountID); err != nil {
+			return nil, nil, nil, false, fmt.Errorf("increment network serial: %w", err)
+		}
 		changedPeerIDs := []string{peer.ID}
 		affectedPeerIDs := am.resolveAffectedPeersForPeerChanges(ctx, am.Store, accountID, changedPeerIDs)
 		if err = am.networkMapController.OnPeersUpdated(ctx, accountID, changedPeerIDs, affectedPeerIDs); err != nil {

management/server/peer_test.go

@@ -16,11 +16,11 @@ import (
 	"testing"
 	"time"
 
-	"go.uber.org/mock/gomock"
 	"github.com/rs/xid"
 	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"golang.org/x/exp/maps"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 
@@ -2828,6 +2828,46 @@ func TestSyncPeer_IPv6CapabilityChangePropagates(t *testing.T) {
 	})
 }
 
+// TestSyncPeer_PeerUpdateBumpsNetworkSerial ensures that a sync which changes
+// map-relevant peer content (agent version, hostname, capabilities, validation
+// state) advances the network serial before other peers receive the recomputed
+// map. The serial versions the distributed map, so its content must not change
+// under an unchanged serial.
+func TestSyncPeer_PeerUpdateBumpsNetworkSerial(t *testing.T) {
+	manager, _, account, _, peer2, _ := setupNetworkMapTest(t)
+
+	network, err := manager.Store.GetAccountNetwork(context.Background(), store.LockingStrengthNone, account.Id)
+	require.NoError(t, err)
+	serialBefore := network.CurrentSerial()
+
+	t.Run("no bump when nothing changed", func(t *testing.T) {
+		_, _, _, _, err := manager.SyncPeer(context.Background(), types.PeerSync{
+			WireGuardPubKey: peer2.Key,
+			Meta:            peer2.Meta,
+		}, peer2.AccountID)
+		require.NoError(t, err)
+
+		network, err := manager.Store.GetAccountNetwork(context.Background(), store.LockingStrengthNone, account.Id)
+		require.NoError(t, err)
+		assert.Equal(t, serialBefore, network.CurrentSerial(), "an unchanged sync should not advance the serial")
+	})
+
+	t.Run("bump when the agent version changes", func(t *testing.T) {
+		newMeta := peer2.Meta
+		newMeta.WtVersion = "0.99.99"
+
+		_, _, _, _, err := manager.SyncPeer(context.Background(), types.PeerSync{
+			WireGuardPubKey: peer2.Key,
+			Meta:            newMeta,
+		}, peer2.AccountID)
+		require.NoError(t, err)
+
+		network, err := manager.Store.GetAccountNetwork(context.Background(), store.LockingStrengthNone, account.Id)
+		require.NoError(t, err)
+		assert.Greater(t, network.CurrentSerial(), serialBefore, "a map-relevant meta change should advance the serial")
+	})
+}
+
 func TestUpdatePeer_DnsLabelCollisionWithFQDN(t *testing.T) {
 	manager, _, err := createManager(t)
 	require.NoError(t, err, "unable to create account manager")