From a2b039921240d70a14e7691e159ff115ba923751 Mon Sep 17 00:00:00 2001 From: pascal Date: Fri, 5 Jun 2026 13:34:52 +0200 Subject: [PATCH] load snapshots and handle before-after states --- .../server/affected_peers_querycount_test.go | 36 ++ management/server/affectedpeers/resolver.go | 344 +++++++++++------- management/server/group.go | 141 +++---- management/server/networks/manager.go | 10 +- .../server/networks/resources/manager.go | 35 +- management/server/networks/routers/manager.go | 56 ++- management/server/peer.go | 47 ++- management/server/policy.go | 31 +- management/server/posture_checks.go | 18 +- management/server/route.go | 47 ++- management/server/user.go | 16 +- 11 files changed, 489 insertions(+), 292 deletions(-) diff --git a/management/server/affected_peers_querycount_test.go b/management/server/affected_peers_querycount_test.go index f37be1243..079d548c5 100644 --- a/management/server/affected_peers_querycount_test.go +++ b/management/server/affected_peers_querycount_test.go @@ -43,6 +43,16 @@ func (c *countingStore) count(name string) int { return c.counts[name] } +func (c *countingStore) total() int { + c.mu.Lock() + defer c.mu.Unlock() + n := 0 + for _, v := range c.counts { + n += v + } + return n +} + func (c *countingStore) GetAccountPolicies(ctx context.Context, ls store.LockingStrength, accountID string) ([]*types.Policy, error) { c.bump("policies") return c.Store.GetAccountPolicies(ctx, ls, accountID) @@ -123,3 +133,29 @@ func TestAffectedPeers_QueryCount_NarrowChangeSkipsLoads(t *testing.T) { assert.Equal(t, 0, cs.count("dnssettings"), "dnssettings must not be loaded for a network-only change") assert.Equal(t, 0, cs.count("services"), "services must not be loaded for a network-only change") } + +// TestAffectedPeers_QueryCount_ExpandReadsNothing is the core invariant of the +// Load/Expand split: Load (run inside the transaction) does all store reads; +// Expand (run after commit) must touch the store ZERO times, so it never holds +// the write lock and never reads post-commit state. +func TestAffectedPeers_QueryCount_ExpandReadsNothing(t *testing.T) { + s := setupRouterScenario(t, true) + ctx := context.Background() + + _, err := s.manager.SavePolicy(ctx, s.accountID, userID, peerToResourcePolicyByGroup(s.sourceGroupID, s.resourceGroupID), true) + require.NoError(t, err) + + change := affectedpeers.Change{ChangedGroupIDs: []string{s.sourceGroupID}} + + cs := newCountingStore(s.manager.Store) + snap, err := affectedpeers.Load(ctx, cs, s.accountID, change) + require.NoError(t, err) + require.Greater(t, cs.total(), 0, "Load must read the store") + + // Any store access during Expand would increment the same counter. Expand + // operates purely on the snapshot, so the count must not move. + readsAfterLoad := cs.total() + affected := snap.Expand(ctx, s.accountID, change) + assert.Contains(t, affected, s.routerPeerID, "Expand must still produce the affected peers from the snapshot") + assert.Equal(t, readsAfterLoad, cs.total(), "Expand must perform zero store reads — it operates purely on the loaded snapshot") +} diff --git a/management/server/affectedpeers/resolver.go b/management/server/affectedpeers/resolver.go index 333df5ad7..917c91c6a 100644 --- a/management/server/affectedpeers/resolver.go +++ b/management/server/affectedpeers/resolver.go @@ -1,3 +1,18 @@ +// Package affectedpeers computes the set of peers whose network map may have +// changed as the result of an account change, so only those peers are refreshed +// instead of the whole account. +// +// Resolution is split into two phases so the expensive dependency walk never +// holds a write transaction open: +// - Load reads the account collections it needs from the store. Call it INSIDE +// the mutating transaction, so the data is consistent and read under the tx. +// For deletes/removals, Load (or the captured Change) must run while the old +// state still exists, since the post-commit store can no longer reach it. +// - Snapshot.Expand walks the loaded data in memory and returns the affected +// peer IDs. It performs NO store access, so it is run AFTER the tx commits. +// +// The resolver never consults an object's Enabled flag: toggling Enabled is +// itself a change the affected peers must observe. package affectedpeers import ( @@ -5,6 +20,7 @@ import ( log "github.com/sirupsen/logrus" + nbdns "github.com/netbirdio/netbird/dns" rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" resourceTypes "github.com/netbirdio/netbird/management/server/networks/resources/types" routerTypes "github.com/netbirdio/netbird/management/server/networks/routers/types" @@ -13,6 +29,97 @@ import ( "github.com/netbirdio/netbird/route" ) +// Snapshot is a consistent, in-memory view of the account collections needed to +// expand a Change into affected peers. It is loaded from the store INSIDE the +// caller's write transaction (so the data is consistent and read under the tx), +// and then Expand runs over it as pure in-memory computation AFTER the tx commits +// — keeping the expensive fan-out walk off the held write lock. +// +// Only the collections a given Change can actually touch are loaded; the rest are +// left nil (see Load). +type Snapshot struct { + policies []*types.Policy + routes []*route.Route + nsGroups []*nbdns.NameServerGroup + dnsSettings *types.DNSSettings + routers []*routerTypes.NetworkRouter + resources []*resourceTypes.NetworkResource + services []*rpservice.Service + proxyByCluster map[string][]string + groups map[string]*types.Group // all groups (for group.Resources lookups) + groupPeers map[string]map[string]struct{} // groupID -> member peer IDs +} + +// Load reads the collections a Change requires from the store, inside the caller's +// transaction. It mirrors Expand's walker preconditions so it loads only what the +// change can touch (e.g. nameserver/DNS only for group changes; services only when +// the account has embedded proxy peers). +func Load(ctx context.Context, s store.Store, accountID string, c Change) (*Snapshot, error) { + snap := &Snapshot{} + if c.isEmpty() { + return snap, nil + } + + hasGroupOrPeerChange := len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 + needsPolicies := hasGroupOrPeerChange || len(c.PostureCheckIDs) > 0 || len(c.Policies) > 0 || len(c.ResourceIDs) > 0 || len(c.NetworkIDs) > 0 + needsRoutersResources := needsPolicies // the resource<->router bridge can fire whenever policies/resources/networks are in play + + var err error + if needsPolicies { + if snap.policies, err = s.GetAccountPolicies(ctx, store.LockingStrengthNone, accountID); err != nil { + return nil, err + } + } + if hasGroupOrPeerChange { + if snap.routes, err = s.GetAccountRoutes(ctx, store.LockingStrengthNone, accountID); err != nil { + return nil, err + } + } + if len(c.ChangedGroupIDs) > 0 { + if snap.nsGroups, err = s.GetAccountNameServerGroups(ctx, store.LockingStrengthNone, accountID); err != nil { + return nil, err + } + if snap.dnsSettings, err = s.GetAccountDNSSettings(ctx, store.LockingStrengthNone, accountID); err != nil { + return nil, err + } + } + if needsRoutersResources { + if snap.routers, err = s.GetNetworkRoutersByAccountID(ctx, store.LockingStrengthNone, accountID); err != nil { + return nil, err + } + if snap.resources, err = s.GetNetworkResourcesByAccountID(ctx, store.LockingStrengthNone, accountID); err != nil { + return nil, err + } + } + if hasGroupOrPeerChange { + if snap.proxyByCluster, err = s.GetEmbeddedProxyPeerIDsByCluster(ctx, accountID); err != nil { + return nil, err + } + if len(snap.proxyByCluster) > 0 { + if snap.services, err = s.GetAccountServices(ctx, store.LockingStrengthNone, accountID); err != nil { + return nil, err + } + } + } + + // Groups (for group.Resources) and the group->peers index are always needed: + // the bridge resolves group.Resources, and the final expansion maps groups to + // member peers. + groups, err := s.GetAccountGroups(ctx, store.LockingStrengthNone, accountID) + if err != nil { + return nil, err + } + snap.groups = make(map[string]*types.Group, len(groups)) + for _, g := range groups { + snap.groups[g.ID] = g + } + if snap.groupPeers, err = s.GetAccountGroupPeers(ctx, store.LockingStrengthNone, accountID); err != nil { + return nil, err + } + + return snap, nil +} + // Change describes what changed in an account. The resolver never consults the // Enabled flag of any object: toggling Enabled is itself an observable change. type Change struct { @@ -23,6 +130,16 @@ type Change struct { PostureCheckIDs []string ResourceIDs []string NetworkIDs []string + + // RemovedPeersByGroup carries peers that left a group during this change, + // keyed by the group they left. A membership change does not alter which + // entities reference the group, so the dependency walk runs once against the + // post-change snapshot; these removed peers are no longer in the group's + // member index but still lose the group's reachability. They are folded into + // the affected set ONLY when their group is referenced (linked) — an unlinked + // group has no network-map impact, matching the included-when-linked semantics + // of current members. + RemovedPeersByGroup map[string][]string } func (c Change) isEmpty() bool { @@ -32,24 +149,41 @@ func (c Change) isEmpty() bool { len(c.Routes) == 0 && len(c.PostureCheckIDs) == 0 && len(c.ResourceIDs) == 0 && - len(c.NetworkIDs) == 0 + len(c.NetworkIDs) == 0 && + len(c.RemovedPeersByGroup) == 0 } -// Resolve returns the deduplicated peer IDs whose network map may have changed by -// the given Change. Safe to call inside or after a transaction. +// Expand computes the deduplicated peer IDs whose network map may have changed by +// the given Change, using only the preloaded Snapshot — no store access. Run it +// AFTER the transaction that produced the Snapshot has committed. // // At trace level it logs the full reasoning — which inputs drove which graph // walks to which groups/peers, including the resource<->router bridge hops — so a // miscalculation can be diagnosed from the logs alone. +func (snap *Snapshot) Expand(ctx context.Context, accountID string, c Change) []string { + if c.isEmpty() { + 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 postureChecks=%v resources=%v networks=%v", + accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, len(c.Policies), len(c.Routes), c.PostureCheckIDs, c.ResourceIDs, c.NetworkIDs) + r.walk() + return r.expand() +} + +// Resolve loads a Snapshot and expands it in one call. Convenience for callers +// that are not inside a transaction (and tests). Transaction-bound callers should +// use Load (inside the tx) + Snapshot.Expand (after commit) so the walk does not +// hold the write lock. func Resolve(ctx context.Context, s store.Store, accountID string, c Change) ([]string, error) { if c.isEmpty() { return nil, nil } - r := newResolver(ctx, s, accountID, c) - log.WithContext(ctx).Tracef("affectedpeers resolve start: account=%s changedGroups=%v changedPeers=%v policies=%d routes=%d postureChecks=%v resources=%v networks=%v", - accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, len(c.Policies), len(c.Routes), c.PostureCheckIDs, c.ResourceIDs, c.NetworkIDs) - r.walk() - return r.expand() + 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 IDs and direct peer IDs without expanding @@ -58,15 +192,20 @@ func Collect(ctx context.Context, s store.Store, accountID string, c Change) (gr if c.isEmpty() { return nil, nil } - r := newResolver(ctx, s, accountID, c) + snap, err := Load(ctx, s, accountID, c) + if err != nil { + log.WithContext(ctx).Errorf("failed to load snapshot for affected peers collect: %v", err) + return nil, nil + } + r := newResolver(ctx, snap, accountID, c) r.walk() return setToSlice(r.groupSet), setToSlice(r.peerSet) } -func newResolver(ctx context.Context, s store.Store, accountID string, c Change) *resolver { +func newResolver(ctx context.Context, snap *Snapshot, accountID string, c Change) *resolver { r := &resolver{ ctx: ctx, - store: s, + snap: snap, accountID: accountID, change: c, changedGroupSet: toSet(c.ChangedGroupIDs), @@ -99,7 +238,7 @@ func (r *resolver) walk() { type resolver struct { ctx context.Context - store store.Store + snap *Snapshot accountID string change Change @@ -112,72 +251,36 @@ type resolver struct { matchedPolicies []*types.Policy resourceIDs map[string]struct{} networkIDs map[string]struct{} - - // Memoized per-account collections: each is loaded from the store at most - // once per Resolve and only when a walker actually needs it. - cachedPolicies []*types.Policy - policiesLoaded bool - cachedResources []*resourceTypes.NetworkResource - resourcesLoaded bool - cachedRouters []*routerTypes.NetworkRouter - routersLoaded bool } -func (r *resolver) policies() []*types.Policy { - if r.policiesLoaded { - return r.cachedPolicies - } - r.policiesLoaded = true - policies, err := r.store.GetAccountPolicies(r.ctx, store.LockingStrengthNone, r.accountID) - if err != nil { - log.WithContext(r.ctx).Errorf("failed to get policies for affected peers resolution: %v", err) - return nil - } - r.cachedPolicies = policies - return r.cachedPolicies -} +func (r *resolver) policies() []*types.Policy { return r.snap.policies } -func (r *resolver) networkResources() []*resourceTypes.NetworkResource { - if r.resourcesLoaded { - return r.cachedResources - } - r.resourcesLoaded = true - resources, err := r.store.GetNetworkResourcesByAccountID(r.ctx, store.LockingStrengthNone, r.accountID) - if err != nil { - log.WithContext(r.ctx).Errorf("failed to get network resources for affected peers resolution: %v", err) - return nil - } - r.cachedResources = resources - return r.cachedResources -} +func (r *resolver) networkResources() []*resourceTypes.NetworkResource { return r.snap.resources } -func (r *resolver) networkRouters() []*routerTypes.NetworkRouter { - if r.routersLoaded { - return r.cachedRouters - } - r.routersLoaded = true - routers, err := r.store.GetNetworkRoutersByAccountID(r.ctx, store.LockingStrengthNone, r.accountID) - if err != nil { - log.WithContext(r.ctx).Errorf("failed to get network routers for affected peers resolution: %v", err) - return nil - } - r.cachedRouters = routers - return r.cachedRouters -} +func (r *resolver) networkRouters() []*routerTypes.NetworkRouter { return r.snap.routers } -func (r *resolver) expand() ([]string, error) { - groupIDs := setToSlice(r.groupSet) - var peerIDs []string - if len(groupIDs) > 0 { - ids, err := r.store.GetPeerIDsByGroups(r.ctx, r.accountID, groupIDs) - if err != nil { - return nil, err +// peerIDsForGroups maps a group set to its member peer IDs using the preloaded +// group->peers index (no store access). +func (r *resolver) peerIDsForGroups(groupSet map[string]struct{}) []string { + seen := make(map[string]struct{}) + var ids []string + for gID := range groupSet { + for pID := range r.snap.groupPeers[gID] { + if _, ok := seen[pID]; ok { + continue + } + seen[pID] = struct{}{} + ids = append(ids, pID) } - peerIDs = ids } + return ids +} - log.WithContext(r.ctx).Tracef("affectedpeers resolve expand: account=%s affectedGroups=%v -> %d group-member peers; direct peers=%v", - r.accountID, groupIDs, len(peerIDs), setToSlice(r.peerSet)) +func (r *resolver) expand() []string { + peerIDs := r.peerIDsForGroups(r.groupSet) + + log.WithContext(r.ctx).Tracef("affectedpeers expand: account=%s affectedGroups=%v -> %d group-member peers; direct peers=%v", + r.accountID, setToSlice(r.groupSet), len(peerIDs), setToSlice(r.peerSet)) seen := make(map[string]struct{}, len(peerIDs)) for _, id := range peerIDs { @@ -190,8 +293,24 @@ func (r *resolver) expand() ([]string, error) { } } - log.WithContext(r.ctx).Tracef("affectedpeers resolve done: account=%s -> %d affected peers: %v", r.accountID, len(peerIDs), peerIDs) - return peerIDs, nil + // Fold in peers removed from a group, but only when that group was referenced + // (folded into groupSet) — i.e. the group is linked. An unlinked group has no + // map impact, so its removed members are not affected. + for groupID, removed := range r.change.RemovedPeersByGroup { + if _, linked := r.groupSet[groupID]; !linked { + continue + } + for _, id := range removed { + if _, ok := seen[id]; !ok { + peerIDs = append(peerIDs, id) + seen[id] = struct{}{} + log.WithContext(r.ctx).Tracef("affectedpeers expand: removed peer %s from linked group %s -> affected", id, groupID) + } + } + } + + log.WithContext(r.ctx).Tracef("affectedpeers expand done: account=%s -> %d affected peers: %v", r.accountID, len(peerIDs), peerIDs) + return peerIDs } func (r *resolver) collectFromExplicitPolicies() { @@ -253,12 +372,7 @@ func (r *resolver) collectFromPolicies() { } func (r *resolver) collectFromRoutes() { - routes, err := r.store.GetAccountRoutes(r.ctx, store.LockingStrengthNone, r.accountID) - if err != nil { - log.WithContext(r.ctx).Errorf("failed to get routes for affected peers resolution: %v", err) - return - } - for _, rt := range routes { + for _, rt := range r.snap.routes { matchedByGroup := anyInSet(rt.Groups, r.changedGroupSet) || anyInSet(rt.PeerGroups, r.changedGroupSet) || anyInSet(rt.AccessControlGroups, r.changedGroupSet) matchedByPeer := rt.Peer != "" && len(r.changedPeerSet) > 0 && isInSet(rt.Peer, r.changedPeerSet) if !matchedByGroup && !matchedByPeer { @@ -277,12 +391,7 @@ func (r *resolver) collectFromNameServers() { if len(r.changedGroupSet) == 0 { return } - nsGroups, err := r.store.GetAccountNameServerGroups(r.ctx, store.LockingStrengthNone, r.accountID) - if err != nil { - log.WithContext(r.ctx).Errorf("failed to get nameserver groups for affected peers resolution: %v", err) - return - } - for _, ns := range nsGroups { + for _, ns := range r.snap.nsGroups { if anyInSet(ns.Groups, r.changedGroupSet) { log.WithContext(r.ctx).Tracef("collectFromNameServers: nameserver group %s references a changed group -> folding its groups %v", ns.ID, ns.Groups) addAll(r.groupSet, ns.Groups) @@ -291,15 +400,10 @@ func (r *resolver) collectFromNameServers() { } func (r *resolver) collectFromDNSSettings() { - if len(r.changedGroupSet) == 0 { + if len(r.changedGroupSet) == 0 || r.snap.dnsSettings == nil { return } - dnsSettings, err := r.store.GetAccountDNSSettings(r.ctx, store.LockingStrengthNone, r.accountID) - if err != nil { - log.WithContext(r.ctx).Errorf("failed to get DNS settings for affected peers resolution: %v", err) - return - } - for _, gID := range dnsSettings.DisabledManagementGroups { + for _, gID := range r.snap.dnsSettings.DisabledManagementGroups { if _, ok := r.changedGroupSet[gID]; ok { log.WithContext(r.ctx).Tracef("collectFromDNSSettings: changed group %s is in DisabledManagementGroups -> folding it", gID) r.groupSet[gID] = struct{}{} @@ -325,10 +429,10 @@ func (r *resolver) collectFromNetworkRouters() { } func (r *resolver) collectFromProxyServices() { - services, proxyByCluster, ok := r.loadProxyServiceContext() - if !ok { + if len(r.snap.proxyByCluster) == 0 || len(r.snap.services) == 0 { return } + services, proxyByCluster := r.snap.services, r.snap.proxyByCluster expanded := r.expandChangedPeersWithGroups() @@ -359,38 +463,11 @@ func (r *resolver) collectFromProxyServices() { } } -func (r *resolver) loadProxyServiceContext() ([]*rpservice.Service, map[string][]string, bool) { - // Embedded proxy peers are the prerequisite for any synthesized proxy policy. - // Probe that first (a narrow, indexed lookup) and skip the services table load - // entirely when the account has no embedded proxy peers. - proxyByCluster, err := r.store.GetEmbeddedProxyPeerIDsByCluster(r.ctx, r.accountID) - if err != nil { - log.WithContext(r.ctx).Errorf("failed to get embedded proxy peers for affected peers resolution: %v", err) - return nil, nil, false - } - if len(proxyByCluster) == 0 { - return nil, nil, false - } - services, err := r.store.GetAccountServices(r.ctx, store.LockingStrengthNone, r.accountID) - if err != nil { - log.WithContext(r.ctx).Errorf("failed to get services for affected peers resolution: %v", err) - return nil, nil, false - } - if len(services) == 0 { - return nil, nil, false - } - return services, proxyByCluster, true -} - func (r *resolver) expandChangedPeersWithGroups() map[string]struct{} { if len(r.changedGroupSet) == 0 { return r.changedPeerSet } - ids, err := r.store.GetPeerIDsByGroups(r.ctx, r.accountID, setToSlice(r.changedGroupSet)) - if err != nil { - log.WithContext(r.ctx).Errorf("failed to expand changed groups to peers for service resolution: %v", err) - return r.changedPeerSet - } + ids := r.peerIDsForGroups(r.changedGroupSet) if len(ids) == 0 { return r.changedPeerSet } @@ -499,12 +576,11 @@ func (r *resolver) policyTargetsResources(policy *types.Policy, resourceIDs map[ if len(destGroupSet) == 0 { return false } - groups, err := r.store.GetGroupsByIDs(r.ctx, store.LockingStrengthNone, r.accountID, setToSlice(destGroupSet)) - if err != nil { - log.WithContext(r.ctx).Errorf("failed to get destination groups for router policy bridge: %v", err) - return false - } - for _, group := range groups { + for gID := range destGroupSet { + group := r.snap.groups[gID] + if group == nil { + continue + } for _, res := range group.Resources { if isInSet(res.ID, resourceIDs) { return true @@ -541,15 +617,11 @@ func collectPolicyDestinations(resourceIDs map[string]struct{}, policies ...*typ // addGroupResourceIDs folds the resource IDs of the given groups into resourceIDs. func (r *resolver) addGroupResourceIDs(groupIDs map[string]struct{}, resourceIDs map[string]struct{}) { - if len(groupIDs) == 0 { - return - } - groups, err := r.store.GetGroupsByIDs(r.ctx, store.LockingStrengthNone, r.accountID, setToSlice(groupIDs)) - if err != nil { - log.WithContext(r.ctx).Errorf("failed to get destination groups for resource router bridge: %v", err) - return - } - for _, group := range groups { + for gID := range groupIDs { + group := r.snap.groups[gID] + if group == nil { + continue + } for _, res := range group.Resources { if res.ID != "" { resourceIDs[res.ID] = struct{}{} diff --git a/management/server/group.go b/management/server/group.go index a8c7593ff..510931462 100644 --- a/management/server/group.go +++ b/management/server/group.go @@ -80,7 +80,8 @@ func (am *DefaultAccountManager) CreateGroup(ctx context.Context, accountID, use } var eventsToStore []func() - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{ChangedGroupIDs: []string{newGroup.ID}} err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { if err = validateNewGroup(ctx, transaction, accountID, newGroup); err != nil { @@ -102,7 +103,10 @@ func (am *DefaultAccountManager) CreateGroup(ctx context.Context, accountID, use } } - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{ChangedGroupIDs: []string{newGroup.ID}}) + snap, err = affectedpeers.Load(ctx, transaction, accountID, change) + if err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) }) @@ -114,12 +118,7 @@ func (am *DefaultAccountManager) CreateGroup(ctx context.Context, accountID, use storeEvent() } - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("CreateGroup %s: updating %d affected peers: %v", newGroup.ID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("CreateGroup %s: no affected peers", newGroup.ID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -135,7 +134,8 @@ func (am *DefaultAccountManager) UpdateGroup(ctx context.Context, accountID, use } var eventsToStore []func() - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{ChangedGroupIDs: []string{newGroup.ID}} err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { if err = validateNewGroup(ctx, transaction, accountID, newGroup); err != nil { @@ -152,8 +152,9 @@ func (am *DefaultAccountManager) UpdateGroup(ctx context.Context, accountID, use return status.Errorf(status.NotFound, "group with ID %s not found", newGroup.ID) } + peersToAdd := util.Difference(newGroup.Peers, oldGroup.Peers) peersToRemove := util.Difference(oldGroup.Peers, newGroup.Peers) - if err = syncGroupMembership(ctx, transaction, accountID, newGroup.ID, util.Difference(newGroup.Peers, oldGroup.Peers), peersToRemove); err != nil { + if err = syncGroupMembership(ctx, transaction, accountID, newGroup.ID, peersToAdd, peersToRemove); err != nil { return err } @@ -165,7 +166,16 @@ func (am *DefaultAccountManager) UpdateGroup(ctx context.Context, accountID, use return err } - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{ChangedGroupIDs: []string{newGroup.ID}, ChangedPeerIDs: peersToRemove}) + // A membership change does not alter which entities reference the group, so + // the dependency walk runs once against the post-change snapshot. The new + // members are already in the snapshot's index; the removed members are + // carried separately and folded in only when the group is linked. + if len(peersToRemove) > 0 { + change.RemovedPeersByGroup = map[string][]string{newGroup.ID: peersToRemove} + } + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) }) @@ -177,12 +187,7 @@ func (am *DefaultAccountManager) UpdateGroup(ctx context.Context, accountID, use storeEvent() } - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("UpdateGroup %s: updating %d affected peers: %v", newGroup.ID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("UpdateGroup %s: no affected peers", newGroup.ID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -441,7 +446,8 @@ func (am *DefaultAccountManager) DeleteGroups(ctx context.Context, accountID, us var allErrors error var groupIDsToDelete []string var deletedGroups []*types.Group - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change extraSettings, err := am.settingsManager.GetExtraSettings(ctx, accountID) if err != nil { @@ -458,7 +464,13 @@ func (am *DefaultAccountManager) DeleteGroups(ctx context.Context, accountID, us return allErrors } - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{ChangedGroupIDs: groupIDsToDelete}) + // Delete: compute affected peers from the PRE-delete state. The groups, + // their members and the entities referencing them still exist, so a plain + // Load+Expand captures everyone — no removed-peer folding needed. + change = affectedpeers.Change{ChangedGroupIDs: groupIDsToDelete} + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } if err = transaction.DeleteGroups(ctx, accountID, groupIDsToDelete); err != nil { return err @@ -478,12 +490,7 @@ func (am *DefaultAccountManager) DeleteGroups(ctx context.Context, accountID, us am.StoreEvent(ctx, userID, group.ID, accountID, activity.GroupDeleted, group.EventMeta()) } - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("DeleteGroups %v: updating %d affected peers: %v", groupIDsToDelete, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("DeleteGroups %v: no affected peers", groupIDsToDelete) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return allErrors } @@ -510,19 +517,22 @@ func collectDeletableGroups(ctx context.Context, transaction store.Store, accoun // GroupAddPeer appends peer to the group func (am *DefaultAccountManager) GroupAddPeer(ctx context.Context, accountID, groupID, peerID string) error { - var affectedPeerIDs []string - var err error + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{ChangedGroupIDs: []string{groupID}} - err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { - if err = transaction.AddPeerToGroup(ctx, accountID, peerID, groupID); err != nil { + err := am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { + if err := transaction.AddPeerToGroup(ctx, accountID, peerID, groupID); err != nil { return err } - if err = am.reconcileIPv6ForGroupChanges(ctx, transaction, accountID, []string{groupID}); err != nil { + if err := am.reconcileIPv6ForGroupChanges(ctx, transaction, accountID, []string{groupID}); err != nil { return err } - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{ChangedGroupIDs: []string{groupID}}) + var err error + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) }) @@ -530,12 +540,7 @@ func (am *DefaultAccountManager) GroupAddPeer(ctx context.Context, accountID, gr return err } - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("GroupAddPeer group=%s peer=%s: updating %d affected peers: %v", groupID, peerID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("GroupAddPeer group=%s peer=%s: no affected peers", groupID, peerID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -543,8 +548,9 @@ func (am *DefaultAccountManager) GroupAddPeer(ctx context.Context, accountID, gr // GroupAddResource appends resource to the group func (am *DefaultAccountManager) GroupAddResource(ctx context.Context, accountID, groupID string, resource types.Resource) error { var group *types.Group - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot var err error + change := affectedpeers.Change{ChangedGroupIDs: []string{groupID}} err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { group, err = transaction.GetGroupByID(context.Background(), store.LockingStrengthUpdate, accountID, groupID) @@ -560,7 +566,9 @@ func (am *DefaultAccountManager) GroupAddResource(ctx context.Context, accountID return err } - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{ChangedGroupIDs: []string{groupID}}) + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) }) @@ -568,30 +576,32 @@ func (am *DefaultAccountManager) GroupAddResource(ctx context.Context, accountID return err } - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("GroupAddResource group=%s resource=%s: updating %d affected peers: %v", groupID, resource.ID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("GroupAddResource group=%s resource=%s: no affected peers", groupID, resource.ID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return nil } // GroupDeletePeer removes peer from the group func (am *DefaultAccountManager) GroupDeletePeer(ctx context.Context, accountID, groupID, peerID string) error { - var affectedPeerIDs []string - var err error + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{ + ChangedGroupIDs: []string{groupID}, + RemovedPeersByGroup: map[string][]string{groupID: {peerID}}, + } - err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { - // Resolve before removing, so the peer being removed is still included - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{ChangedGroupIDs: []string{groupID}}) - - if err = transaction.RemovePeerFromGroup(ctx, peerID, groupID); err != nil { + err := am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { + if err := transaction.RemovePeerFromGroup(ctx, peerID, groupID); err != nil { return err } - if err = am.reconcileIPv6ForGroupChanges(ctx, transaction, accountID, []string{groupID}); err != nil { + if err := am.reconcileIPv6ForGroupChanges(ctx, transaction, accountID, []string{groupID}); err != nil { + return err + } + + // The removed peer is carried in change.RemovedPeersByGroup and folded in + // only when the group is linked, so loading post-removal is correct. + var err error + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { return err } @@ -601,12 +611,7 @@ func (am *DefaultAccountManager) GroupDeletePeer(ctx context.Context, accountID, return err } - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("GroupDeletePeer group=%s peer=%s: updating %d affected peers: %v", groupID, peerID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("GroupDeletePeer group=%s peer=%s: no affected peers", groupID, peerID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -614,8 +619,9 @@ func (am *DefaultAccountManager) GroupDeletePeer(ctx context.Context, accountID, // GroupDeleteResource removes resource from the group func (am *DefaultAccountManager) GroupDeleteResource(ctx context.Context, accountID, groupID string, resource types.Resource) error { var group *types.Group - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot var err error + change := affectedpeers.Change{ChangedGroupIDs: []string{groupID}} err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { group, err = transaction.GetGroupByID(context.Background(), store.LockingStrengthUpdate, accountID, groupID) @@ -627,11 +633,15 @@ func (am *DefaultAccountManager) GroupDeleteResource(ctx context.Context, accoun return nil } - if err = transaction.UpdateGroup(ctx, group); err != nil { + // Load before persisting the removal, so the snapshot still maps the group + // to the resource and the bridge can reach its routing peers. + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { return err } - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{ChangedGroupIDs: []string{groupID}}) + if err = transaction.UpdateGroup(ctx, group); err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) }) @@ -639,12 +649,7 @@ func (am *DefaultAccountManager) GroupDeleteResource(ctx context.Context, accoun return err } - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("GroupDeleteResource group=%s resource=%s: updating %d affected peers: %v", groupID, resource.ID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("GroupDeleteResource group=%s resource=%s: no affected peers", groupID, resource.ID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return nil } diff --git a/management/server/networks/manager.go b/management/server/networks/manager.go index d96b5d447..4137c3122 100644 --- a/management/server/networks/manager.go +++ b/management/server/networks/manager.go @@ -128,9 +128,14 @@ func (m *managerImpl) DeleteNetwork(ctx context.Context, accountID, userID, netw } var eventsToStore []func() - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{NetworkIDs: []string{networkID}} err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { - affectedPeerIDs = m.accountManager.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{NetworkIDs: []string{networkID}}) + // Load before the cascade deletes: pre-state still references the network. + var lerr error + if snap, lerr = affectedpeers.Load(ctx, transaction, accountID, change); lerr != nil { + return lerr + } resources, err := transaction.GetNetworkResourcesByNetID(ctx, store.LockingStrengthUpdate, accountID, networkID) if err != nil { @@ -182,6 +187,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) diff --git a/management/server/networks/resources/manager.go b/management/server/networks/resources/manager.go index 03b37db6f..ae60e34cf 100644 --- a/management/server/networks/resources/manager.go +++ b/management/server/networks/resources/manager.go @@ -115,10 +115,11 @@ func (m *managerImpl) CreateResource(ctx context.Context, userID string, resourc } var eventsToStore []func() - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{ResourceIDs: []string{resource.ID}} err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { var txErr error - eventsToStore, affectedPeerIDs, txErr = m.createResourceInTransaction(ctx, transaction, userID, resource) + eventsToStore, snap, txErr = m.createResourceInTransaction(ctx, transaction, userID, resource, change) return txErr }) if err != nil { @@ -129,6 +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) @@ -139,7 +141,7 @@ func (m *managerImpl) CreateResource(ctx context.Context, userID string, resourc return resource, nil } -func (m *managerImpl) createResourceInTransaction(ctx context.Context, transaction store.Store, userID string, resource *types.NetworkResource) ([]func(), []string, error) { +func (m *managerImpl) createResourceInTransaction(ctx context.Context, transaction store.Store, userID string, resource *types.NetworkResource, change affectedpeers.Change) ([]func(), *affectedpeers.Snapshot, error) { _, err := transaction.GetNetworkResourceByName(ctx, store.LockingStrengthNone, resource.AccountID, resource.Name) if err == nil { return nil, nil, status.Errorf(status.InvalidArgument, "resource with name %s already exists", resource.Name) @@ -175,9 +177,12 @@ func (m *managerImpl) createResourceInTransaction(ctx context.Context, transacti return nil, nil, fmt.Errorf("failed to increment network serial: %w", err) } - affectedPeerIDs := m.accountManager.ResolveAffectedPeers(ctx, transaction, resource.AccountID, affectedpeers.Change{ResourceIDs: []string{resource.ID}}) + snap, err := affectedpeers.Load(ctx, transaction, resource.AccountID, change) + if err != nil { + return nil, nil, err + } - return eventsToStore, affectedPeerIDs, nil + return eventsToStore, snap, nil } func (m *managerImpl) GetResource(ctx context.Context, accountID, userID, networkID, resourceID string) (*types.NetworkResource, error) { @@ -220,7 +225,8 @@ func (m *managerImpl) UpdateResource(ctx context.Context, userID string, resourc resource.Prefix = prefix var eventsToStore []func() - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { network, err := transaction.GetNetworkByID(ctx, store.LockingStrengthUpdate, resource.AccountID, resource.NetworkID) if err != nil { @@ -272,10 +278,13 @@ func (m *managerImpl) UpdateResource(ctx context.Context, userID string, resourc // Pass both old and new resource group IDs so policies that targeted the // resource via a now-detached group still refresh their source peers. - affectedPeerIDs = m.accountManager.ResolveAffectedPeers(ctx, transaction, resource.AccountID, affectedpeers.Change{ + change = affectedpeers.Change{ ResourceIDs: []string{resource.ID}, ChangedGroupIDs: append(oldGroupIDs, resource.GroupIDs...), - }) + } + if snap, err = affectedpeers.Load(ctx, transaction, resource.AccountID, change); err != nil { + return err + } err = transaction.IncrementNetworkSerial(ctx, resource.AccountID) if err != nil { @@ -300,6 +309,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) @@ -366,9 +376,13 @@ func (m *managerImpl) DeleteResource(ctx context.Context, accountID, userID, net } var events []func() - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{ResourceIDs: []string{resourceID}} err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { - affectedPeerIDs = m.accountManager.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{ResourceIDs: []string{resourceID}}) + // Load before delete: pre-state snapshot still references the resource. + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } events, err = m.DeleteResourceInTransaction(ctx, transaction, accountID, userID, networkID, resourceID) if err != nil { @@ -390,6 +404,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) diff --git a/management/server/networks/routers/manager.go b/management/server/networks/routers/manager.go index 7384cab5f..2ed455b8a 100644 --- a/management/server/networks/routers/manager.go +++ b/management/server/networks/routers/manager.go @@ -91,7 +91,8 @@ func (m *managerImpl) CreateRouter(ctx context.Context, userID string, router *t } var network *networkTypes.Network - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{NetworkIDs: []string{router.NetworkID}} err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { network, err = transaction.GetNetworkByID(ctx, store.LockingStrengthNone, router.AccountID, router.NetworkID) if err != nil { @@ -114,7 +115,9 @@ func (m *managerImpl) CreateRouter(ctx context.Context, userID string, router *t return fmt.Errorf("failed to increment network serial: %w", err) } - affectedPeerIDs = m.accountManager.ResolveAffectedPeers(ctx, transaction, router.AccountID, affectedpeers.Change{NetworkIDs: []string{router.NetworkID}}) + if snap, err = affectedpeers.Load(ctx, transaction, router.AccountID, change); err != nil { + return err + } return nil }) @@ -124,6 +127,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) @@ -165,10 +169,12 @@ func (m *managerImpl) UpdateRouter(ctx context.Context, userID string, router *t } var network *networkTypes.Network - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change + var oldRouting []string err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { var txErr error - network, affectedPeerIDs, txErr = m.updateRouterInTransaction(ctx, transaction, router) + network, snap, change, oldRouting, txErr = m.updateRouterInTransaction(ctx, transaction, router) return txErr }) if err != nil { @@ -177,6 +183,9 @@ 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)) + // The previous routing peers lose their routing role and are no longer reachable + // from the post-update state, so union them as directly affected after Expand. + affectedPeerIDs := append(snap.Expand(ctx, router.AccountID, change), oldRouting...) 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) @@ -187,31 +196,34 @@ func (m *managerImpl) UpdateRouter(ctx context.Context, userID string, router *t return router, nil } -func (m *managerImpl) updateRouterInTransaction(ctx context.Context, transaction store.Store, router *types.NetworkRouter) (*networkTypes.Network, []string, error) { +func (m *managerImpl) updateRouterInTransaction(ctx context.Context, transaction store.Store, router *types.NetworkRouter) (*networkTypes.Network, *affectedpeers.Snapshot, affectedpeers.Change, []string, error) { network, err := transaction.GetNetworkByID(ctx, store.LockingStrengthNone, router.AccountID, router.NetworkID) if err != nil { - return nil, nil, fmt.Errorf("failed to get network: %w", err) + return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to get network: %w", err) } existing, err := transaction.GetNetworkRouterByID(ctx, store.LockingStrengthUpdate, router.AccountID, router.ID) if err != nil { - return nil, nil, fmt.Errorf("failed to get network router: %w", err) + return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to get network router: %w", err) } if existing.AccountID != router.AccountID { - return nil, nil, status.NewNetworkRouterNotFoundError(router.ID) + return nil, nil, affectedpeers.Change{}, nil, status.NewNetworkRouterNotFoundError(router.ID) } if existing.NetworkID != router.NetworkID { - return nil, nil, status.NewRouterNotPartOfNetworkError(router.ID, router.NetworkID) + return nil, nil, affectedpeers.Change{}, nil, status.NewRouterNotPartOfNetworkError(router.ID, router.NetworkID) } + // Capture the previous routing peers before persisting the update. + oldRouting := oldRoutingPeerIDs(ctx, transaction, router.AccountID, existing) + if err = transaction.UpdateNetworkRouter(ctx, router); err != nil { - return nil, nil, fmt.Errorf("failed to update network router: %w", err) + return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to update network router: %w", err) } if err = transaction.IncrementNetworkSerial(ctx, router.AccountID); err != nil { - return nil, nil, fmt.Errorf("failed to increment network serial: %w", err) + return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to increment network serial: %w", err) } networkIDs := []string{router.NetworkID} @@ -219,14 +231,13 @@ func (m *managerImpl) updateRouterInTransaction(ctx context.Context, transaction networkIDs = append(networkIDs, existing.NetworkID) } - affectedPeerIDs := m.accountManager.ResolveAffectedPeers(ctx, transaction, router.AccountID, affectedpeers.Change{NetworkIDs: networkIDs}) + change := affectedpeers.Change{NetworkIDs: networkIDs} + snap, err := affectedpeers.Load(ctx, transaction, router.AccountID, change) + if err != nil { + return nil, nil, affectedpeers.Change{}, nil, err + } - // The previous routing peer / peer-group members lose their routing role and - // are no longer reachable from the post-update network state, so add them - // explicitly. - affectedPeerIDs = append(affectedPeerIDs, oldRoutingPeerIDs(ctx, transaction, router.AccountID, existing)...) - - return network, affectedPeerIDs, nil + return network, snap, change, oldRouting, nil } // oldRoutingPeerIDs returns the peer IDs that served as the router's routing peers @@ -257,9 +268,13 @@ func (m *managerImpl) DeleteRouter(ctx context.Context, accountID, userID, netwo } var event func() - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{NetworkIDs: []string{networkID}} err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { - affectedPeerIDs = m.accountManager.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{NetworkIDs: []string{networkID}}) + // Load before delete: pre-state still references the router and its peers. + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } event, err = m.DeleteRouterInTransaction(ctx, transaction, accountID, userID, networkID, routerID) if err != nil { @@ -279,6 +294,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) diff --git a/management/server/peer.go b/management/server/peer.go index 948a19602..1ca97f4a1 100644 --- a/management/server/peer.go +++ b/management/server/peer.go @@ -512,7 +512,8 @@ func (am *DefaultAccountManager) DeletePeer(ctx context.Context, accountID, peer var peer *nbpeer.Peer var settings *types.Settings var eventsToStore []func() - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change serviceID, err := am.serviceManager.GetServiceIDByTargetID(ctx, accountID, peerID) if err != nil { @@ -537,7 +538,15 @@ func (am *DefaultAccountManager) DeletePeer(ctx context.Context, accountID, peer return err } - affectedPeerIDs = am.resolveAffectedPeersForPeerChanges(ctx, transaction, accountID, []string{peerID}) + // Load before delete: pre-state still has the peer's group memberships. + groupIDs, err := transaction.GetGroupIDsByPeerIDs(ctx, accountID, []string{peerID}) + if err != nil { + return fmt.Errorf("failed to get group IDs for peer: %w", err) + } + change = affectedpeers.Change{ChangedGroupIDs: groupIDs} + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } eventsToStore, err = deletePeers(ctx, am, transaction, accountID, userID, []*nbpeer.Peer{peer}, settings) if err != nil { @@ -562,6 +571,7 @@ func (am *DefaultAccountManager) DeletePeer(ctx context.Context, accountID, peer log.WithContext(ctx).Errorf("failed to delete peer %s from integrated validator: %v", peerID, err) } + affectedPeerIDs := snap.Expand(ctx, accountID, change) if err = am.networkMapController.OnPeersDeleted(ctx, accountID, []string{peerID}, affectedPeerIDs); err != nil { log.WithContext(ctx).Errorf("failed to delete peer %s from network map: %v", peerID, err) } @@ -1435,6 +1445,39 @@ func (am *DefaultAccountManager) UpdateAffectedPeers(ctx context.Context, accoun _ = am.networkMapController.UpdateAffectedPeers(ctx, accountID, peerIDs) } +// 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 := snap.Expand(ctx, accountID, change) + if len(directlyAffected) > 0 { + seen := make(map[string]struct{}, len(affectedPeerIDs)) + for _, id := range affectedPeerIDs { + seen[id] = struct{}{} + } + for _, id := range directlyAffected { + if _, ok := seen[id]; !ok { + affectedPeerIDs = append(affectedPeerIDs, id) + seen[id] = struct{}{} + } + } + } + + if len(affectedPeerIDs) > 0 { + log.WithContext(ctx).Debugf("expandAndUpdateAffected: account %s updating %d affected peers: %v", accountID, len(affectedPeerIDs), affectedPeerIDs) + am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) + } else { + log.WithContext(ctx).Tracef("expandAndUpdateAffected: account %s no affected peers", accountID) + } +} + // 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) diff --git a/management/server/policy.go b/management/server/policy.go index d9b8598ae..f72679e81 100644 --- a/management/server/policy.go +++ b/management/server/policy.go @@ -49,7 +49,8 @@ func (am *DefaultAccountManager) SavePolicy(ctx context.Context, accountID, user var existingPolicy *types.Policy var action = activity.PolicyAdded var unchanged bool - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { existingPolicy, err = validatePolicy(ctx, transaction, accountID, policy) @@ -75,7 +76,10 @@ func (am *DefaultAccountManager) SavePolicy(ctx context.Context, accountID, user } } - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{Policies: []*types.Policy{policy, existingPolicy}}) + change = affectedpeers.Change{Policies: []*types.Policy{policy, existingPolicy}} + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) }) @@ -89,12 +93,7 @@ func (am *DefaultAccountManager) SavePolicy(ctx context.Context, accountID, user am.StoreEvent(ctx, userID, policy.ID, accountID, action, policy.EventMeta()) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Tracef("SavePolicy %s: updating %d affected peers: %v", policy.ID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("SavePolicy %s: no affected peers", policy.ID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return policy, nil } @@ -110,7 +109,8 @@ func (am *DefaultAccountManager) DeletePolicy(ctx context.Context, accountID, po } var policy *types.Policy - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{} err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { policy, err = transaction.GetPolicyByID(ctx, store.LockingStrengthUpdate, accountID, policyID) @@ -118,7 +118,11 @@ func (am *DefaultAccountManager) DeletePolicy(ctx context.Context, accountID, po return err } - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{Policies: []*types.Policy{policy}}) + // Load before delete: pre-state still references the policy. + change = affectedpeers.Change{Policies: []*types.Policy{policy}} + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } if err = transaction.DeletePolicy(ctx, accountID, policyID); err != nil { return err @@ -132,12 +136,7 @@ func (am *DefaultAccountManager) DeletePolicy(ctx context.Context, accountID, po am.StoreEvent(ctx, userID, policyID, accountID, activity.PolicyRemoved, policy.EventMeta()) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("DeletePolicy %s: updating %d affected peers: %v", policyID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("DeletePolicy %s: no affected peers", policyID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return nil } diff --git a/management/server/posture_checks.go b/management/server/posture_checks.go index f460145b5..54009cdd7 100644 --- a/management/server/posture_checks.go +++ b/management/server/posture_checks.go @@ -5,7 +5,6 @@ import ( "slices" "github.com/rs/xid" - log "github.com/sirupsen/logrus" "github.com/netbirdio/netbird/management/server/activity" "github.com/netbirdio/netbird/management/server/affectedpeers" @@ -44,7 +43,8 @@ func (am *DefaultAccountManager) SavePostureChecks(ctx context.Context, accountI var isUpdate = postureChecks.ID != "" var action = activity.PostureCheckCreated - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + change := affectedpeers.Change{PostureCheckIDs: []string{postureChecks.ID}} err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { if err = validatePostureChecks(ctx, transaction, accountID, postureChecks); err != nil { @@ -53,8 +53,6 @@ func (am *DefaultAccountManager) SavePostureChecks(ctx context.Context, accountI if isUpdate { action = activity.PostureCheckUpdated - - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{PostureCheckIDs: []string{postureChecks.ID}}) } postureChecks.AccountID = accountID @@ -63,6 +61,11 @@ func (am *DefaultAccountManager) SavePostureChecks(ctx context.Context, accountI } if isUpdate { + // Editing a posture check does not change which policies reference it, + // so loading after the save is fine. + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) } @@ -74,12 +77,7 @@ func (am *DefaultAccountManager) SavePostureChecks(ctx context.Context, accountI am.StoreEvent(ctx, userID, postureChecks.ID, accountID, action, postureChecks.EventMeta()) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("SavePostureChecks %s: updating %d affected peers: %v", postureChecks.ID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("SavePostureChecks %s: no affected peers", postureChecks.ID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return postureChecks, nil } diff --git a/management/server/route.go b/management/server/route.go index 44f2528b7..31d2f8108 100644 --- a/management/server/route.go +++ b/management/server/route.go @@ -8,7 +8,6 @@ import ( "unicode/utf8" "github.com/rs/xid" - log "github.com/sirupsen/logrus" "github.com/netbirdio/netbird/management/server/activity" "github.com/netbirdio/netbird/management/server/affectedpeers" @@ -149,7 +148,8 @@ func (am *DefaultAccountManager) CreateRoute(ctx context.Context, accountID stri } var newRoute *route.Route - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { newRoute = &route.Route{ @@ -179,7 +179,10 @@ func (am *DefaultAccountManager) CreateRoute(ctx context.Context, accountID stri return err } - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{Routes: []*route.Route{newRoute}}) + change = affectedpeers.Change{Routes: []*route.Route{newRoute}} + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) }) @@ -189,12 +192,7 @@ func (am *DefaultAccountManager) CreateRoute(ctx context.Context, accountID stri am.StoreEvent(ctx, userID, string(newRoute.ID), accountID, activity.RouteCreated, newRoute.EventMeta()) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("CreateRoute %s: updating %d affected peers: %v", newRoute.ID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("CreateRoute %s: no affected peers", newRoute.ID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return newRoute, nil } @@ -210,7 +208,8 @@ func (am *DefaultAccountManager) SaveRoute(ctx context.Context, accountID, userI } var oldRoute *route.Route - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { if err = validateRoute(ctx, transaction, accountID, routeToSave); err != nil { @@ -228,7 +227,10 @@ func (am *DefaultAccountManager) SaveRoute(ctx context.Context, accountID, userI return err } - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{Routes: []*route.Route{routeToSave, oldRoute}}) + change = affectedpeers.Change{Routes: []*route.Route{routeToSave, oldRoute}} + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } return transaction.IncrementNetworkSerial(ctx, accountID) }) @@ -238,12 +240,7 @@ func (am *DefaultAccountManager) SaveRoute(ctx context.Context, accountID, userI am.StoreEvent(ctx, userID, string(routeToSave.ID), accountID, activity.RouteUpdated, routeToSave.EventMeta()) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("SaveRoute %s: updating %d affected peers: %v", routeToSave.ID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("SaveRoute %s: no affected peers", routeToSave.ID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return nil } @@ -259,7 +256,8 @@ func (am *DefaultAccountManager) DeleteRoute(ctx context.Context, accountID stri } var rt *route.Route - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { rt, err = transaction.GetRouteByID(ctx, store.LockingStrengthUpdate, accountID, string(routeID)) @@ -267,7 +265,11 @@ func (am *DefaultAccountManager) DeleteRoute(ctx context.Context, accountID stri return err } - affectedPeerIDs = am.ResolveAffectedPeers(ctx, transaction, accountID, affectedpeers.Change{Routes: []*route.Route{rt}}) + // Load before delete: pre-state captures everyone referencing the route. + change = affectedpeers.Change{Routes: []*route.Route{rt}} + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } if err = transaction.DeleteRoute(ctx, accountID, string(routeID)); err != nil { return err @@ -281,12 +283,7 @@ func (am *DefaultAccountManager) DeleteRoute(ctx context.Context, accountID stri am.StoreEvent(ctx, userID, string(rt.ID), accountID, activity.RouteRemoved, rt.EventMeta()) - if len(affectedPeerIDs) > 0 { - log.WithContext(ctx).Debugf("DeleteRoute %s: updating %d affected peers: %v", routeID, len(affectedPeerIDs), affectedPeerIDs) - am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) - } else { - log.WithContext(ctx).Tracef("DeleteRoute %s: no affected peers", routeID) - } + am.expandAndUpdateAffected(ctx, accountID, snap, change) return nil } diff --git a/management/server/user.go b/management/server/user.go index 60dde7aba..35536b43a 100644 --- a/management/server/user.go +++ b/management/server/user.go @@ -18,6 +18,7 @@ import ( "github.com/netbirdio/netbird/idp/dex" "github.com/netbirdio/netbird/management/server/account" "github.com/netbirdio/netbird/management/server/activity" + "github.com/netbirdio/netbird/management/server/affectedpeers" "github.com/netbirdio/netbird/management/server/idp" nbpeer "github.com/netbirdio/netbird/management/server/peer" "github.com/netbirdio/netbird/management/server/permissions/modules" @@ -1274,7 +1275,8 @@ func (am *DefaultAccountManager) deleteRegularUser(ctx context.Context, accountI var userPeers []*nbpeer.Peer var targetUser *types.User var settings *types.Settings - var affectedPeerIDs []string + var snap *affectedpeers.Snapshot + var change affectedpeers.Change var err error err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { @@ -1300,8 +1302,15 @@ func (am *DefaultAccountManager) deleteRegularUser(ctx context.Context, accountI for _, peer := range userPeers { peerIDs = append(peerIDs, peer.ID) } - // Resolve before delete so group memberships are still present. - affectedPeerIDs = am.resolveAffectedPeersForPeerChanges(ctx, transaction, accountID, peerIDs) + // Load before delete: pre-state still has the peers' group memberships. + groupIDs, err := transaction.GetGroupIDsByPeerIDs(ctx, accountID, peerIDs) + if err != nil { + return fmt.Errorf("failed to get group IDs for user peers: %w", err) + } + change = affectedpeers.Change{ChangedGroupIDs: groupIDs} + if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { + return err + } addPeerRemovedEvents, err = deletePeers(ctx, am, transaction, accountID, targetUserInfo.ID, userPeers, settings) if err != nil { @@ -1326,6 +1335,7 @@ func (am *DefaultAccountManager) deleteRegularUser(ctx context.Context, accountI log.WithContext(ctx).Errorf("failed to delete peer %s from integrated validator: %v", peer.ID, err) } } + affectedPeerIDs := snap.Expand(ctx, accountID, change) if err := am.networkMapController.OnPeersDeleted(ctx, accountID, peerIDs, affectedPeerIDs); err != nil { log.WithContext(ctx).Errorf("failed to delete peers %s from network map: %v", peerIDs, err) }