sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-07-23 17:01:29 +02:00
Code Issues Packages Projects Releases Wiki Activity

simplify code

Browse Source
This commit is contained in:
pascal
2026-06-09 18:21:32 +02:00
parent 9d5f3a7b86
commit 20d95c6323
20 changed files with 141 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
management/server/account/manager.go
View File

@@ -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

38
management/server/account/manager_mock.go
View File

@@ -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.

2
management/server/affected_peers_coverage_test.go
View File

@@ -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")

6
management/server/affected_peers_property_test.go
View File

@@ -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

6
management/server/affected_peers_querycount_test.go
View File

@@ -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")

14
management/server/affected_peers_router_paths_test.go
View File

@@ -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) {

7
management/server/affected_peers_router_test.go
View File

@@ -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) {

10
management/server/affected_peers_test.go
View File

@@ -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) {

29
management/server/affectedpeers/resolver.go
View File

@@ -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()

17
management/server/dns.go
View File

@@ -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
}

14
management/server/group.go
View File

@@ -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
}

23
management/server/mock_server/account_mock.go
View File

@@ -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)
}
}

48
management/server/nameserver.go
View File

@@ -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
}

9
management/server/networks/manager.go
View File

@@ -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
}

24
management/server/networks/resources/manager.go
View File

@@ -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
}

25
management/server/networks/routers/manager.go
View File

@@ -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
}

102
management/server/peer.go
View File

@@ -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) {

4
management/server/policy.go
View File

@@ -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
}

2
management/server/posture_checks.go
View File

@@ -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
}

6
management/server/route.go
View File

@@ -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
}