From aa90ac102e3fc372fd16f5189ddc974f56913998 Mon Sep 17 00:00:00 2001 From: Viktor Liu Date: Thu, 20 Aug 2026 14:28:15 +0200 Subject: [PATCH] Requeue ephemeral peers whose deletion was vetoed instead of dropping them --- .../peers/ephemeral/manager/ephemeral.go | 43 ++++++++++++- .../peers/ephemeral/manager/ephemeral_test.go | 64 ++++++++++++++++--- management/internals/modules/peers/manager.go | 36 +++++++++-- .../internals/modules/peers/manager_mock.go | 7 +- 4 files changed, 130 insertions(+), 20 deletions(-) diff --git a/management/internals/modules/peers/ephemeral/manager/ephemeral.go b/management/internals/modules/peers/ephemeral/manager/ephemeral.go index 7867d6579..4a8683aa0 100644 --- a/management/internals/modules/peers/ephemeral/manager/ephemeral.go +++ b/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) + }) } } diff --git a/management/internals/modules/peers/ephemeral/manager/ephemeral_test.go b/management/internals/modules/peers/ephemeral/manager/ephemeral_test.go index 1b64c447a..97b19e360 100644 --- a/management/internals/modules/peers/ephemeral/manager/ephemeral_test.go +++ b/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) diff --git a/management/internals/modules/peers/manager.go b/management/internals/modules/peers/manager.go index 3274ec524..bcf0da70c 100644 --- a/management/internals/modules/peers/manager.go +++ b/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) { diff --git a/management/internals/modules/peers/manager_mock.go b/management/internals/modules/peers/manager_mock.go index 8c26d43b1..b520b1b78 100644 --- a/management/internals/modules/peers/manager_mock.go +++ b/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.