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

client/internal/engine.go

@@ -863,40 +863,19 @@ func (e *Engine) modifyPeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
 		}
 	}
 
-	// second, look up the activation state of all modified peers before removing
-	// any of them, so an unavailable state leaves the current connections intact
-	active := make(map[string]bool, len(modified))
+	// second, close all modified connections and remove them from the state map
 	for _, p := range modified {
-		peerPubKey := p.GetWgPubKey()
-		state, err := e.statusRecorder.GetPeer(peerPubKey)
+		err := e.removePeer(p.GetWgPubKey())
 		if err != nil {
-			return fmt.Errorf("get status of modified peer %s: %w", peerPubKey, err)
-		}
-		active[peerPubKey] = state.ConnStatus != peer.StatusIdle
-	}
-	// then close all modified connections and remove them from the state map
-	for _, p := range modified {
-		if err := e.removePeer(p.GetWgPubKey()); err != nil {
 			return err
 		}
 	}
-	// third, add the peer connections again, restoring each peer's activation
-	// state: under lazy connections a re-added peer starts idle, but the remote
-	// side of an established connection keeps its state and sends no further
-	// offers, so a previously active peer left idle cannot reconnect until the
-	// remote's connection expires.
+	// third, add the peer connections again
 	for _, p := range modified {
-		if err := e.addNewPeer(p); err != nil {
+		err := e.addNewPeer(p)
+		if err != nil {
 			return err
 		}
-		if !active[p.GetWgPubKey()] {
-			continue
-		}
-		conn, ok := e.peerStore.PeerConn(p.GetWgPubKey())
-		if !ok {
-			continue
-		}
-		e.connMgr.ActivatePeer(e.ctx, conn)
 	}
 	return nil
 }

client/internal/engine_test.go

@@ -26,7 +26,6 @@ import (
 	"github.com/netbirdio/netbird/client/iface/wgaddr"
 	"github.com/netbirdio/netbird/client/iface/wgproxy"
 	"github.com/netbirdio/netbird/client/internal/dns"
-	"github.com/netbirdio/netbird/client/internal/lazyconn"
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/client/internal/peer/guard"
 	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
@@ -467,163 +466,6 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 	}
 }
 
-// TestEngine_ModifiedPeerKeepsActivationState verifies that a peer re-added by
-// modifyPeers keeps its previous activation state under lazy connections. A
-// modified peer is removed and re-added, and a re-add defaults to idle; the
-// remote side of an established connection keeps its state and sends no further
-// offers, so a previously active peer parked idle leaves the pair unable to
-// reconnect until the remote's connection expires.
-func TestEngine_ModifiedPeerKeepsActivationState(t *testing.T) {
-	key, err := wgtypes.GeneratePrivateKey()
-	require.NoError(t, err)
-
-	ctx, cancel := context.WithCancel(CtxInitState(context.Background()))
-	t.Cleanup(cancel)
-
-	relayMgr := relayClient.NewManager(ctx, nil, key.PublicKey().String(), iface.DefaultMTU)
-	engine := NewEngine(ctx, cancel, &EngineConfig{
-		WgIfaceName:    "utun103",
-		WgAddr:         wgaddr.MustParseWGAddress("100.64.0.1/24"),
-		WgPrivateKey:   key,
-		WgPort:         33101,
-		MTU:            iface.DefaultMTU,
-		LazyConnection: lazyconn.StateOn,
-	}, EngineServices{
-		SignalClient:   &signal.MockClient{},
-		MgmClient:      &mgmt.MockClient{},
-		RelayManager:   relayMgr,
-		StatusRecorder: peer.NewRecorder("https://mgm"),
-	}, MobileDependency{})
-
-	wgIface := &MockWGIface{
-		NameFunc: func() string { return "utun103" },
-		IsUserspaceBindFunc: func() bool {
-			return false
-		},
-		RemovePeerFunc: func(peerKey string) error {
-			return nil
-		},
-		AddressFunc: func() wgaddr.Address {
-			return wgaddr.Address{
-				IP:      netip.MustParseAddr("10.20.0.1"),
-				Network: netip.MustParsePrefix("10.20.0.0/24"),
-			}
-		},
-		UpdatePeerFunc: func(peerKey string, allowedIps []netip.Prefix, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error {
-			return nil
-		},
-	}
-	engine.wgInterface = wgIface
-	engine.routeManager = routemanager.NewManager(routemanager.ManagerConfig{
-		Context:          ctx,
-		PublicKey:        key.PublicKey().String(),
-		DNSRouteInterval: time.Minute,
-		WGInterface:      engine.wgInterface,
-		StatusRecorder:   engine.statusRecorder,
-		RelayManager:     relayMgr,
-	})
-	require.NoError(t, engine.routeManager.Init())
-	engine.dnsServer = &dns.MockServer{
-		UpdateDNSServerFunc: func(serial uint64, update nbdns.Config) error { return nil },
-	}
-	udpConn, err := net.ListenUDP("udp4", nil)
-	require.NoError(t, err)
-	t.Cleanup(func() {
-		if err := udpConn.Close(); err != nil {
-			t.Errorf("close UDP listener: %v", err)
-		}
-	})
-	engine.udpMux = udpmux.NewUniversalUDPMuxDefault(udpmux.UniversalUDPMuxParams{UDPConn: udpConn, MTU: 1280})
-	engine.ctx = ctx
-	engine.srWatcher = guard.NewSRWatcher(nil, nil, nil, icemaker.Config{})
-	engine.connMgr = NewConnMgr(engine.config, engine.statusRecorder, engine.peerStore, wgIface)
-	engine.connMgr.Start(ctx)
-	t.Cleanup(engine.connMgr.Close)
-
-	// No agent version: not lazy-capable, so the connection opens permanently.
-	activePeer := &mgmtProto.RemotePeerConfig{
-		WgPubKey:   "RRHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=",
-		AllowedIps: []string{"100.64.0.10/24"},
-	}
-	// Lazy-capable, never activated: managed as idle.
-	idlePeer := &mgmtProto.RemotePeerConfig{
-		WgPubKey:     "LLHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=",
-		AllowedIps:   []string{"100.64.0.11/24"},
-		AgentVersion: "development",
-	}
-
-	err = engine.updateNetworkMap(&mgmtProto.NetworkMap{
-		Serial:      1,
-		RemotePeers: []*mgmtProto.RemotePeerConfig{activePeer, idlePeer},
-	})
-	require.NoError(t, err)
-
-	state, err := engine.statusRecorder.GetPeer(activePeer.WgPubKey)
-	require.NoError(t, err)
-	require.Equal(t, peer.StatusConnecting, state.ConnStatus, "peer without lazy support should open a permanent connection")
-
-	state, err = engine.statusRecorder.GetPeer(idlePeer.WgPubKey)
-	require.NoError(t, err)
-	require.Equal(t, peer.StatusIdle, state.ConnStatus, "lazy-capable peer should be managed as idle")
-
-	// The active peer's agent version changes, as when a peer registered over the
-	// API logs in and fills in its meta; the idle peer's allowed IPs change. Both
-	// count as modified and are removed and re-added.
-	err = engine.updateNetworkMap(&mgmtProto.NetworkMap{
-		Serial: 2,
-		RemotePeers: []*mgmtProto.RemotePeerConfig{
-			{
-				WgPubKey:     activePeer.WgPubKey,
-				AllowedIps:   activePeer.AllowedIps,
-				AgentVersion: "development",
-			},
-			{
-				WgPubKey:     idlePeer.WgPubKey,
-				AllowedIps:   []string{"100.64.0.21/24"},
-				AgentVersion: "development",
-			},
-		},
-	})
-	require.NoError(t, err)
-
-	state, err = engine.statusRecorder.GetPeer(activePeer.WgPubKey)
-	require.NoError(t, err)
-	assert.NotEqual(t, peer.StatusIdle, state.ConnStatus, "previously active peer should stay active after a modify")
-
-	state, err = engine.statusRecorder.GetPeer(idlePeer.WgPubKey)
-	require.NoError(t, err)
-	assert.Equal(t, peer.StatusIdle, state.ConnStatus, "previously idle peer should stay idle after a modify")
-
-	// A missing status entry fails the modify before any connection is removed.
-	require.NoError(t, engine.statusRecorder.RemovePeer(activePeer.WgPubKey))
-	err = engine.updateNetworkMap(&mgmtProto.NetworkMap{
-		Serial: 3,
-		RemotePeers: []*mgmtProto.RemotePeerConfig{
-			{
-				WgPubKey:     activePeer.WgPubKey,
-				AllowedIps:   []string{"100.64.0.30/24"},
-				AgentVersion: "development",
-			},
-			{
-				WgPubKey:     idlePeer.WgPubKey,
-				AllowedIps:   []string{"100.64.0.31/24"},
-				AgentVersion: "development",
-			},
-		},
-	})
-	require.ErrorContains(t, err, "get status of modified peer", "a modify with an unavailable peer state should fail")
-
-	activeConn, ok := engine.peerStore.PeerConn(activePeer.WgPubKey)
-	require.True(t, ok, "peer with unavailable state should keep its connection")
-	assert.True(t, compareNetIPLists(activeConn.WgConfig().AllowedIps, activePeer.AllowedIps),
-		"peer with unavailable state should keep its allowed IPs")
-
-	idleConn, ok := engine.peerStore.PeerConn(idlePeer.WgPubKey)
-	require.True(t, ok, "the other modified peer should keep its connection")
-	assert.True(t, compareNetIPLists(idleConn.WgConfig().AllowedIps, []string{"100.64.0.21/24"}),
-		"the other modified peer should keep its allowed IPs")
-}
-
 func TestEngine_UpdateNetworkMapWithRoutes(t *testing.T) {
 	testCases := []struct {
 		name                 string

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.