Merge remote-tracking branch 'origin/main' into merge-ephemeral-gc-requeue

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()
+		}
+		if len(skipped) > 0 {
+			// A skipped peer was not deleted: it is still connected in the
+			// store, was seen too recently, or its deletion failed. None of
+			// that says whether a disconnect will ever be observed for it
+			// again, so 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.metrics.CountPeersCleaned(int64(len(peerIDs)))
+		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

@@ -2,14 +2,15 @@ package manager
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"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 +105,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 +143,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 +184,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 +241,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 +277,94 @@ 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")
+}
+
+// TestCleanupRequeuesFailedDeletes covers a peer whose deletion attempt errors
+// (DeletePeers reports it as skipped alongside the aggregate error): it must be
+// scheduled for another attempt rather than dropped from the list, or a
+// transient store failure leaks it forever.
+func TestCleanupRequeuesFailedDeletes(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 fails, the second deletes the peer.
+	first := peersManager.EXPECT().
+		DeletePeers(gomock.Any(), gomock.Any(), []string{"ephemeral_peer_0"}, gomock.Any(), true).
+		Return([]string{"ephemeral_peer_0"}, errors.New("transient store failure"))
+	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, "peer whose deletion failed must still exist")
+
+	startTime = startTime.Add(ephemeral.EphemeralLifeTime + time.Second)
+	mgr.cleanup(context.Background())
+	assert.Len(t, mockStore.account.Peers, 0, "peer whose deletion failed should be retried and deleted")
+}
+
 func seedPeers(store *MockStore, numberOfPeers int, numberOfEphemeralPeers int) {
 	store.account = newAccountWithId(context.Background(), "my account", "", "", false)
 

management/internals/modules/peers/manager.go

@@ -8,9 +8,11 @@ import (
 	"net"
 	"time"
 
+	"github.com/hashicorp/go-multierror"
 	"github.com/rs/xid"
 	log "github.com/sirupsen/logrus"
 
+	nberrors "github.com/netbirdio/netbird/client/errors"
 	"github.com/netbirdio/netbird/management/internals/controllers/network_map"
 	"github.com/netbirdio/netbird/management/internals/modules/peers/ephemeral"
 	"github.com/netbirdio/netbird/management/server/account"
@@ -30,7 +32,12 @@ 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. Every peer that was not deleted is returned in skipped, so the
+	// caller can retry it later: with checkConnected, a peer that is still
+	// connected or was seen too recently is left in place, and a peer whose
+	// deletion failed is skipped with the failure aggregated into err.
+	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,91 +135,153 @@ 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 {
+// deletePeerOutcome is the per-peer result of a DeletePeers pass.
+type deletePeerOutcome int
+
+const (
+	// peerMissing means the peer no longer exists, so nothing changed.
+	peerMissing deletePeerOutcome = iota
+	// peerVetoed means the peer cannot be deleted yet: it is still connected or
+	// was seen too recently.
+	peerVetoed
+	// peerDeleted means the peer and its objects were removed.
+	peerDeleted
+)
+
+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 peerIDs, err
 	}
 	dnsDomain := m.networkMapController.GetDNSDomain(settings)
 
+	var skipped []string
+	var merr *multierror.Error
+	deletedAny := false
 	for _, peerID := range peerIDs {
-		var eventsToStore []func()
-		err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
-			peer, err := transaction.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peerID)
-			if err != nil {
-				if e, ok := status.FromError(err); ok && e.Type() == status.NotFound {
-					log.WithContext(ctx).Tracef("DeletePeers: peer %s not found, skipping", peerID)
-					return nil
-				}
-				return err
-			}
-
-			if checkConnected && (peer.Status.Connected || peer.Status.LastSeen.After(time.Now().Add(-(ephemeral.EphemeralLifeTime - 10*time.Second)))) {
-				log.WithContext(ctx).Tracef("DeletePeers: peer %s skipped (connected=%t, lastSeen=%s, threshold=%s, ephemeral=%t)",
-					peerID, peer.Status.Connected,
-					peer.Status.LastSeen.Format(time.RFC3339),
-					time.Now().Add(-(ephemeral.EphemeralLifeTime - 10*time.Second)).Format(time.RFC3339),
-					peer.Ephemeral)
-				return nil
-			}
-
-			if err := transaction.RemovePeerFromAllGroups(ctx, peerID); err != nil {
-				return fmt.Errorf("failed to remove peer %s from groups", peerID)
-			}
-
-			peerPolicyRules, err := transaction.GetPolicyRulesByResourceID(ctx, store.LockingStrengthNone, accountID, peerID)
-			if err != nil {
-				return err
-			}
-			for _, rule := range peerPolicyRules {
-				policy, err := transaction.GetPolicyByID(ctx, store.LockingStrengthNone, accountID, rule.PolicyID)
-				if err != nil {
-					return err
-				}
-
-				err = transaction.DeletePolicy(ctx, accountID, rule.PolicyID)
-				if err != nil {
-					return err
-				}
-
-				eventsToStore = append(eventsToStore, func() {
-					m.accountManager.StoreEvent(ctx, userID, peer.ID, accountID, activity.PolicyRemoved, policy.EventMeta())
-				})
-			}
-
-			if err = transaction.DeletePeer(ctx, accountID, peerID); err != nil {
-				return err
-			}
-
-			log.WithContext(ctx).Debugf("DeletePeers: deleted peer %s", peerID)
-
-			if !(peer.ProxyMeta.Embedded || peer.Meta.KernelVersion == "wasm") {
-				eventsToStore = append(eventsToStore, func() {
-					m.accountManager.StoreEvent(ctx, userID, peer.ID, accountID, activity.PeerRemovedByUser, peer.EventMeta(dnsDomain))
-				})
-			}
-
-			return nil
-		})
+		outcome, events, err := m.deleteSinglePeer(ctx, accountID, peerID, userID, checkConnected, dnsDomain)
 		if err != nil {
-			log.WithContext(ctx).Errorf("DeletePeers: failed to delete peer %s: %v", peerID, err)
+			merr = multierror.Append(merr, fmt.Errorf("delete peer %s: %w", peerID, err))
+			skipped = append(skipped, peerID)
 			continue
 		}
 
-		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)
-			}
-		}
-
-		for _, event := range eventsToStore {
-			event()
+		switch outcome {
+		case peerVetoed:
+			skipped = append(skipped, peerID)
+		case peerDeleted:
+			deletedAny = true
+			m.notifyPeerDeleted(ctx, accountID, peerID, settings.Extra, events)
+		case peerMissing:
 		}
 	}
 
-	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, nberrors.FormatErrorOrNil(merr)
+}
+
+// deleteSinglePeer deletes one peer along with its group memberships and
+// policies in a single transaction. With checkConnected, a peer that is still
+// connected or was seen too recently is left untouched and reported as vetoed.
+// The returned events must be stored by the caller once the deletion is final.
+func (m *managerImpl) deleteSinglePeer(ctx context.Context, accountID, peerID, userID string, checkConnected bool, dnsDomain string) (deletePeerOutcome, []func(), error) {
+	outcome := peerMissing
+	var eventsToStore []func()
+	err := m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
+		outcome = peerMissing
+		eventsToStore = nil
+
+		p, err := transaction.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peerID)
+		if err != nil {
+			if e, ok := status.FromError(err); ok && e.Type() == status.NotFound {
+				log.WithContext(ctx).Tracef("DeletePeers: peer %s not found, skipping", peerID)
+				return nil
+			}
+			return err
+		}
+
+		if checkConnected && (p.Status.Connected || p.Status.LastSeen.After(time.Now().Add(-(ephemeral.EphemeralLifeTime - 10*time.Second)))) {
+			log.WithContext(ctx).Tracef("DeletePeers: peer %s skipped (connected=%t, lastSeen=%s, threshold=%s, ephemeral=%t)",
+				peerID, p.Status.Connected,
+				p.Status.LastSeen.Format(time.RFC3339),
+				time.Now().Add(-(ephemeral.EphemeralLifeTime - 10*time.Second)).Format(time.RFC3339),
+				p.Ephemeral)
+			outcome = peerVetoed
+			return nil
+		}
+
+		eventsToStore, err = m.deletePeerObjects(ctx, transaction, accountID, userID, dnsDomain, p)
+		if err != nil {
+			return err
+		}
+		outcome = peerDeleted
+		return nil
+	})
+	if err != nil {
+		return outcome, nil, err
+	}
+	return outcome, eventsToStore, nil
+}
+
+// deletePeerObjects removes the peer's group memberships, its policies and the
+// peer itself within the given transaction, returning the activity events to
+// store once the transaction commits.
+func (m *managerImpl) deletePeerObjects(ctx context.Context, transaction store.Store, accountID, userID, dnsDomain string, p *peer.Peer) ([]func(), error) {
+	if err := transaction.RemovePeerFromAllGroups(ctx, p.ID); err != nil {
+		return nil, fmt.Errorf("remove peer %s from groups: %w", p.ID, err)
+	}
+
+	var eventsToStore []func()
+	peerPolicyRules, err := transaction.GetPolicyRulesByResourceID(ctx, store.LockingStrengthNone, accountID, p.ID)
+	if err != nil {
+		return nil, err
+	}
+	for _, rule := range peerPolicyRules {
+		policy, err := transaction.GetPolicyByID(ctx, store.LockingStrengthNone, accountID, rule.PolicyID)
+		if err != nil {
+			return nil, err
+		}
+
+		if err := transaction.DeletePolicy(ctx, accountID, rule.PolicyID); err != nil {
+			return nil, err
+		}
+
+		eventsToStore = append(eventsToStore, func() {
+			m.accountManager.StoreEvent(ctx, userID, p.ID, accountID, activity.PolicyRemoved, policy.EventMeta())
+		})
+	}
+
+	if err := transaction.DeletePeer(ctx, accountID, p.ID); err != nil {
+		return nil, err
+	}
+
+	log.WithContext(ctx).Debugf("DeletePeers: deleted peer %s", p.ID)
+
+	if !(p.ProxyMeta.Embedded || p.Meta.KernelVersion == "wasm") {
+		eventsToStore = append(eventsToStore, func() {
+			m.accountManager.StoreEvent(ctx, userID, p.ID, accountID, activity.PeerRemovedByUser, p.EventMeta(dnsDomain))
+		})
+	}
+
+	return eventsToStore, nil
+}
+
+// notifyPeerDeleted reports a completed deletion to the integrated validator and
+// stores the deletion's activity events.
+func (m *managerImpl) notifyPeerDeleted(ctx context.Context, accountID, peerID string, extraSettings *types.ExtraSettings, events []func()) {
+	if m.integratedPeerValidator != nil {
+		if err := m.integratedPeerValidator.PeerDeleted(ctx, accountID, peerID, extraSettings); err != nil {
+			log.WithContext(ctx).Errorf("failed to delete peer %s from integrated validator: %v", peerID, err)
+		}
+	}
+	for _, event := range events {
+		event()
+	}
 }
 
 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.