Requeue ephemeral peers whose deletion was vetoed instead of dropping them

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

@@ -216,13 +216,52 @@ 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)
-		err := e.peersManager.DeletePeers(ctx, accountID, peerIDs, activity.SystemInitiator, true)
+		skipped, 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
 		}
-		e.metrics.CountPeersCleaned(int64(len(peerIDs)))
+		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)
+		})
 	}
 }
 

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) error {
+		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
 			for _, peerID := range peerIDs {
 				delete(store.account.Peers, peerID)
 			}
-			return nil
+			return nil, 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) error {
+		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
 			for _, peerID := range peerIDs {
 				delete(store.account.Peers, peerID)
 			}
-			return nil
+			return nil, 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) error {
+		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
 			for _, peerID := range peerIDs {
 				delete(store.account.Peers, peerID)
 			}
-			return nil
+			return nil, 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) error {
+		DoAndReturn(func(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
 			// Simulate the actual deletion behavior
 			for _, peerID := range peerIDs {
 				err := mockAM.DeletePeer(ctx, accountID, peerID, userID)
 				if err != nil {
-					return err
+					return nil, err
 				}
 			}
 			mockAM.BufferUpdateAccountPeers(ctx, accountID, types.UpdateReason{})
-			return nil
+			return nil, nil
 		}).
 		Times(1)
 
@@ -276,6 +276,50 @@ 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,7 +30,11 @@ 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(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) 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)
 	SetNetworkMapController(networkMapController network_map.Controller)
 	SetIntegratedPeerValidator(integratedPeerValidator integrated_validator.IntegratedValidator)
 	SetAccountManager(accountManager account.Manager)
@@ -128,16 +132,22 @@ 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) error {
+func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
 	settings, err := m.store.GetAccountSettings(ctx, store.LockingStrengthNone, accountID)
 	if err != nil {
-		return err
+		return nil, 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 {
@@ -153,6 +163,7 @@ 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
 			}
 
@@ -183,6 +194,7 @@ 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)
 
@@ -199,6 +211,16 @@ 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)
@@ -210,9 +232,13 @@ func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs
 		}
 	}
 
-	m.accountManager.UpdateAccountPeers(ctx, accountID, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationDelete})
+	// 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})
+	}
 
-	return nil
+	return skipped, nil
 }
 
 func (m *managerImpl) GetPeerID(ctx context.Context, peerKey string) (string, error) {

management/internals/modules/peers/manager_mock.go

@@ -61,11 +61,12 @@ 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) error {
+func (m *MockManager) DeletePeers(ctx context.Context, accountID string, peerIDs []string, userID string, checkConnected bool) ([]string, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "DeletePeers", ctx, accountID, peerIDs, userID, checkConnected)
-	ret0, _ := ret[0].(error)
-	return ret0
+	ret0, _ := ret[0].([]string)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
 }
 
 // DeletePeers indicates an expected call of DeletePeers.