This commit is contained in:
pascal
2026-06-05 16:29:53 +02:00
parent 4360c49108
commit ffdae35a1f
8 changed files with 86 additions and 74 deletions
@@ -512,10 +512,14 @@ func (c *Controller) BufferUpdateAffectedPeers(ctx context.Context, accountID st
b.stopTimer() 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() collected := b.drainPeerIDs()
go func() { go func() {
defer b.sendMu.Unlock() defer b.sendMu.Unlock()
_ = c.sendUpdateForAffectedPeers(ctx, accountID, collected) _ = c.sendUpdateForAffectedPeers(bgCtx, accountID, collected)
// Check if more peer IDs accumulated while we were sending. // Check if more peer IDs accumulated while we were sending.
if !b.hasPending() { if !b.hasPending() {
@@ -526,7 +530,7 @@ func (c *Controller) BufferUpdateAffectedPeers(ctx context.Context, accountID st
b.setTimer(time.Duration(c.updateAccountPeersBufferInterval.Load()), func() { b.setTimer(time.Duration(c.updateAccountPeersBufferInterval.Load()), func() {
ids := b.drainPeerIDs() ids := b.drainPeerIDs()
if len(ids) > 0 { 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 { 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 nil
} }
return c.BufferUpdateAffectedPeers(ctx, accountID, affectedPeerIDs, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationDelete}) return c.BufferUpdateAffectedPeers(ctx, accountID, affectedPeerIDs, types.UpdateReason{Resource: types.UpdateResourcePeer, Operation: types.UpdateOperationDelete})
@@ -8,6 +8,8 @@ import (
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/management/server/affectedpeers" "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/posture"
"github.com/netbirdio/netbird/management/server/types" "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) { 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) _, err := s.manager.SavePolicy(ctx, s.accountID, userID, peerToResourcePolicyByGroup(s.sourceGroupID, s.resourceGroupID), true)
require.NoError(t, err) 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} []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) { 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) _, err := s.manager.SavePolicy(ctx, s.accountID, userID, peerToResourcePolicyByGroup(s.sourceGroupID, s.resourceGroupID), true)
require.NoError(t, err) 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} []string{s.sourcePeerID, s.routerPeerID}, []string{s.unrelatedPeerID}
}, },
}, },
@@ -13,6 +13,7 @@ import (
"github.com/netbirdio/netbird/management/server/affectedpeers" "github.com/netbirdio/netbird/management/server/affectedpeers"
resourceTypes "github.com/netbirdio/netbird/management/server/networks/resources/types" resourceTypes "github.com/netbirdio/netbird/management/server/networks/resources/types"
routerTypes "github.com/netbirdio/netbird/management/server/networks/routers/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/store"
"github.com/netbirdio/netbird/management/server/types" "github.com/netbirdio/netbird/management/server/types"
"github.com/netbirdio/netbird/route" "github.com/netbirdio/netbird/route"
@@ -123,9 +124,9 @@ func TestAffectedPeers_QueryCount_NarrowChangeSkipsLoads(t *testing.T) {
cs := newCountingStore(s.manager.Store) 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. // 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) require.NoError(t, err)
assert.Equal(t, 0, cs.count("routes"), "routes must not be loaded for a network-only change") assert.Equal(t, 0, cs.count("routes"), "routes must not be loaded for a network-only change")
+4 -13
View File
@@ -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 // Changed resources contribute their group IDs to the changed-group set during
// the walk (see collectFromExplicitResources), so they drive the group walkers. // the walk (see collectFromExplicitResources), so they drive the group walkers.
hasGroupOrPeerChange := len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 || len(c.Resources) > 0 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 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 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 Resources []*resourceTypes.NetworkResource
Networks []*networkTypes.Network Networks []*networkTypes.Network
PostureCheckIDs []string PostureCheckIDs []string
ResourceIDs []string
NetworkIDs []string
// RemovedPeersByGroup carries peers that left a group during this change, // RemovedPeersByGroup carries peers that left a group during this change,
// keyed by the group they left. A membership change does not alter which // 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.Resources) == 0 &&
len(c.Networks) == 0 && len(c.Networks) == 0 &&
len(c.PostureCheckIDs) == 0 && len(c.PostureCheckIDs) == 0 &&
len(c.ResourceIDs) == 0 &&
len(c.NetworkIDs) == 0 &&
len(c.RemovedPeersByGroup) == 0 len(c.RemovedPeersByGroup) == 0
} }
@@ -178,8 +174,8 @@ func (snap *Snapshot) Expand(ctx context.Context, accountID string, c Change) []
return nil return nil
} }
r := newResolver(ctx, snap, accountID, c) 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", 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, c.ResourceIDs, c.NetworkIDs) accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, len(c.Policies), len(c.Routes), len(c.Routers), len(c.Resources), len(c.Networks), c.PostureCheckIDs)
r.walk() r.walk()
return r.expand() return r.expand()
} }
@@ -225,8 +221,7 @@ func newResolver(ctx context.Context, snap *Snapshot, accountID string, c Change
changedPeerSet: toSet(c.ChangedPeerIDs), changedPeerSet: toSet(c.ChangedPeerIDs),
groupSet: make(map[string]struct{}), groupSet: make(map[string]struct{}),
peerSet: make(map[string]struct{}), peerSet: make(map[string]struct{}),
resourceIDs: toSet(c.ResourceIDs), networkIDs: make(map[string]struct{}),
networkIDs: toSet(c.NetworkIDs),
} }
// A changed peer affects every entity referencing a group it belongs to, so // 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 // 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{} peerSet map[string]struct{}
matchedPolicies []*types.Policy matchedPolicies []*types.Policy
resourceIDs map[string]struct{}
networkIDs map[string]struct{} networkIDs map[string]struct{}
} }
@@ -588,9 +582,6 @@ func (r *resolver) collectResourceRouterBridge() {
func (r *resolver) bridgeSourceToRouters() { func (r *resolver) bridgeSourceToRouters() {
resourceIDs := r.policyDestinationResourceIDs(r.matchedPolicies...) resourceIDs := r.policyDestinationResourceIDs(r.matchedPolicies...)
for id := range r.resourceIDs {
resourceIDs[id] = struct{}{}
}
if len(resourceIDs) == 0 { if len(resourceIDs) == 0 {
return return
} }
@@ -5,6 +5,8 @@ import (
"github.com/stretchr/testify/assert" "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" "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{ChangedGroupIDs: []string{"g"}}.isEmpty())
assert.False(t, Change{ChangedPeerIDs: []string{"p"}}.isEmpty()) assert.False(t, Change{ChangedPeerIDs: []string{"p"}}.isEmpty())
assert.False(t, Change{Policies: []*types.Policy{{}}}.isEmpty()) assert.False(t, Change{Policies: []*types.Policy{{}}}.isEmpty())
assert.False(t, Change{ResourceIDs: []string{"r"}}.isEmpty()) assert.False(t, Change{Resources: []*resourceTypes.NetworkResource{{ID: "r"}}}.isEmpty())
assert.False(t, Change{NetworkIDs: []string{"n"}}.isEmpty()) assert.False(t, Change{Networks: []*networkTypes.Network{{ID: "n"}}}.isEmpty())
assert.False(t, Change{PostureCheckIDs: []string{"pc"}}.isEmpty()) assert.False(t, Change{PostureCheckIDs: []string{"pc"}}.isEmpty())
} }
+26 -10
View File
@@ -131,17 +131,38 @@ func (m *managerImpl) DeleteNetwork(ctx context.Context, accountID, userID, netw
var snap *affectedpeers.Snapshot var snap *affectedpeers.Snapshot
change := affectedpeers.Change{Networks: []*types.Network{network}} change := affectedpeers.Change{Networks: []*types.Network{network}}
err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { 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. // Load before the cascade deletes: pre-state still references the network.
var lerr error var lerr error
if snap, lerr = affectedpeers.Load(ctx, transaction, accountID, change); lerr != nil { if snap, lerr = affectedpeers.Load(ctx, transaction, accountID, change); lerr != nil {
return lerr 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 { for _, resource := range resources {
event, err := m.resourcesManager.DeleteResourceInTransaction(ctx, transaction, accountID, userID, networkID, resource.ID) event, err := m.resourcesManager.DeleteResourceInTransaction(ctx, transaction, accountID, userID, networkID, resource.ID)
if err != nil { if err != nil {
@@ -150,11 +171,6 @@ func (m *managerImpl) DeleteNetwork(ctx context.Context, accountID, userID, netw
eventsToStore = append(eventsToStore, event...) 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 { for _, router := range netRouters {
event, err := m.routersManager.DeleteRouterInTransaction(ctx, transaction, accountID, userID, networkID, router.ID) event, err := m.routersManager.DeleteRouterInTransaction(ctx, transaction, accountID, userID, networkID, router.ID)
if err != nil { if err != nil {
+29 -41
View File
@@ -1453,26 +1453,8 @@ func (am *DefaultAccountManager) expandAndUpdateAffected(ctx context.Context, ac
if snap == nil { if snap == nil {
return return
} }
affectedPeerIDs := snap.Expand(ctx, accountID, change) affectedPeerIDs := unionStrings(snap.Expand(ctx, accountID, change), directlyAffected)
if len(directlyAffected) > 0 { am.updateAffectedPeerIDs(ctx, accountID, affectedPeerIDs)
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)
}
} }
// dispatchAffected expands a set of (snapshot, change) pairs collected across // 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 // 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. // 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) { func (am *DefaultAccountManager) dispatchAffected(ctx context.Context, accountID string, snaps []*affectedpeers.Snapshot, changes []affectedpeers.Change) {
seen := make(map[string]struct{}) var lists [][]string
var affectedPeerIDs []string
for i, snap := range snaps { for i, snap := range snaps {
if snap == nil { if snap == nil {
continue continue
} }
for _, id := range snap.Expand(ctx, accountID, changes[i]) { lists = append(lists, snap.Expand(ctx, accountID, changes[i]))
if _, ok := seen[id]; !ok {
seen[id] = struct{}{}
affectedPeerIDs = append(affectedPeerIDs, id)
}
}
} }
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 { 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) am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs)
} else { } 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. // 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 { func (am *DefaultAccountManager) resolvePeerIDs(ctx context.Context, s store.Store, accountID string, groupIDs []string, directPeerIDs []string) []string {
peerIDs, err := s.GetPeerIDsByGroups(ctx, accountID, groupIDs) peerIDs, err := s.GetPeerIDsByGroups(ctx, accountID, groupIDs)
@@ -1515,17 +1513,7 @@ func (am *DefaultAccountManager) resolvePeerIDs(ctx context.Context, s store.Sto
return peerIDs return peerIDs
} }
seen := make(map[string]struct{}, len(peerIDs)) peerIDs = unionStrings(peerIDs, directPeerIDs)
for _, id := range peerIDs {
seen[id] = struct{}{}
}
for _, id := range directPeerIDs {
if _, exists := seen[id]; !exists {
peerIDs = append(peerIDs, id)
seen[id] = struct{}{}
}
}
log.WithContext(ctx).Tracef("resolvePeerIDs: groups=%v + directPeers=%v -> %d peers: %v", groupIDs, directPeerIDs, len(peerIDs), peerIDs) log.WithContext(ctx).Tracef("resolvePeerIDs: groups=%v + directPeers=%v -> %d peers: %v", groupIDs, directPeerIDs, len(peerIDs), peerIDs)
return peerIDs return peerIDs
} }
+7 -1
View File
@@ -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 { if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil {
return err return err
} }