diff --git a/management/internals/controllers/network_map/controller/controller.go b/management/internals/controllers/network_map/controller/controller.go index 7d6ea4152..5157ca4fa 100644 --- a/management/internals/controllers/network_map/controller/controller.go +++ b/management/internals/controllers/network_map/controller/controller.go @@ -512,10 +512,14 @@ func (c *Controller) BufferUpdateAffectedPeers(ctx context.Context, accountID st b.stopTimer() + // The send and the debounced timer outlive the calling request, so detach from + // its context to avoid sending with a cancelled context once the handler returns. + bgCtx := context.WithoutCancel(ctx) + collected := b.drainPeerIDs() go func() { defer b.sendMu.Unlock() - _ = c.sendUpdateForAffectedPeers(ctx, accountID, collected) + _ = c.sendUpdateForAffectedPeers(bgCtx, accountID, collected) // Check if more peer IDs accumulated while we were sending. if !b.hasPending() { @@ -526,7 +530,7 @@ func (c *Controller) BufferUpdateAffectedPeers(ctx context.Context, accountID st b.setTimer(time.Duration(c.updateAccountPeersBufferInterval.Load()), func() { ids := b.drainPeerIDs() if len(ids) > 0 { - _ = c.sendUpdateForAffectedPeers(ctx, accountID, ids) + _ = c.sendUpdateForAffectedPeers(bgCtx, accountID, ids) } }) }() @@ -828,7 +832,7 @@ func (c *Controller) OnPeersDeleted(ctx context.Context, accountID string, peerI } if len(affectedPeerIDs) == 0 { - log.WithContext(ctx).Tracef("no affected peers for peer delete in account %s, skipping network map update", accountID) + log.WithContext(ctx).Tracef("no affected peers for peer delete in account %s, skipping", accountID) return nil } return c.BufferUpdateAffectedPeers(ctx, accountID, affectedPeerIDs, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationDelete}) diff --git a/management/server/affected_peers_coverage_test.go b/management/server/affected_peers_coverage_test.go index e62aba90e..6942a603d 100644 --- a/management/server/affected_peers_coverage_test.go +++ b/management/server/affected_peers_coverage_test.go @@ -8,6 +8,8 @@ import ( "github.com/stretchr/testify/require" "github.com/netbirdio/netbird/management/server/affectedpeers" + resourceTypes "github.com/netbirdio/netbird/management/server/networks/resources/types" + networkTypes "github.com/netbirdio/netbird/management/server/networks/types" "github.com/netbirdio/netbird/management/server/posture" "github.com/netbirdio/netbird/management/server/types" ) @@ -63,7 +65,9 @@ func TestAffectedPeers_DependencyCoverageMatrix(t *testing.T) { build: func(t *testing.T, s *routerScenario, ctx context.Context) (affectedpeers.Change, []string, []string) { _, err := s.manager.SavePolicy(ctx, s.accountID, userID, peerToResourcePolicyByGroup(s.sourceGroupID, s.resourceGroupID), true) require.NoError(t, err) - return affectedpeers.Change{ResourceIDs: []string{s.resourceID}}, + return affectedpeers.Change{Resources: []*resourceTypes.NetworkResource{ + {ID: s.resourceID, NetworkID: s.networkID, GroupIDs: []string{s.resourceGroupID}}, + }}, []string{s.sourcePeerID, s.routerPeerID}, []string{s.unrelatedPeerID} }, }, @@ -72,7 +76,7 @@ func TestAffectedPeers_DependencyCoverageMatrix(t *testing.T) { build: func(t *testing.T, s *routerScenario, ctx context.Context) (affectedpeers.Change, []string, []string) { _, err := s.manager.SavePolicy(ctx, s.accountID, userID, peerToResourcePolicyByGroup(s.sourceGroupID, s.resourceGroupID), true) require.NoError(t, err) - return affectedpeers.Change{NetworkIDs: []string{s.networkID}}, + return affectedpeers.Change{Networks: []*networkTypes.Network{{ID: s.networkID}}}, []string{s.sourcePeerID, s.routerPeerID}, []string{s.unrelatedPeerID} }, }, diff --git a/management/server/affected_peers_querycount_test.go b/management/server/affected_peers_querycount_test.go index 079d548c5..56e9c1bd4 100644 --- a/management/server/affected_peers_querycount_test.go +++ b/management/server/affected_peers_querycount_test.go @@ -13,6 +13,7 @@ import ( "github.com/netbirdio/netbird/management/server/affectedpeers" resourceTypes "github.com/netbirdio/netbird/management/server/networks/resources/types" routerTypes "github.com/netbirdio/netbird/management/server/networks/routers/types" + networkTypes "github.com/netbirdio/netbird/management/server/networks/types" "github.com/netbirdio/netbird/management/server/store" "github.com/netbirdio/netbird/management/server/types" "github.com/netbirdio/netbird/route" @@ -123,9 +124,9 @@ func TestAffectedPeers_QueryCount_NarrowChangeSkipsLoads(t *testing.T) { cs := newCountingStore(s.manager.Store) - // A bare network-id change drives only the router->source bridge: routers and + // 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{NetworkIDs: []string{s.networkID}}) + _, err := affectedpeers.Resolve(ctx, cs, s.accountID, affectedpeers.Change{Networks: []*networkTypes.Network{{ID: s.networkID}}}) require.NoError(t, err) assert.Equal(t, 0, cs.count("routes"), "routes must not be loaded for a network-only change") diff --git a/management/server/affectedpeers/resolver.go b/management/server/affectedpeers/resolver.go index e142e262a..b93856040 100644 --- a/management/server/affectedpeers/resolver.go +++ b/management/server/affectedpeers/resolver.go @@ -64,7 +64,7 @@ func Load(ctx context.Context, s store.Store, accountID string, c Change) (*Snap // Changed resources contribute their group IDs to the changed-group set during // the walk (see collectFromExplicitResources), so they drive the group walkers. hasGroupOrPeerChange := len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 || len(c.Resources) > 0 - hasNetworkObject := len(c.ResourceIDs) > 0 || len(c.NetworkIDs) > 0 || len(c.Routers) > 0 || len(c.Resources) > 0 || len(c.Networks) > 0 + hasNetworkObject := len(c.Routers) > 0 || len(c.Resources) > 0 || len(c.Networks) > 0 needsPolicies := hasGroupOrPeerChange || len(c.PostureCheckIDs) > 0 || len(c.Policies) > 0 || hasNetworkObject needsRoutersResources := needsPolicies // the resource<->router bridge can fire whenever policies/resources/networks are in play @@ -138,8 +138,6 @@ type Change struct { Resources []*resourceTypes.NetworkResource Networks []*networkTypes.Network 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 @@ -161,8 +159,6 @@ func (c Change) isEmpty() bool { len(c.Resources) == 0 && len(c.Networks) == 0 && len(c.PostureCheckIDs) == 0 && - len(c.ResourceIDs) == 0 && - len(c.NetworkIDs) == 0 && len(c.RemovedPeersByGroup) == 0 } @@ -178,8 +174,8 @@ 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 resourceIDs=%v networkIDs=%v", - accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, len(c.Policies), len(c.Routes), len(c.Routers), len(c.Resources), len(c.Networks), c.PostureCheckIDs, c.ResourceIDs, c.NetworkIDs) + 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) r.walk() return r.expand() } @@ -225,8 +221,7 @@ func newResolver(ctx context.Context, snap *Snapshot, accountID string, c Change changedPeerSet: toSet(c.ChangedPeerIDs), groupSet: make(map[string]struct{}), peerSet: make(map[string]struct{}), - resourceIDs: toSet(c.ResourceIDs), - networkIDs: toSet(c.NetworkIDs), + networkIDs: make(map[string]struct{}), } // A changed peer affects every entity referencing a group it belongs to, so // seed the changed-group set with the peer's memberships from the snapshot's @@ -288,7 +283,6 @@ type resolver struct { peerSet map[string]struct{} matchedPolicies []*types.Policy - resourceIDs map[string]struct{} networkIDs map[string]struct{} } @@ -588,9 +582,6 @@ func (r *resolver) collectResourceRouterBridge() { func (r *resolver) bridgeSourceToRouters() { resourceIDs := r.policyDestinationResourceIDs(r.matchedPolicies...) - for id := range r.resourceIDs { - resourceIDs[id] = struct{}{} - } if len(resourceIDs) == 0 { return } diff --git a/management/server/affectedpeers/resolver_test.go b/management/server/affectedpeers/resolver_test.go index 2f9c2d523..dcd304a56 100644 --- a/management/server/affectedpeers/resolver_test.go +++ b/management/server/affectedpeers/resolver_test.go @@ -5,6 +5,8 @@ import ( "github.com/stretchr/testify/assert" + resourceTypes "github.com/netbirdio/netbird/management/server/networks/resources/types" + networkTypes "github.com/netbirdio/netbird/management/server/networks/types" "github.com/netbirdio/netbird/management/server/types" ) @@ -73,8 +75,8 @@ func TestChangeIsEmpty(t *testing.T) { assert.False(t, Change{ChangedGroupIDs: []string{"g"}}.isEmpty()) assert.False(t, Change{ChangedPeerIDs: []string{"p"}}.isEmpty()) assert.False(t, Change{Policies: []*types.Policy{{}}}.isEmpty()) - assert.False(t, Change{ResourceIDs: []string{"r"}}.isEmpty()) - assert.False(t, Change{NetworkIDs: []string{"n"}}.isEmpty()) + assert.False(t, Change{Resources: []*resourceTypes.NetworkResource{{ID: "r"}}}.isEmpty()) + assert.False(t, Change{Networks: []*networkTypes.Network{{ID: "n"}}}.isEmpty()) assert.False(t, Change{PostureCheckIDs: []string{"pc"}}.isEmpty()) } diff --git a/management/server/networks/manager.go b/management/server/networks/manager.go index d0b7a32c6..7a339d1a5 100644 --- a/management/server/networks/manager.go +++ b/management/server/networks/manager.go @@ -131,17 +131,38 @@ func (m *managerImpl) DeleteNetwork(ctx context.Context, accountID, userID, netw var snap *affectedpeers.Snapshot change := affectedpeers.Change{Networks: []*types.Network{network}} err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { + resources, err := transaction.GetNetworkResourcesByNetID(ctx, store.LockingStrengthUpdate, accountID, networkID) + if err != nil { + return fmt.Errorf("failed to get resources in network: %w", err) + } + + netRouters, err := transaction.GetNetworkRoutersByNetID(ctx, store.LockingStrengthUpdate, accountID, networkID) + if err != nil { + return fmt.Errorf("failed to get routers in network: %w", err) + } + + // Carry the cascade-deleted resources and routers in the Change so the + // post-commit Expand walks their groups too: a resource whose group is a + // policy source affects that source's peers, which a network-only Change + // would miss. Hydrate each resource's GroupIDs (gorm:"-") before Load. + for _, resource := range resources { + groups, err := transaction.GetResourceGroups(ctx, store.LockingStrengthNone, accountID, resource.ID) + if err != nil { + return fmt.Errorf("failed to get resource groups: %w", err) + } + for _, g := range groups { + resource.GroupIDs = append(resource.GroupIDs, g.ID) + } + } + change.Resources = resources + change.Routers = netRouters + // 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 { - return fmt.Errorf("failed to get resources in network: %w", err) - } - for _, resource := range resources { event, err := m.resourcesManager.DeleteResourceInTransaction(ctx, transaction, accountID, userID, networkID, resource.ID) if err != nil { @@ -150,11 +171,6 @@ func (m *managerImpl) DeleteNetwork(ctx context.Context, accountID, userID, netw eventsToStore = append(eventsToStore, event...) } - netRouters, err := transaction.GetNetworkRoutersByNetID(ctx, store.LockingStrengthUpdate, accountID, networkID) - if err != nil { - return fmt.Errorf("failed to get routers in network: %w", err) - } - for _, router := range netRouters { event, err := m.routersManager.DeleteRouterInTransaction(ctx, transaction, accountID, userID, networkID, router.ID) if err != nil { diff --git a/management/server/peer.go b/management/server/peer.go index e1fe08d1b..0d7c32f8a 100644 --- a/management/server/peer.go +++ b/management/server/peer.go @@ -1453,26 +1453,8 @@ func (am *DefaultAccountManager) expandAndUpdateAffected(ctx context.Context, ac 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) - } + affectedPeerIDs := unionStrings(snap.Expand(ctx, accountID, change), directlyAffected) + am.updateAffectedPeerIDs(ctx, accountID, affectedPeerIDs) } // dispatchAffected expands a set of (snapshot, change) pairs collected across @@ -1480,28 +1462,44 @@ func (am *DefaultAccountManager) expandAndUpdateAffected(ctx context.Context, ac // 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. func (am *DefaultAccountManager) dispatchAffected(ctx context.Context, accountID string, snaps []*affectedpeers.Snapshot, changes []affectedpeers.Change) { - seen := make(map[string]struct{}) - var affectedPeerIDs []string + var lists [][]string for i, snap := range snaps { if snap == nil { continue } - for _, id := range snap.Expand(ctx, accountID, changes[i]) { - if _, ok := seen[id]; !ok { - seen[id] = struct{}{} - affectedPeerIDs = append(affectedPeerIDs, id) - } - } + 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("dispatchAffected: account %s updating %d affected peers: %v", accountID, len(affectedPeerIDs), affectedPeerIDs) + log.WithContext(ctx).Debugf("updating %d affected peers for account %s: %v", len(affectedPeerIDs), accountID, affectedPeerIDs) am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs) } else { - log.WithContext(ctx).Tracef("dispatchAffected: account %s no affected peers", accountID) + log.WithContext(ctx).Tracef("no affected peers for account %s", accountID) } } +// unionStrings concatenates the given string lists into one deduplicated slice, +// preserving first-occurrence order. +func unionStrings(lists ...[]string) []string { + seen := make(map[string]struct{}) + var out []string + for _, list := range lists { + for _, id := range list { + if _, ok := seen[id]; ok { + continue + } + seen[id] = struct{}{} + out = append(out, id) + } + } + 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) @@ -1515,17 +1513,7 @@ func (am *DefaultAccountManager) resolvePeerIDs(ctx context.Context, s store.Sto return peerIDs } - seen := make(map[string]struct{}, len(peerIDs)) - for _, id := range peerIDs { - seen[id] = struct{}{} - } - for _, id := range directPeerIDs { - if _, exists := seen[id]; !exists { - peerIDs = append(peerIDs, id) - seen[id] = struct{}{} - } - } - + peerIDs = unionStrings(peerIDs, directPeerIDs) log.WithContext(ctx).Tracef("resolvePeerIDs: groups=%v + directPeers=%v -> %d peers: %v", groupIDs, directPeerIDs, len(peerIDs), peerIDs) return peerIDs } diff --git a/management/server/policy.go b/management/server/policy.go index f72679e81..a39ee0d67 100644 --- a/management/server/policy.go +++ b/management/server/policy.go @@ -76,7 +76,13 @@ func (am *DefaultAccountManager) SavePolicy(ctx context.Context, accountID, user } } - change = affectedpeers.Change{Policies: []*types.Policy{policy, existingPolicy}} + // On update carry both the old and new policy so peers losing access via a + // removed rule still refresh; on create there is no prior policy. + if isUpdate { + change = affectedpeers.Change{Policies: []*types.Policy{existingPolicy, policy}} + } else { + change = affectedpeers.Change{Policies: []*types.Policy{policy}} + } if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil { return err }