sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-09-07 15:31:30 +02:00
Code Issues Packages Projects Releases Wiki Activity

cleanup comments

Browse Source
This commit is contained in:
pascal
2026-06-09 16:44:47 +02:00
parent 67a806a0db
commit 30d78c841c
3 changed files with 86 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
management/server/affected_peers_router_test.go
View File

@@ -184,32 +184,23 @@ func peerToResourcePolicyByResource(sourceGroupID, resourceID string) *types.Pol
}
}
// ---------------------------------------------------------------------------
// Resolution-level tests: collectPolicyAffectedGroupsAndPeers + resolvePeerIDs.
//
// These isolate the resolver from the controller and assert directly on the set
// of peer IDs the policy path would refresh. They make the gap explicit: the
// routing peer is expected to be in the affected set but is not.
// ---------------------------------------------------------------------------
// resolvePolicyAffected mirrors SavePolicy's resolution: resolve the affected
// peers for the given policy.
func (s *routerScenario) resolvePolicyAffected(ctx context.Context, policy *types.Policy) []string {
return s.manager.ResolveAffectedPeers(ctx, s.manager.Store, s.accountID, affectedpeers.Change{Policies: []*types.Policy{policy}})
}
func TestAffectedPeers_PolicyToResourceByGroup_IncludesSourcePeer_DirectRouter(t *testing.T) {
func TestAffectedPeers_SourcePeer_DirectRouter(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
policy := peerToResourcePolicyByGroup(s.sourceGroupID, s.resourceGroupID)
affected := s.resolvePolicyAffected(ctx, policy)
// The source peer is in the source group and must always be present.
assert.Contains(t, affected, s.sourcePeerID, "source peer must be affected")
}
func TestAffectedPeers_PolicyToResourceByGroup_IncludesRoutingPeer_DirectRouter(t *testing.T) {
func TestAffectedPeers_RoutingPeer_DirectRouter(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -225,19 +216,19 @@ func TestAffectedPeers_PolicyToResourceByGroup_IncludesRoutingPeer_DirectRouter(
"routing peer (router.Peer) serving the resource must be affected by a policy granting access to it")
}
func TestAffectedPeers_PolicyToResourceByGroup_IncludesRoutingPeer_RouterPeerGroups(t *testing.T) {
func TestAffectedPeers_RoutingPeer_RouterPeerGroups(t *testing.T) {
s := setupRouterScenario(t, false)
ctx := context.Background()
policy := peerToResourcePolicyByGroup(s.sourceGroupID, s.resourceGroupID)
affected := s.resolvePolicyAffected(ctx, policy)
// Same gap when the router is defined via PeerGroups instead of a direct peer.
// Router defined via PeerGroups instead of a direct peer.
assert.Contains(t, affected, s.routerGroupPeerID,
"routing peer (router.PeerGroups member) serving the resource must be affected")
}
func TestAffectedPeers_PolicyToResourceByDestinationResource_IncludesRoutingPeer_DirectRouter(t *testing.T) {
func TestAffectedPeers_DestResource_RoutingPeer_DirectRouter(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -252,7 +243,7 @@ func TestAffectedPeers_PolicyToResourceByDestinationResource_IncludesRoutingPeer
"routing peer must be affected when the resource is referenced via DestinationResource")
}
func TestAffectedPeers_PolicyToResourceByDestinationResource_IncludesRoutingPeer_RouterPeerGroups(t *testing.T) {
func TestAffectedPeers_DestResource_RoutingPeer_RouterPeerGroups(t *testing.T) {
s := setupRouterScenario(t, false)
ctx := context.Background()
@@ -263,7 +254,7 @@ func TestAffectedPeers_PolicyToResourceByDestinationResource_IncludesRoutingPeer
"routing peer (PeerGroups) must be affected when the resource is referenced via DestinationResource")
}
func TestAffectedPeers_PolicyToResourceWithSourceResourcePeer_IncludesRoutingPeer(t *testing.T) {
func TestAffectedPeers_SourceResourcePeer_RoutingPeer(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -300,12 +291,6 @@ func TestAffectedPeers_PolicyToResource_UnrelatedPeerNotAffected(t *testing.T) {
assert.NotContains(t, affected, s.unrelatedPeerID, "unrelated peer must not be affected")
}
// ---------------------------------------------------------------------------
// Control: the resource/router managers DO bridge resource-group -> router.
// These document the existing (correct) behaviour on the resource side and
// highlight the asymmetry with the policy side above.
// ---------------------------------------------------------------------------
func TestAffectedPeers_ResourceSideBridgesToRoutingPeer_DirectRouter(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -355,22 +340,12 @@ func TestAffectedPeers_ResourceSideBridgesToRoutingPeer_DirectRouter(t *testing.
}
}
// ---------------------------------------------------------------------------
// End-to-end: reproduce the reported symptom through SavePolicy with channels.
//
// Creating the peer->resource policy must wake the routing peer. These fail on
// the current code because the policy path never resolves the routing peer.
//
// IMPORTANT: setup (CreateNetwork/CreateResource/CreateRouter) fires async
// `go UpdateAffectedPeers` goroutines. Channels are opened and then drained via
// settleAffectedUpdates before the action under test, so the assertion only
// observes updates caused by that action and not stragglers from setup. The
// resolution-level tests above are the timing-free, authoritative proof; these
// reproduce the operator-visible symptom.
// ---------------------------------------------------------------------------
// settleAffectedUpdates waits for in-flight async updates to arrive, then drains
// every given channel so subsequent assertions start from a clean slate.
//
// Setup (CreateNetwork/CreateResource/CreateRouter) fires async UpdateAffectedPeers
// goroutines; draining first means the assertion only observes updates from the
// action under test, not setup stragglers.
func settleAffectedUpdates(chans ...<-chan *network_map.UpdateMessage) {
time.Sleep(300 * time.Millisecond)
for _, ch := range chans {
@@ -378,7 +353,7 @@ func settleAffectedUpdates(chans ...<-chan *network_map.UpdateMessage) {
}
}
func TestAffectedPeers_E2E_CreatePolicyToResource_RefreshesRoutingPeer_DirectRouter(t *testing.T) {
func TestAffectedPeers_E2E_CreatePolicy_RoutingPeer_DirectRouter(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -396,7 +371,7 @@ func TestAffectedPeers_E2E_CreatePolicyToResource_RefreshesRoutingPeer_DirectRou
done := make(chan struct{})
go func() {
peerShouldReceiveUpdate(t, srcCh)
peerShouldReceiveUpdate(t, routerCh) // FAILS today: routing peer not resolved
peerShouldReceiveUpdate(t, routerCh)
peerShouldNotReceiveUpdate(t, unrelatedCh)
close(done)
}()
@@ -411,7 +386,7 @@ func TestAffectedPeers_E2E_CreatePolicyToResource_RefreshesRoutingPeer_DirectRou
}
}
func TestAffectedPeers_E2E_CreatePolicyToResource_RefreshesRoutingPeer_RouterPeerGroups(t *testing.T) {
func TestAffectedPeers_E2E_CreatePolicy_RoutingPeer_RouterPeerGroups(t *testing.T) {
s := setupRouterScenario(t, false)
ctx := context.Background()
@@ -427,7 +402,7 @@ func TestAffectedPeers_E2E_CreatePolicyToResource_RefreshesRoutingPeer_RouterPee
done := make(chan struct{})
go func() {
peerShouldReceiveUpdate(t, srcCh)
peerShouldReceiveUpdate(t, routerCh) // FAILS today
peerShouldReceiveUpdate(t, routerCh)
close(done)
}()
@@ -441,7 +416,7 @@ func TestAffectedPeers_E2E_CreatePolicyToResource_RefreshesRoutingPeer_RouterPee
}
}
func TestAffectedPeers_E2E_CreatePolicyByDestinationResource_RefreshesRoutingPeer(t *testing.T) {
func TestAffectedPeers_E2E_DestResource_RoutingPeer(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -457,7 +432,7 @@ func TestAffectedPeers_E2E_CreatePolicyByDestinationResource_RefreshesRoutingPee
done := make(chan struct{})
go func() {
peerShouldReceiveUpdate(t, srcCh)
peerShouldReceiveUpdate(t, routerCh) // FAILS today
peerShouldReceiveUpdate(t, routerCh)
close(done)
}()
@@ -471,7 +446,7 @@ func TestAffectedPeers_E2E_CreatePolicyByDestinationResource_RefreshesRoutingPee
}
}
func TestAffectedPeers_E2E_DeletePolicyToResource_RefreshesRoutingPeer(t *testing.T) {
func TestAffectedPeers_E2E_DeletePolicy_RoutingPeer(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -490,7 +465,7 @@ func TestAffectedPeers_E2E_DeletePolicyToResource_RefreshesRoutingPeer(t *testin
done := make(chan struct{})
go func() {
peerShouldReceiveUpdate(t, srcCh)
peerShouldReceiveUpdate(t, routerCh) // FAILS today: deleting the policy must also refresh the router
peerShouldReceiveUpdate(t, routerCh)
close(done)
}()
@@ -568,7 +543,7 @@ func (s *routerScenario) addSecondTopology(t *testing.T, suffix string) secondTo
}
}
func TestAffectedPeers_E2E_UpdatePolicyRepointResource_RefreshesBothRoutingPeers(t *testing.T) {
func TestAffectedPeers_E2E_UpdatePolicy_BothRoutingPeers(t *testing.T) {
s := setupRouterScenario(t, true)
second := s.addSecondTopology(t, "b")
ctx := context.Background()
@@ -606,7 +581,7 @@ func TestAffectedPeers_E2E_UpdatePolicyRepointResource_RefreshesBothRoutingPeers
}
}
func TestAffectedPeers_E2E_UpdatePolicyAddSourceGroup_RefreshesRoutingPeer(t *testing.T) {
func TestAffectedPeers_E2E_UpdatePolicy_AddSource(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -648,7 +623,7 @@ func TestAffectedPeers_E2E_UpdatePolicyAddSourceGroup_RefreshesRoutingPeer(t *te
}
}
func TestAffectedPeers_E2E_CreatePolicyByDestinationResource_RefreshesRoutingPeer_RouterPeerGroups(t *testing.T) {
func TestAffectedPeers_E2E_DestResource_RouterPeerGroups(t *testing.T) {
s := setupRouterScenario(t, false)
ctx := context.Background()
@@ -678,7 +653,7 @@ func TestAffectedPeers_E2E_CreatePolicyByDestinationResource_RefreshesRoutingPee
}
}
func TestAffectedPeers_PolicyToResource_IncludesAllRoutingPeersOnNetwork(t *testing.T) {
func TestAffectedPeers_AllRoutingPeers_Network(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -702,7 +677,7 @@ func TestAffectedPeers_PolicyToResource_IncludesAllRoutingPeersOnNetwork(t *test
assert.Contains(t, affected, secondRouterPeer.ID, "second routing peer on the same network must also be affected")
}
func TestAffectedPeers_PolicyToResource_DisabledRouterStillAffected(t *testing.T) {
func TestAffectedPeers_DisabledRouter(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -719,7 +694,7 @@ func TestAffectedPeers_PolicyToResource_DisabledRouterStillAffected(t *testing.T
"disabled router's peer must still be affected: Enabled must not gate affected-peers")
}
func TestAffectedPeers_PolicyToResource_DisabledResourceStillAffected(t *testing.T) {
func TestAffectedPeers_DisabledResource(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -735,7 +710,7 @@ func TestAffectedPeers_PolicyToResource_DisabledResourceStillAffected(t *testing
"disabled resource must still resolve the routing peer: Enabled must not gate affected-peers")
}
func TestAffectedPeers_PolicyToResource_DisabledRuleStillAffected(t *testing.T) {
func TestAffectedPeers_DisabledRule(t *testing.T) {
s := setupRouterScenario(t, true)
ctx := context.Background()
@@ -748,7 +723,7 @@ func TestAffectedPeers_PolicyToResource_DisabledRuleStillAffected(t *testing.T)
"disabled rule must still resolve the routing peer: Enabled must not gate affected-peers")
}
func TestAffectedPeers_MultiRulePolicy_IncludesAllRoutingPeers(t *testing.T) {
func TestAffectedPeers_MultiRule(t *testing.T) {
s := setupRouterScenario(t, true)
second := s.addSecondTopology(t, "c")
ctx := context.Background()
@@ -778,7 +753,7 @@ func TestAffectedPeers_MultiRulePolicy_IncludesAllRoutingPeers(t *testing.T) {
assert.Contains(t, affected, second.routerPeerID, "routing peer for resource B must be affected")
}
func TestAffectedPeers_PolicyToResource_RouterInOtherNetworkNotAffected(t *testing.T) {
func TestAffectedPeers_RouterOtherNetwork(t *testing.T) {
s := setupRouterScenario(t, true)
second := s.addSecondTopology(t, "d")
ctx := context.Background()

55
management/server/affected_peers_test.go
View File

@@ -443,10 +443,6 @@ func TestCollectGroupChange_MultipleNameServerGroups_OnlyLinkedAffected(t *testi
// Pure policy/route/router extraction unit tests moved to the affectedpeers
// package (management/server/affectedpeers) along with the logic they cover.
// ---------------------------------------------------------------------------
// resolvePeerIDs tests
// ---------------------------------------------------------------------------
func TestResolvePeerIDs_GroupsOnly(t *testing.T) {
manager, s, accountID, peerIDs, groupIDs := setupAffectedPeersTest(t)
ctx := context.Background()
@@ -480,10 +476,6 @@ func TestResolvePeerIDs_EmptyInputs(t *testing.T) {
assert.Empty(t, result)
}
// ---------------------------------------------------------------------------
// resolveAffectedPeersForPeerChanges tests
// ---------------------------------------------------------------------------
func TestResolveAffectedPeers_NoPoliciesOrRoutes(t *testing.T) {
manager, s, accountID, peerIDs, _ := setupAffectedPeersTest(t)
ctx := context.Background()
@@ -849,10 +841,6 @@ func TestResolveAffectedPeers_NoDuplicates(t *testing.T) {
assert.Equal(t, 1, count, "peer0 should appear exactly once")
}
// ---------------------------------------------------------------------------
// Posture check affected peers tests
// ---------------------------------------------------------------------------
func TestCollectPostureCheckAffected_NoMatch(t *testing.T) {
_, s, accountID, _, _ := setupAffectedPeersTest(t)
ctx := context.Background()
@@ -900,10 +888,6 @@ func TestCollectPostureCheckAffected_LinkedToPolicy(t *testing.T) {
assert.Empty(t, directPeers)
}
// ---------------------------------------------------------------------------
// Isolation tests: verify peers NOT in any relevant entity are NOT affected
// ---------------------------------------------------------------------------
func TestAffectedPeers_IsolatedPolicies(t *testing.T) {
manager, s, accountID, peerIDs, groupIDs := setupAffectedPeersTest(t)
ctx := context.Background()
@@ -998,10 +982,6 @@ func TestAffectedPeers_IsolatedRouteAndPolicy(t *testing.T) {
assert.NotContains(t, result, peerIDs[1])
}
// ---------------------------------------------------------------------------
// Integration tests with update channels (peerShouldReceiveUpdate / peerShouldNotReceiveUpdate)
// ---------------------------------------------------------------------------
func TestAffectedPeers_GroupUpdateOnlyAffectsLinkedPeers(t *testing.T) {
manager, updateManager, account, peer1, peer2, peer3 := setupNetworkMapTest(t)
ctx := context.Background()
@@ -1049,7 +1029,6 @@ func TestAffectedPeers_GroupUpdateOnlyAffectsLinkedPeers(t *testing.T) {
result := manager.resolveAffectedPeersForPeerChanges(ctx, manager.Store, accountID, []string{peer1.ID})
assert.ElementsMatch(t, []string{peer1.ID, peer2.ID}, result)
// Adding peer3 to grpA makes it part of the policy, so all 3 peers get updated
t.Run("group change updates all peers in policy groups", func(t *testing.T) {
done := make(chan struct{})
go func() {
@@ -1072,10 +1051,6 @@ func TestAffectedPeers_GroupUpdateOnlyAffectsLinkedPeers(t *testing.T) {
t.Error("timeout")
}
})
_ = updMsg1
_ = updMsg2
_ = updMsg3
}
func TestAffectedPeers_UnlinkedGroupChange_NoUpdates(t *testing.T) {
@@ -1118,7 +1093,6 @@ func TestAffectedPeers_PolicyChange_UnrelatedPeerNoUpdate(t *testing.T) {
updateManager.CloseChannel(ctx, peer3.ID)
})
// Create policy linking only peer1 (grpA) <-> peer2 (grpB). Peer3 should not receive update.
t.Run("create policy only affects linked peers", func(t *testing.T) {
done := make(chan struct{})
go func() {
@@ -1182,7 +1156,6 @@ func TestAffectedPeers_RouteChange_UnrelatedPeerNoUpdate(t *testing.T) {
updateManager.CloseChannel(ctx, peer3.ID)
})
// Create route with peer groups grpA and distribution group grpB. Peer3 should not get update.
t.Run("create route only affects linked peers", func(t *testing.T) {
done := make(chan struct{})
go func() {
@@ -1250,7 +1223,6 @@ func TestAffectedPeers_NameServerChange_UnrelatedPeerNoUpdate(t *testing.T) {
updateManager.CloseChannel(ctx, peer3.ID)
})
// Create NS group using only grpA. peer2 and peer3 should not get update.
t.Run("create nameserver group only affects linked peers", func(t *testing.T) {
done := make(chan struct{})
go func() {
@@ -1310,7 +1282,6 @@ func TestAffectedPeers_DNSSettingsChange_UnrelatedPeerNoUpdate(t *testing.T) {
updateManager.CloseChannel(ctx, peer3.ID)
})
// Save DNS settings that only affects grpA. peer2 and peer3 should not be affected.
t.Run("dns settings change only affects linked peers", func(t *testing.T) {
done := make(chan struct{})
go func() {
@@ -1388,13 +1359,11 @@ func TestAffectedPeers_UnlinkedGroupChange_NoUpdateIntegration(t *testing.T) {
})
}
// TestAffectedPeers_NetworkRouter_UnrelatedPeerNoUpdate verifies that when a network
// router is added with specific peer groups, only peers in those groups (and policy
// sources for resources) get updates. Unrelated peers should not.
func TestAffectedPeers_NetworkRouter_UnrelatedPeerNoUpdate(t *testing.T) {
// Use custom setup: delete default policy BEFORE adding peers so that
// AddPeer's BufferUpdateAffectedPeers finds no affected peers and
// doesn't schedule async updates that race with the test.
// TestAffectedPeers_NetworkRouterUnlinkedPeerNoUpdate: a network router with peer
// groups updates only those groups' peers (and resource policy sources), not others.
func TestAffectedPeers_NetworkRouterUnlinkedPeerNoUpdate(t *testing.T) {
// Delete the default policy before adding peers so AddPeer schedules no async
// update that races with the test.
manager, updateManager, err := createManager(t)
require.NoError(t, err)
@@ -1451,9 +1420,6 @@ func TestAffectedPeers_NetworkRouter_UnrelatedPeerNoUpdate(t *testing.T) {
updateManager.CloseChannel(ctx, peer3.ID)
})
// When the group linked to the network router changes, only peers in that
// group should be updated. Peer2 is unrelated. Peer3 is added to the
// router's group so it should also receive an update.
t.Run("network router group change only affects linked peers", func(t *testing.T) {
done := make(chan struct{})
go func() {
@@ -1463,7 +1429,6 @@ func TestAffectedPeers_NetworkRouter_UnrelatedPeerNoUpdate(t *testing.T) {
close(done)
}()
// Updating the group linked to router should affect peer1 and peer3 (now in nr-grpA).
err = manager.UpdateGroup(ctx, accountID, userID, &types.Group{
ID: "nr-grpA",
Name: "NR-A",
@@ -1479,10 +1444,9 @@ func TestAffectedPeers_NetworkRouter_UnrelatedPeerNoUpdate(t *testing.T) {
})
}
// TestAffectedPeers_MultipleIsolatedEntities_OnlyLinkedPeersUpdated creates multiple
// isolated entities (policy for peer1<->peer2, route for peer3) and verifies that
// changing one entity's groups only affects its peers.
func TestAffectedPeers_MultipleIsolatedEntities_OnlyLinkedPeersUpdated(t *testing.T) {
// TestAffectedPeers_IsolatedEntitiesOnlyAffectTheirPeers: with a policy (peer1<->peer2)
// and a separate route (peer3), changing one entity's groups affects only its peers.
func TestAffectedPeers_IsolatedEntitiesOnlyAffectTheirPeers(t *testing.T) {
manager, updateManager, account, peer1, peer2, peer3 := setupNetworkMapTest(t)
ctx := context.Background()
accountID := account.Id
@@ -1503,7 +1467,6 @@ func TestAffectedPeers_MultipleIsolatedEntities_OnlyLinkedPeersUpdated(t *testin
require.NoError(t, err)
}
// Policy: peer1 <-> peer2
_, err = manager.SavePolicy(ctx, accountID, userID, &types.Policy{
Enabled: true,
Rules: []*types.PolicyRule{
@@ -1518,7 +1481,6 @@ func TestAffectedPeers_MultipleIsolatedEntities_OnlyLinkedPeersUpdated(t *testin
}, true)
require.NoError(t, err)
// Route: only peer3's group as distribution group
_, err = manager.CreateRoute(ctx, accountID,
netip.MustParsePrefix("10.20.0.0/24"),
route.IPv4Network,
@@ -1547,7 +1509,6 @@ func TestAffectedPeers_MultipleIsolatedEntities_OnlyLinkedPeersUpdated(t *testin
updateManager.CloseChannel(ctx, peer3.ID)
})
// Updating policy group (iso-grpA) should affect peer1+peer2 but NOT peer3
t.Run("policy group change does not affect route-only peer", func(t *testing.T) {
done := make(chan struct{})
go func() {

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

@@ -1,18 +1,12 @@
// 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.
// Package affectedpeers computes which peers' network maps a change touches, 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.
// Two phases keep the dependency walk off the write transaction:
// - Load: reads the needed collections. Call INSIDE the mutating tx (consistent,
// and before a delete/removal severs the old state).
// - Snapshot.Expand: in-memory walk, no store access. 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.
// Enabled is never consulted: toggling it is itself an observable change.
package affectedpeers
import (
@@ -30,14 +24,9 @@ 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).
// Snapshot is an in-memory view of the collections needed to expand a Change.
// Loaded in-tx, walked by Expand after commit. Only the collections the Change
// can touch are loaded; the rest stay nil (see Load).
type Snapshot struct {
policies []*types.Policy
routes []*route.Route
@@ -47,26 +36,22 @@ type Snapshot struct {
resources []*resourceTypes.NetworkResource
services []*rpservice.Service
proxyByCluster map[string][]string
groups map[string]*types.Group // all groups (for group.Resources lookups)
groups map[string]*types.Group
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).
// Load reads the collections a Change requires, inside the caller's tx. It mirrors
// Expand's walker preconditions, loading only what the change can touch.
func Load(ctx context.Context, s store.Store, accountID string, c Change) (*Snapshot, error) {
snap := &Snapshot{}
if c.isEmpty() {
return snap, nil
}
// 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.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
needsRoutersResources := needsPolicies // the resource<->router bridge can fire for any of these
var err error
if needsPolicies {
@@ -79,9 +64,6 @@ func Load(ctx context.Context, s store.Store, accountID string, c Change) (*Snap
return nil, err
}
}
// A changed peer is resolved to its groups during the walk (see
// seedChangedGroupsFromPeers), so the nameserver/DNS group walkers can fire for
// peer changes too — load those tables whenever groups or peers changed.
if len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 {
if snap.nsGroups, err = s.GetAccountNameServerGroups(ctx, store.LockingStrengthNone, accountID); err != nil {
return nil, err
@@ -109,26 +91,25 @@ func Load(ctx context.Context, s store.Store, accountID string, c Change) (*Snap
}
}
// 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))
snap.groupPeers = make(map[string]map[string]struct{}, 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
members := make(map[string]struct{}, len(g.Peers))
for _, pID := range g.Peers {
members[pID] = struct{}{}
}
snap.groupPeers[g.ID] = members
}
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.
// Change describes what changed in an account.
type Change struct {
ChangedGroupIDs []string
ChangedPeerIDs []string
@@ -139,14 +120,10 @@ type Change struct {
Networks []*networkTypes.Network
PostureCheckIDs []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: peers that left a group, keyed by that group. They are no
// longer in the group's member index but still lose its reachability, so they are
// folded in — but only when the group is linked (an unlinked group has no map
// impact), matching how current members are handled.
RemovedPeersByGroup map[string][]string
}
@@ -162,13 +139,9 @@ func (c Change) isEmpty() bool {
len(c.RemovedPeersByGroup) == 0
}
// 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.
// Expand returns the deduplicated affected peer IDs from the preloaded Snapshot,
// no store access. Run after the producing tx commits. Logs the full walk at
// trace level for diagnosing a miscalculation.
func (snap *Snapshot) Expand(ctx context.Context, accountID string, c Change) []string {
if c.isEmpty() {
return nil
@@ -180,10 +153,8 @@ func (snap *Snapshot) Expand(ctx context.Context, accountID string, c Change) []
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.
// Resolve does Load+Expand in one call, for callers not inside a tx. Tx-bound
// callers should use Load (in-tx) + Snapshot.Expand (after commit) instead.
func Resolve(ctx context.Context, s store.Store, accountID string, c Change) ([]string, error) {
if c.isEmpty() {
return nil, nil
@@ -195,8 +166,8 @@ func Resolve(ctx context.Context, s store.Store, accountID string, c Change) ([]
return snap.Expand(ctx, accountID, c), nil
}
// Collect returns the affected group IDs and direct peer IDs without expanding
// groups to members. For tests asserting on the intermediate sets; use Resolve otherwise.
// Collect returns the affected group and direct-peer IDs without expanding groups
// to members. Test-only introspection; use Resolve otherwise.
func Collect(ctx context.Context, s store.Store, accountID string, c Change) (groupIDs []string, directPeerIDs []string) {
if c.isEmpty() {
return nil, nil
@@ -223,19 +194,14 @@ func newResolver(ctx context.Context, snap *Snapshot, accountID string, c Change
peerSet: make(map[string]struct{}),
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
// group->peers index. Callers pass only ChangedPeerIDs; the peer->group lookup
// is the resolver's job, not theirs.
// Resolve each changed peer to its groups here so callers pass only ChangedPeerIDs.
r.seedChangedGroupsFromPeers()
r.matchedPolicies = append(r.matchedPolicies, c.Policies...)
return r
}
// seedChangedGroupsFromPeers adds, for each changed peer, the groups it belongs
// to into changedGroupSet, so the group-driven walkers (policies, routes,
// nameservers, DNS, routers) fire for memberships — not only for entities that
// reference the peer directly.
// seedChangedGroupsFromPeers adds each changed peer's groups to changedGroupSet so
// the group-driven walkers fire for memberships, not just direct peer references.
func (r *resolver) seedChangedGroupsFromPeers() {
if len(r.changedPeerSet) == 0 {
return
@@ -292,8 +258,7 @@ func (r *resolver) networkResources() []*resourceTypes.NetworkResource { return
func (r *resolver) networkRouters() []*routerTypes.NetworkRouter { return r.snap.routers }
// peerIDsForGroups maps a group set to its member peer IDs using the preloaded
// group->peers index (no store access).
// peerIDsForGroups maps a group set to its member peer IDs via the preloaded index.
func (r *resolver) peerIDsForGroups(groupSet map[string]struct{}) []string {
seen := make(map[string]struct{})
var ids []string
@@ -326,9 +291,7 @@ func (r *resolver) expand() []string {
}
}
// 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.
// Fold in removed peers only when their group is linked (in groupSet).
for groupID, removed := range r.change.RemovedPeersByGroup {
if _, linked := r.groupSet[groupID]; !linked {
continue
@@ -372,11 +335,9 @@ func (r *resolver) collectFromExplicitRoutes(routes []*route.Route) {
}
}
// collectFromExplicitRouters folds the routing peers carried by changed router
// objects (old and/or new state) directly, and marks their networks so the
// router<->source bridge folds the source peers of policies serving them. Carrying
// the old router object here is how a repointed router's previous routing peers
// stay affected without a post-commit read.
// collectFromExplicitRouters folds changed routers' peers and marks their networks
// for the bridge. Passing the old router keeps a repointed router's previous peers
// affected without a post-commit read.
func (r *resolver) collectFromExplicitRouters(routers []*routerTypes.NetworkRouter) {
for _, router := range routers {
if router == nil {
@@ -394,11 +355,9 @@ func (r *resolver) collectFromExplicitRouters(routers []*routerTypes.NetworkRout
}
}
// collectFromExplicitResources marks the networks of changed resource objects
// (old and/or new state) so the bridge folds their routers and the source peers
// of policies targeting them, and treats each resource's group IDs as changed
// groups so policies targeting the resource through a now-detached (old) group —
// which the post-update group.Resources no longer links — still refresh.
// collectFromExplicitResources marks changed resources' networks for the bridge and
// treats their group IDs as changed, so policies targeting the resource via a
// now-detached (old) group still refresh.
func (r *resolver) collectFromExplicitResources(resources []*resourceTypes.NetworkResource) {
for _, resource := range resources {
if resource == nil {
@@ -413,9 +372,8 @@ func (r *resolver) collectFromExplicitResources(resources []*resourceTypes.Netwo
}
}
// collectFromExplicitNetworks marks changed network objects (old and/or new
// state) so the bridge folds their routers and the source peers of policies
// serving their resources. A network has no groups/peers of its own.
// collectFromExplicitNetworks marks changed networks for the bridge. A network has
// no groups/peers of its own.
func (r *resolver) collectFromExplicitNetworks(networks []*networkTypes.Network) {
for _, network := range networks {
if network == nil {
@@ -570,11 +528,9 @@ func (r *resolver) expandChangedPeersWithGroups() map[string]struct{} {
return merged
}
// collectResourceRouterBridge folds in the routing peers serving the resources
// targeted by matched/explicit policies (source -> router), and the source peers
// of policies serving resources on the affected networks (router -> source). The
// routing peer is reachable only through resource -> network -> router, never
// through the policy's own groups, so it must be collected here.
// collectResourceRouterBridge crosses between source peers and routing peers, which
// are reachable only via resource -> network -> router, not through the policy's own
// groups: source -> router (targeted resources' networks), then router -> source.
func (r *resolver) collectResourceRouterBridge() {
r.bridgeSourceToRouters()
r.bridgeRoutersToSources()
@@ -683,8 +639,8 @@ func (r *resolver) policyDestinationResourceIDs(policies ...*types.Policy) map[s
return resourceIDs
}
// collectPolicyDestinations adds each rule's direct destination resource IDs to
// resourceIDs and returns the set of destination group IDs referenced.
// collectPolicyDestinations adds direct destination resource IDs to resourceIDs and
// returns the referenced destination group IDs.
func collectPolicyDestinations(resourceIDs map[string]struct{}, policies ...*types.Policy) map[string]struct{} {
destGroupSet := make(map[string]struct{})
for _, policy := range policies {