diff --git a/management/server/account/manager.go b/management/server/account/manager.go index 2e6a11f05..3960008cc 100644 --- a/management/server/account/manager.go +++ b/management/server/account/manager.go @@ -129,9 +129,7 @@ type Manager interface { GetAccountSettings(ctx context.Context, accountID string, userID string) (*types.Settings, error) DeleteSetupKey(ctx context.Context, accountID, userID, keyID string) error UpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) - UpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string) - BufferUpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string, reason types.UpdateReason) - ResolveAffectedPeers(ctx context.Context, s store.Store, accountID string, change affectedpeers.Change) []string + ExpandAndUpdateAffected(ctx context.Context, accountID string, snap *affectedpeers.Snapshot, change affectedpeers.Change) BufferUpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) BuildUserInfosForAccount(ctx context.Context, accountID, initiatorUserID string, accountUsers []*types.User) (map[string]*types.UserInfo, error) SyncUserJWTGroups(ctx context.Context, userAuth auth.UserAuth) error diff --git a/management/server/account/manager_mock.go b/management/server/account/manager_mock.go index dcd9a671f..be104c8ab 100644 --- a/management/server/account/manager_mock.go +++ b/management/server/account/manager_mock.go @@ -123,18 +123,6 @@ func (mr *MockManagerMockRecorder) BufferUpdateAccountPeers(ctx, accountID, reas return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BufferUpdateAccountPeers", reflect.TypeOf((*MockManager)(nil).BufferUpdateAccountPeers), ctx, accountID, reason) } -// BufferUpdateAffectedPeers mocks base method. -func (m *MockManager) BufferUpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string, reason types.UpdateReason) { - m.ctrl.T.Helper() - m.ctrl.Call(m, "BufferUpdateAffectedPeers", ctx, accountID, peerIDs, reason) -} - -// BufferUpdateAffectedPeers indicates an expected call of BufferUpdateAffectedPeers. -func (mr *MockManagerMockRecorder) BufferUpdateAffectedPeers(ctx, accountID, peerIDs, reason interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BufferUpdateAffectedPeers", reflect.TypeOf((*MockManager)(nil).BufferUpdateAffectedPeers), ctx, accountID, peerIDs, reason) -} - // BuildUserInfosForAccount mocks base method. func (m *MockManager) BuildUserInfosForAccount(ctx context.Context, accountID, initiatorUserID string, accountUsers []*types.User) (map[string]*types.UserInfo, error) { m.ctrl.T.Helper() @@ -1650,30 +1638,16 @@ func (mr *MockManagerMockRecorder) UpdateAccountPeers(ctx, accountID, reason int return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateAccountPeers", reflect.TypeOf((*MockManager)(nil).UpdateAccountPeers), ctx, accountID, reason) } -// UpdateAffectedPeers mocks base method. -func (m *MockManager) UpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string) { +// ExpandAndUpdateAffected mocks base method. +func (m *MockManager) ExpandAndUpdateAffected(ctx context.Context, accountID string, snap *affectedpeers.Snapshot, change affectedpeers.Change) { m.ctrl.T.Helper() - m.ctrl.Call(m, "UpdateAffectedPeers", ctx, accountID, peerIDs) + m.ctrl.Call(m, "ExpandAndUpdateAffected", ctx, accountID, snap, change) } -// UpdateAffectedPeers indicates an expected call of UpdateAffectedPeers. -func (mr *MockManagerMockRecorder) UpdateAffectedPeers(ctx, accountID, peerIDs interface{}) *gomock.Call { +// ExpandAndUpdateAffected indicates an expected call of ExpandAndUpdateAffected. +func (mr *MockManagerMockRecorder) ExpandAndUpdateAffected(ctx, accountID, snap, change interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateAffectedPeers", reflect.TypeOf((*MockManager)(nil).UpdateAffectedPeers), ctx, accountID, peerIDs) -} - -// ResolveAffectedPeers mocks base method. -func (m *MockManager) ResolveAffectedPeers(ctx context.Context, s store.Store, accountID string, change affectedpeers.Change) []string { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "ResolveAffectedPeers", ctx, s, accountID, change) - ret0, _ := ret[0].([]string) - return ret0 -} - -// ResolveAffectedPeers indicates an expected call of ResolveAffectedPeers. -func (mr *MockManagerMockRecorder) ResolveAffectedPeers(ctx, s, accountID, change interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolveAffectedPeers", reflect.TypeOf((*MockManager)(nil).ResolveAffectedPeers), ctx, s, accountID, change) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ExpandAndUpdateAffected", reflect.TypeOf((*MockManager)(nil).ExpandAndUpdateAffected), ctx, accountID, snap, change) } // UpdateAccountSettings mocks base method. diff --git a/management/server/affected_peers_coverage_test.go b/management/server/affected_peers_coverage_test.go index a82a66168..56917905f 100644 --- a/management/server/affected_peers_coverage_test.go +++ b/management/server/affected_peers_coverage_test.go @@ -104,7 +104,7 @@ func TestAffectedPeers_DependencyCoverageMatrix(t *testing.T) { ctx := context.Background() change, mustContain, mustExclude := r.build(t, s, ctx) - affected := s.manager.ResolveAffectedPeers(ctx, s.manager.Store, s.accountID, change) + affected := resolveAffected(t, s.manager.Store, s.accountID, change) for _, id := range mustContain { assert.Contains(t, affected, id, "expected peer to be affected") diff --git a/management/server/affected_peers_property_test.go b/management/server/affected_peers_property_test.go index 82dd91cf4..f393465bc 100644 --- a/management/server/affected_peers_property_test.go +++ b/management/server/affected_peers_property_test.go @@ -146,7 +146,11 @@ func TestAffectedPeers_Property_ResolverSupersetsRealChanges(t *testing.T) { resolvedSet := make(map[string]struct{}) resolve := func() { require.NoError(t, s.manager.Store.ExecuteInTransaction(ctx, func(tx store.Store) error { - for _, id := range s.manager.ResolveAffectedPeers(ctx, tx, s.accountID, change) { + snap, err := affectedpeers.Load(ctx, tx, s.accountID, change) + if err != nil { + return err + } + for _, id := range snap.Expand(ctx, s.accountID, change) { resolvedSet[id] = struct{}{} } return nil diff --git a/management/server/affected_peers_querycount_test.go b/management/server/affected_peers_querycount_test.go index 56e9c1bd4..d451a0a29 100644 --- a/management/server/affected_peers_querycount_test.go +++ b/management/server/affected_peers_querycount_test.go @@ -103,8 +103,10 @@ func TestAffectedPeers_QueryCount_NoRedundantFullTableLoads(t *testing.T) { cs := newCountingStore(s.manager.Store) // A group change that exercises policies, routers, resources and the bridge. - affected, err := affectedpeers.Resolve(ctx, cs, s.accountID, affectedpeers.Change{ChangedGroupIDs: []string{s.sourceGroupID}}) + change := affectedpeers.Change{ChangedGroupIDs: []string{s.sourceGroupID}} + snap, err := affectedpeers.Load(ctx, cs, s.accountID, change) require.NoError(t, err) + affected := snap.Expand(ctx, s.accountID, change) assert.Contains(t, affected, s.routerPeerID, "bridge must still resolve the routing peer") for _, name := range []string{"policies", "routes", "nameservers", "dnssettings", "routers", "resources"} { @@ -126,7 +128,7 @@ func TestAffectedPeers_QueryCount_NarrowChangeSkipsLoads(t *testing.T) { // A bare network change drives only the router->source bridge: routers and // resources are needed, but routes/nameservers/dnssettings/services are not. - _, err := affectedpeers.Resolve(ctx, cs, s.accountID, affectedpeers.Change{Networks: []*networkTypes.Network{{ID: s.networkID}}}) + _, err := affectedpeers.Load(ctx, cs, s.accountID, affectedpeers.Change{Networks: []*networkTypes.Network{{ID: s.networkID}}}) require.NoError(t, err) assert.Equal(t, 0, cs.count("routes"), "routes must not be loaded for a network-only change") diff --git a/management/server/affected_peers_router_paths_test.go b/management/server/affected_peers_router_paths_test.go index 22db1bee6..11313c387 100644 --- a/management/server/affected_peers_router_paths_test.go +++ b/management/server/affected_peers_router_paths_test.go @@ -16,11 +16,21 @@ import ( ) func (s *routerScenario) resolveGroupChangeAffected(ctx context.Context, changedGroupIDs []string) []string { - return s.manager.ResolveAffectedPeers(ctx, s.manager.Store, s.accountID, affectedpeers.Change{ChangedGroupIDs: changedGroupIDs}) + change := affectedpeers.Change{ChangedGroupIDs: changedGroupIDs} + snap, err := affectedpeers.Load(ctx, s.manager.Store, s.accountID, change) + if err != nil { + return nil + } + return snap.Expand(ctx, s.accountID, change) } func (s *routerScenario) resolvePeerChangeAffected(ctx context.Context, changedPeerIDs []string) []string { - return s.manager.resolveAffectedPeersForPeerChanges(ctx, s.manager.Store, s.accountID, changedPeerIDs) + change := affectedpeers.Change{ChangedPeerIDs: changedPeerIDs} + snap, err := affectedpeers.Load(ctx, s.manager.Store, s.accountID, change) + if err != nil { + return nil + } + return snap.Expand(ctx, s.accountID, change) } func TestAffectedPeers_GroupChange_SourceGroupMembership_RefreshesRoutingPeer_DirectRouter(t *testing.T) { diff --git a/management/server/affected_peers_router_test.go b/management/server/affected_peers_router_test.go index 696beb597..dc064e787 100644 --- a/management/server/affected_peers_router_test.go +++ b/management/server/affected_peers_router_test.go @@ -187,7 +187,12 @@ func peerToResourcePolicyByResource(sourceGroupID, resourceID string) *types.Pol // resolvePolicyAffected mirrors SavePolicy's resolution: resolve the affected // peers for the given policy. func (s *routerScenario) resolvePolicyAffected(ctx context.Context, policy *types.Policy) []string { - return s.manager.ResolveAffectedPeers(ctx, s.manager.Store, s.accountID, affectedpeers.Change{Policies: []*types.Policy{policy}}) + change := affectedpeers.Change{Policies: []*types.Policy{policy}} + snap, err := affectedpeers.Load(ctx, s.manager.Store, s.accountID, change) + if err != nil { + return nil + } + return snap.Expand(ctx, s.accountID, change) } func TestAffectedPeers_SourcePeer_DirectRouter(t *testing.T) { diff --git a/management/server/affected_peers_test.go b/management/server/affected_peers_test.go index 6d49002af..1a9d2d5d9 100644 --- a/management/server/affected_peers_test.go +++ b/management/server/affected_peers_test.go @@ -23,6 +23,16 @@ import ( "github.com/netbirdio/netbird/route" ) +// resolveAffected is a test helper for the resolver's Load+Expand, used where a +// test asserts on the fully expanded affected peer set. +func resolveAffected(t *testing.T, s store.Store, accountID string, change affectedpeers.Change) []string { + t.Helper() + ctx := context.Background() + snap, err := affectedpeers.Load(ctx, s, accountID, change) + require.NoError(t, err) + return snap.Expand(ctx, accountID, change) +} + // Thin test adapters over affectedpeers.Collect, preserving the (groups, peers) // shape these tests assert on after the resolver was unified. func collectGroupChangeAffectedGroups(ctx context.Context, s store.Store, accountID string, changedGroupIDs []string) ([]string, []string) { diff --git a/management/server/affectedpeers/resolver.go b/management/server/affectedpeers/resolver.go index 118f5b86d..32ccb67d8 100644 --- a/management/server/affectedpeers/resolver.go +++ b/management/server/affectedpeers/resolver.go @@ -120,6 +120,12 @@ type Change struct { Networks []*networkTypes.Network PostureCheckIDs []string + // DistributionGroupIDs are groups whose members are directly affected, with no + // dependency walk — the change distributes config to the groups' member peers + // only (nameserver groups, DNS DisabledManagementGroups), not through the + // policy/route reachability graph. Pass old∪new so both states refresh. + DistributionGroupIDs []string + // RemovedPeersByGroup: peers that left a group, keyed by that group. They are no // longer in the group's member index but still lose its reachability, so they are // folded in — but only when the group is linked (an unlinked group has no map @@ -136,6 +142,7 @@ func (c Change) isEmpty() bool { len(c.Resources) == 0 && len(c.Networks) == 0 && len(c.PostureCheckIDs) == 0 && + len(c.DistributionGroupIDs) == 0 && len(c.RemovedPeersByGroup) == 0 } @@ -147,25 +154,12 @@ func (snap *Snapshot) Expand(ctx context.Context, accountID string, c Change) [] return nil } r := newResolver(ctx, snap, accountID, c) - log.WithContext(ctx).Tracef("affectedpeers expand start: account=%s changedGroups=%v changedPeers=%v policies=%d routes=%d routers=%d resources=%d networks=%d postureChecks=%v", - accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, len(c.Policies), len(c.Routes), len(c.Routers), len(c.Resources), len(c.Networks), c.PostureCheckIDs) + log.WithContext(ctx).Tracef("affectedpeers expand start: account=%s changedGroups=%v changedPeers=%v policies=%d routes=%d routers=%d resources=%d networks=%d postureChecks=%v distributionGroups=%v", + accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, len(c.Policies), len(c.Routes), len(c.Routers), len(c.Resources), len(c.Networks), c.PostureCheckIDs, c.DistributionGroupIDs) r.walk() return r.expand() } -// Resolve does Load+Expand in one call, for callers not inside a tx. Tx-bound -// callers should use Load (in-tx) + Snapshot.Expand (after commit) instead. -func Resolve(ctx context.Context, s store.Store, accountID string, c Change) ([]string, error) { - if c.isEmpty() { - return nil, nil - } - snap, err := Load(ctx, s, accountID, c) - if err != nil { - return nil, err - } - return snap.Expand(ctx, accountID, c), nil -} - // Collect returns the affected group and direct-peer IDs without expanding groups // to members. Test-only introspection; use Resolve otherwise. func Collect(ctx context.Context, s store.Store, accountID string, c Change) (groupIDs []string, directPeerIDs []string) { @@ -224,6 +218,11 @@ func (r *resolver) walk() { r.collectFromExplicitNetworks(r.change.Networks) r.collectFromPostureChecks(r.change.PostureCheckIDs) + // Distribution groups (nameserver/DNS) affect only their member peers: fold them + // straight into groupSet so expand() maps them to members, without the policy/ + // route walk that changedGroupSet would trigger. + addAll(r.groupSet, r.change.DistributionGroupIDs) + if len(r.changedGroupSet) > 0 || len(r.changedPeerSet) > 0 { r.collectFromPolicies() r.collectFromRoutes() diff --git a/management/server/dns.go b/management/server/dns.go index 027860c98..612c8ecba 100644 --- a/management/server/dns.go +++ b/management/server/dns.go @@ -8,6 +8,7 @@ import ( nbdns "github.com/netbirdio/netbird/dns" "github.com/netbirdio/netbird/management/server/activity" + "github.com/netbirdio/netbird/management/server/affectedpeers" "github.com/netbirdio/netbird/management/server/permissions/modules" "github.com/netbirdio/netbird/management/server/permissions/operations" "github.com/netbirdio/netbird/management/server/store" @@ -48,7 +49,8 @@ func (am *DefaultAccountManager) SaveDNSSettings(ctx context.Context, accountID } var eventsToStore []func() - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { if err = validateDNSSettings(ctx, transaction, accountID, dnsSettingsToSave); err != nil { @@ -70,8 +72,10 @@ func (am *DefaultAccountManager) SaveDNSSettings(ctx context.Context, accountID return err } - allGroups := slices.Concat(addedGroups, removedGroups) - affectedPeerIDs = am.resolvePeerIDs(ctx, transaction, accountID, allGroups, nil) + change = affectedpeers.Change{DistributionGroupIDs: slices.Concat(addedGroups, removedGroups)} + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) }) @@ -83,12 +87,7 @@ func (am *DefaultAccountManager) SaveDNSSettings(ctx context.Context, accountID storeEvent() } - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("SaveDNSSettings: updating %d affected peers: %v", len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("SaveDNSSettings: no affected peers") - } + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } diff --git a/management/server/group.go b/management/server/group.go index 32dc6252f..51c558c08 100644 --- a/management/server/group.go +++ b/management/server/group.go @@ -118,7 +118,7 @@ func (am *DefaultAccountManager) CreateGroup(ctx context.Context, accountID, use storeEvent() } - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -187,7 +187,7 @@ func (am *DefaultAccountManager) UpdateGroup(ctx context.Context, accountID, use storeEvent() } - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -492,7 +492,7 @@ func (am *DefaultAccountManager) DeleteGroups(ctx context.Context, accountID, us am.StoreEvent(ctx, userID, group.ID, accountID, activity.GroupDeleted, group.EventMeta()) } - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return allErrors } @@ -542,7 +542,7 @@ func (am *DefaultAccountManager) GroupAddPeer(ctx context.Context, accountID, gr return err } - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -578,7 +578,7 @@ func (am *DefaultAccountManager) GroupAddResource(ctx context.Context, accountID return err } - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -613,7 +613,7 @@ func (am *DefaultAccountManager) GroupDeletePeer(ctx context.Context, accountID, return err } - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -651,7 +651,7 @@ func (am *DefaultAccountManager) GroupDeleteResource(ctx context.Context, accoun return err } - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } diff --git a/management/server/mock_server/account_mock.go b/management/server/mock_server/account_mock.go index 894f7ce1e..4d03526d2 100644 --- a/management/server/mock_server/account_mock.go +++ b/management/server/mock_server/account_mock.go @@ -133,9 +133,7 @@ type MockAccountManager struct { AllowSyncFunc func(string, uint64) bool UpdateAccountPeersFunc func(ctx context.Context, accountID string, reason types.UpdateReason) - UpdateAffectedPeersFunc func(ctx context.Context, accountID string, peerIDs []string) - BufferUpdateAffectedPeersFunc func(ctx context.Context, accountID string, peerIDs []string, reason types.UpdateReason) - ResolveAffectedPeersFunc func(ctx context.Context, s store.Store, accountID string, change affectedpeers.Change) []string + ExpandAndUpdateAffectedFunc func(ctx context.Context, accountID string, snap *affectedpeers.Snapshot, change affectedpeers.Change) BufferUpdateAccountPeersFunc func(ctx context.Context, accountID string, reason types.UpdateReason) RecalculateNetworkMapCacheFunc func(ctx context.Context, accountId string) error @@ -213,22 +211,9 @@ func (am *MockAccountManager) UpdateAccountPeers(ctx context.Context, accountID } } -func (am *MockAccountManager) UpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string) { - if am.UpdateAffectedPeersFunc != nil { - am.UpdateAffectedPeersFunc(ctx, accountID, peerIDs) - } -} - -func (am *MockAccountManager) ResolveAffectedPeers(ctx context.Context, s store.Store, accountID string, change affectedpeers.Change) []string { - if am.ResolveAffectedPeersFunc != nil { - return am.ResolveAffectedPeersFunc(ctx, s, accountID, change) - } - return nil -} - -func (am *MockAccountManager) BufferUpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string, reason types.UpdateReason) { - if am.BufferUpdateAffectedPeersFunc != nil { - am.BufferUpdateAffectedPeersFunc(ctx, accountID, peerIDs, reason) +func (am *MockAccountManager) ExpandAndUpdateAffected(ctx context.Context, accountID string, snap *affectedpeers.Snapshot, change affectedpeers.Change) { + if am.ExpandAndUpdateAffectedFunc != nil { + am.ExpandAndUpdateAffectedFunc(ctx, accountID, snap, change) } } diff --git a/management/server/nameserver.go b/management/server/nameserver.go index 18e8f54ea..b9cebf726 100644 --- a/management/server/nameserver.go +++ b/management/server/nameserver.go @@ -9,10 +9,10 @@ import ( "unicode/utf8" "github.com/rs/xid" - log "github.com/sirupsen/logrus" nbdns "github.com/netbirdio/netbird/dns" "github.com/netbirdio/netbird/management/server/activity" + "github.com/netbirdio/netbird/management/server/affectedpeers" "github.com/netbirdio/netbird/management/server/permissions/modules" "github.com/netbirdio/netbird/management/server/permissions/operations" "github.com/netbirdio/netbird/management/server/store" @@ -59,7 +59,8 @@ func (am *DefaultAccountManager) CreateNameServerGroup(ctx context.Context, acco SearchDomainsEnabled: searchDomainEnabled, } - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{DistributionGroupIDs: newNSGroup.Groups} err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { if err = validateNameServerGroup(ctx, transaction, accountID, newNSGroup); err != nil { @@ -70,7 +71,9 @@ func (am *DefaultAccountManager) CreateNameServerGroup(ctx context.Context, acco return err } - affectedPeerIDs = am.resolvePeerIDs(ctx, transaction, accountID, newNSGroup.Groups, nil) + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) }) @@ -80,12 +83,7 @@ func (am *DefaultAccountManager) CreateNameServerGroup(ctx context.Context, acco am.StoreEvent(ctx, userID, newNSGroup.ID, accountID, activity.NameserverGroupCreated, newNSGroup.EventMeta()) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("CreateNameServerGroup %s: updating %d affected peers: %v", newNSGroup.ID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("CreateNameServerGroup %s: no affected peers", newNSGroup.ID) - } + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return newNSGroup.Copy(), nil } @@ -104,7 +102,8 @@ func (am *DefaultAccountManager) SaveNameServerGroup(ctx context.Context, accoun return status.NewPermissionDeniedError() } - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { oldNSGroup, err := transaction.GetNameServerGroupByID(ctx, store.LockingStrengthNone, accountID, nsGroupToSave.ID) @@ -121,8 +120,10 @@ func (am *DefaultAccountManager) SaveNameServerGroup(ctx context.Context, accoun return err } - allGroups := slices.Concat(nsGroupToSave.Groups, oldNSGroup.Groups) - affectedPeerIDs = am.resolvePeerIDs(ctx, transaction, accountID, allGroups, nil) + change = affectedpeers.Change{DistributionGroupIDs: slices.Concat(nsGroupToSave.Groups, oldNSGroup.Groups)} + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) }) @@ -132,12 +133,7 @@ func (am *DefaultAccountManager) SaveNameServerGroup(ctx context.Context, accoun am.StoreEvent(ctx, userID, nsGroupToSave.ID, accountID, activity.NameserverGroupUpdated, nsGroupToSave.EventMeta()) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("SaveNameServerGroup %s: updating %d affected peers: %v", nsGroupToSave.ID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("SaveNameServerGroup %s: no affected peers", nsGroupToSave.ID) - } + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -153,7 +149,8 @@ func (am *DefaultAccountManager) DeleteNameServerGroup(ctx context.Context, acco } var nsGroup *nbdns.NameServerGroup - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { nsGroup, err = transaction.GetNameServerGroupByID(ctx, store.LockingStrengthUpdate, accountID, nsGroupID) @@ -161,7 +158,11 @@ func (am *DefaultAccountManager) DeleteNameServerGroup(ctx context.Context, acco return err } - affectedPeerIDs = am.resolvePeerIDs(ctx, transaction, accountID, nsGroup.Groups, nil) + // Load before delete: the post-delete state no longer references the groups. + change = affectedpeers.Change{DistributionGroupIDs: nsGroup.Groups} + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } if err = transaction.DeleteNameServerGroup(ctx, accountID, nsGroupID); err != nil { return err @@ -175,12 +176,7 @@ func (am *DefaultAccountManager) DeleteNameServerGroup(ctx context.Context, acco am.StoreEvent(ctx, userID, nsGroup.ID, accountID, activity.NameserverGroupDeleted, nsGroup.EventMeta()) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("DeleteNameServerGroup %s: updating %d affected peers: %v", nsGroupID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("DeleteNameServerGroup %s: no affected peers", nsGroupID) - } + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } diff --git a/management/server/networks/manager.go b/management/server/networks/manager.go index 7a339d1a5..efd35c29f 100644 --- a/management/server/networks/manager.go +++ b/management/server/networks/manager.go @@ -5,7 +5,6 @@ import ( "fmt" "github.com/rs/xid" - log "github.com/sirupsen/logrus" "github.com/netbirdio/netbird/management/server/account" "github.com/netbirdio/netbird/management/server/activity" @@ -203,13 +202,7 @@ func (m *managerImpl) DeleteNetwork(ctx context.Context, accountID, userID, netw event() } - affectedPeerIDs := snap.Expand(ctx, accountID, change) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("DeleteNetwork %s: updating %d affected peers: %v", networkID, len(affectedPeerIDs), affectedPeerIDs) - go m.accountManager.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("DeleteNetwork %s: no affected peers", networkID) - } + go m.accountManager.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } diff --git a/management/server/networks/resources/manager.go b/management/server/networks/resources/manager.go index 2a5f10ebd..7cd766634 100644 --- a/management/server/networks/resources/manager.go +++ b/management/server/networks/resources/manager.go @@ -130,13 +130,7 @@ func (m *managerImpl) CreateResource(ctx context.Context, userID string, resourc event() } - affectedPeerIDs := snap.Expand(ctx, resource.AccountID, change) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("CreateResource %s: updating %d affected peers: %v", resource.ID, len(affectedPeerIDs), affectedPeerIDs) - go m.accountManager.UpdateAffectedPeers(ctx, resource.AccountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("CreateResource %s: no affected peers", resource.ID) - } + go m.accountManager.ExpandAndUpdateAffected(ctx, resource.AccountID, snap, change) return resource, nil } @@ -305,13 +299,7 @@ func (m *managerImpl) UpdateResource(ctx context.Context, userID string, resourc } }() - affectedPeerIDs := snap.Expand(ctx, resource.AccountID, change) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("UpdateResource %s: updating %d affected peers: %v", resource.ID, len(affectedPeerIDs), affectedPeerIDs) - go m.accountManager.UpdateAffectedPeers(ctx, resource.AccountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("UpdateResource %s: no affected peers", resource.ID) - } + go m.accountManager.ExpandAndUpdateAffected(ctx, resource.AccountID, snap, change) return resource, nil } @@ -413,13 +401,7 @@ func (m *managerImpl) DeleteResource(ctx context.Context, accountID, userID, net event() } - affectedPeerIDs := snap.Expand(ctx, accountID, change) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("DeleteResource %s: updating %d affected peers: %v", resourceID, len(affectedPeerIDs), affectedPeerIDs) - go m.accountManager.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("DeleteResource %s: no affected peers", resourceID) - } + go m.accountManager.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } diff --git a/management/server/networks/routers/manager.go b/management/server/networks/routers/manager.go index aca6e5517..7700422d8 100644 --- a/management/server/networks/routers/manager.go +++ b/management/server/networks/routers/manager.go @@ -6,7 +6,6 @@ import ( "fmt" "github.com/rs/xid" - log "github.com/sirupsen/logrus" "github.com/netbirdio/netbird/management/server/account" "github.com/netbirdio/netbird/management/server/activity" @@ -127,13 +126,7 @@ func (m *managerImpl) CreateRouter(ctx context.Context, userID string, router *t m.accountManager.StoreEvent(ctx, userID, router.ID, router.AccountID, activity.NetworkRouterCreated, router.EventMeta(network)) - affectedPeerIDs := snap.Expand(ctx, router.AccountID, change) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("CreateRouter %s: updating %d affected peers: %v", router.ID, len(affectedPeerIDs), affectedPeerIDs) - go m.accountManager.UpdateAffectedPeers(ctx, router.AccountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("CreateRouter %s: no affected peers", router.ID) - } + go m.accountManager.ExpandAndUpdateAffected(ctx, router.AccountID, snap, change) return router, nil } @@ -182,13 +175,7 @@ func (m *managerImpl) UpdateRouter(ctx context.Context, userID string, router *t m.accountManager.StoreEvent(ctx, userID, router.ID, router.AccountID, activity.NetworkRouterUpdated, router.EventMeta(network)) - affectedPeerIDs := snap.Expand(ctx, router.AccountID, change) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("UpdateRouter %s: updating %d affected peers: %v", router.ID, len(affectedPeerIDs), affectedPeerIDs) - go m.accountManager.UpdateAffectedPeers(ctx, router.AccountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("UpdateRouter %s: no affected peers", router.ID) - } + go m.accountManager.ExpandAndUpdateAffected(ctx, router.AccountID, snap, change) return router, nil } @@ -274,13 +261,7 @@ func (m *managerImpl) DeleteRouter(ctx context.Context, accountID, userID, netwo event() - affectedPeerIDs := snap.Expand(ctx, accountID, change) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("DeleteRouter %s: updating %d affected peers: %v", routerID, len(affectedPeerIDs), affectedPeerIDs) - go m.accountManager.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("DeleteRouter %s: no affected peers", routerID) - } + go m.accountManager.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } diff --git a/management/server/peer.go b/management/server/peer.go index fc3660a91..996642096 100644 --- a/management/server/peer.go +++ b/management/server/peer.go @@ -1436,32 +1436,19 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account _ = am.networkMapController.UpdateAccountPeers(ctx, accountID, reason) } -// UpdateAffectedPeers updates only the specified peers that belong to an account. -func (am *DefaultAccountManager) UpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string) { - ctx = context.WithoutCancel(ctx) - log.WithContext(ctx).Tracef("UpdateAffectedPeers: %d peers for account %s", len(peerIDs), accountID) - _ = am.networkMapController.UpdateAffectedPeers(ctx, accountID, peerIDs) +// ExpandAndUpdateAffected expands a Snapshot (loaded INSIDE the now-committed +// transaction) into the affected peers and dispatches the network-map refresh. +// Pure in-memory work plus dispatch, so it runs AFTER commit — the fan-out walk +// never holds the write lock, over the consistent in-tx snapshot. Exported so the +// networks sub-package managers (which hold only account.Manager) share it. +func (am *DefaultAccountManager) ExpandAndUpdateAffected(ctx context.Context, accountID string, snap *affectedpeers.Snapshot, change affectedpeers.Change) { + am.dispatchAffected(ctx, accountID, []*affectedpeers.Snapshot{snap}, []affectedpeers.Change{change}) } -// expandAndUpdateAffected expands a Snapshot (loaded INSIDE the now-committed -// transaction) into the affected peers and dispatches the update. It is pure -// in-memory work plus the dispatch, so it must run AFTER the transaction commits — -// the fan-out walk never holds the write lock, and the data is the consistent -// in-tx snapshot. directlyAffected are peers whose OWN map changed and that the -// walk cannot rediscover post-mutation (notably peers removed from a group); they -// are unioned into the result. -func (am *DefaultAccountManager) expandAndUpdateAffected(ctx context.Context, accountID string, snap *affectedpeers.Snapshot, change affectedpeers.Change, directlyAffected ...string) { - if snap == nil { - return - } - affectedPeerIDs := unionStrings(snap.Expand(ctx, accountID, change), directlyAffected) - am.updateAffectedPeerIDs(ctx, accountID, affectedPeerIDs) -} - -// dispatchAffected expands a set of (snapshot, change) pairs collected across -// several transactions, unions their affected peers, and dispatches one update. -// Used by batch operations that commit per-item transactions but want a single -// affected-peers pass — each snapshot is still loaded inside its own transaction. +// dispatchAffected expands one or more (snapshot, change) pairs — collected across +// one or several transactions — unions their affected peers, and dispatches a +// single network-map refresh. Each snapshot must already be loaded inside its +// transaction; this runs AFTER commit (pure in-memory + dispatch). func (am *DefaultAccountManager) dispatchAffected(ctx context.Context, accountID string, snaps []*affectedpeers.Snapshot, changes []affectedpeers.Change) { var lists [][]string for i, snap := range snaps { @@ -1470,18 +1457,15 @@ func (am *DefaultAccountManager) dispatchAffected(ctx context.Context, accountID } lists = append(lists, snap.Expand(ctx, accountID, changes[i])) } - am.updateAffectedPeerIDs(ctx, accountID, unionStrings(lists...)) -} -// updateAffectedPeerIDs dispatches a network-map refresh for the given peers, or -// logs that there are none. Shared dispatch tail for the affected-peers helpers. -func (am *DefaultAccountManager) updateAffectedPeerIDs(ctx context.Context, accountID string, affectedPeerIDs []string) { - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("updating %d affected peers for account %s: %v", len(affectedPeerIDs), accountID, affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { + affectedPeerIDs := unionStrings(lists...) + if len(affectedPeerIDs) == 0 { log.WithContext(ctx).Tracef("no affected peers for account %s", accountID) + return } + + log.WithContext(ctx).Debugf("updating %d affected peers for account %s: %v", len(affectedPeerIDs), accountID, affectedPeerIDs) + _ = am.networkMapController.UpdateAffectedPeers(context.WithoutCancel(ctx), accountID, affectedPeerIDs) } // unionStrings concatenates the given string lists into one deduplicated slice, @@ -1501,41 +1485,6 @@ func unionStrings(lists ...[]string) []string { return out } -// resolvePeerIDs resolves group IDs and direct peer IDs into a deduplicated peer ID list. -func (am *DefaultAccountManager) resolvePeerIDs(ctx context.Context, s store.Store, accountID string, groupIDs []string, directPeerIDs []string) []string { - peerIDs, err := s.GetPeerIDsByGroups(ctx, accountID, groupIDs) - if err != nil { - log.WithContext(ctx).Errorf("failed to resolve peer IDs by groups: %v", err) - return nil - } - - if len(directPeerIDs) == 0 { - log.WithContext(ctx).Tracef("resolvePeerIDs: groups=%v -> %d peers: %v", groupIDs, len(peerIDs), peerIDs) - return peerIDs - } - - peerIDs = unionStrings(peerIDs, directPeerIDs) - log.WithContext(ctx).Tracef("resolvePeerIDs: groups=%v + directPeers=%v -> %d peers: %v", groupIDs, directPeerIDs, len(peerIDs), peerIDs) - return peerIDs -} - -// BufferUpdateAffectedPeers accumulates peer IDs and flushes them after the buffer interval. -func (am *DefaultAccountManager) BufferUpdateAffectedPeers(ctx context.Context, accountID string, peerIDs []string, reason types.UpdateReason) { - _ = am.networkMapController.BufferUpdateAffectedPeers(ctx, accountID, peerIDs, reason) -} - -// ResolveAffectedPeers resolves a description of what changed into the peer IDs -// whose network map may have changed. It is the single entry point shared by the -// server package and the networks managers (via the account.Manager interface). -func (am *DefaultAccountManager) ResolveAffectedPeers(ctx context.Context, s store.Store, accountID string, change affectedpeers.Change) []string { - peerIDs, err := affectedpeers.Resolve(ctx, s, accountID, change) - if err != nil { - log.WithContext(ctx).Errorf("failed to resolve affected peers: %v", err) - return nil - } - return peerIDs -} - // affectedPeerIDsFromNetworkMap returns the peer IDs referenced by a peer's // network map (its connected and offline peers, which include routing and proxy // peers), excluding the peer itself. For a freshly added peer these are, by ACL @@ -1563,13 +1512,18 @@ func affectedPeerIDsFromNetworkMap(nmap *types.NetworkMap, selfPeerID string) [] return ids } -// resolveAffectedPeersForPeerChanges resolves changed peer IDs into the full set -// of affected peer IDs. The resolver derives each peer's group memberships during -// the walk, so the caller passes only the changed peer IDs. +// resolveAffectedPeersForPeerChanges loads a snapshot and expands it for a peer +// change. The graph is unchanged by these paths, so it runs out of the mutating +// transaction (after commit); the resolver derives the peers' group memberships +// during the walk, so the caller passes only the changed peer IDs. func (am *DefaultAccountManager) resolveAffectedPeersForPeerChanges(ctx context.Context, s store.Store, accountID string, changedPeerIDs []string) []string { - return am.ResolveAffectedPeers(ctx, s, accountID, affectedpeers.Change{ - ChangedPeerIDs: changedPeerIDs, - }) + change := affectedpeers.Change{ChangedPeerIDs: changedPeerIDs} + snap, err := affectedpeers.Load(ctx, s, accountID, change) + if err != nil { + log.WithContext(ctx).Errorf("failed to load snapshot for affected peers: %v", err) + return nil + } + return snap.Expand(ctx, accountID, change) } func (am *DefaultAccountManager) BufferUpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) { diff --git a/management/server/policy.go b/management/server/policy.go index a39ee0d67..187c879cb 100644 --- a/management/server/policy.go +++ b/management/server/policy.go @@ -99,7 +99,7 @@ func (am *DefaultAccountManager) SavePolicy(ctx context.Context, accountID, user am.StoreEvent(ctx, userID, policy.ID, accountID, action, policy.EventMeta()) - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return policy, nil } @@ -142,7 +142,7 @@ func (am *DefaultAccountManager) DeletePolicy(ctx context.Context, accountID, po am.StoreEvent(ctx, userID, policyID, accountID, activity.PolicyRemoved, policy.EventMeta()) - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } diff --git a/management/server/posture_checks.go b/management/server/posture_checks.go index 54009cdd7..1d962438c 100644 --- a/management/server/posture_checks.go +++ b/management/server/posture_checks.go @@ -77,7 +77,7 @@ func (am *DefaultAccountManager) SavePostureChecks(ctx context.Context, accountI am.StoreEvent(ctx, userID, postureChecks.ID, accountID, action, postureChecks.EventMeta()) - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return postureChecks, nil } diff --git a/management/server/route.go b/management/server/route.go index 31d2f8108..08e1489b2 100644 --- a/management/server/route.go +++ b/management/server/route.go @@ -192,7 +192,7 @@ func (am *DefaultAccountManager) CreateRoute(ctx context.Context, accountID stri am.StoreEvent(ctx, userID, string(newRoute.ID), accountID, activity.RouteCreated, newRoute.EventMeta()) - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return newRoute, nil } @@ -240,7 +240,7 @@ func (am *DefaultAccountManager) SaveRoute(ctx context.Context, accountID, userI am.StoreEvent(ctx, userID, string(routeToSave.ID), accountID, activity.RouteUpdated, routeToSave.EventMeta()) - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -283,7 +283,7 @@ func (am *DefaultAccountManager) DeleteRoute(ctx context.Context, accountID stri am.StoreEvent(ctx, userID, string(rt.ID), accountID, activity.RouteRemoved, rt.EventMeta()) - am.expandAndUpdateAffected(ctx, accountID, snap, change) + am.ExpandAndUpdateAffected(ctx, accountID, snap, change) return nil }