diff --git a/management/server/affectedpeers/resolver.go b/management/server/affectedpeers/resolver.go index 34f05ae80..333df5ad7 100644 --- a/management/server/affectedpeers/resolver.go +++ b/management/server/affectedpeers/resolver.go @@ -415,7 +415,7 @@ func (r *resolver) collectResourceRouterBridge() { } func (r *resolver) bridgeSourceToRouters() { - resourceIDs := policyDestinationResourceIDs(r.ctx, r.store, r.accountID, r.matchedPolicies...) + resourceIDs := r.policyDestinationResourceIDs(r.matchedPolicies...) for id := range r.resourceIDs { resourceIDs[id] = struct{}{} } @@ -514,40 +514,48 @@ func (r *resolver) policyTargetsResources(policy *types.Policy, resourceIDs map[ return false } -func policyDestinationResourceIDs(ctx context.Context, s store.Store, accountID string, policies ...*types.Policy) map[string]struct{} { - destGroupSet := make(map[string]struct{}) +func (r *resolver) policyDestinationResourceIDs(policies ...*types.Policy) map[string]struct{} { resourceIDs := make(map[string]struct{}) + destGroupSet := collectPolicyDestinations(resourceIDs, policies...) + r.addGroupResourceIDs(destGroupSet, resourceIDs) + return resourceIDs +} +// collectPolicyDestinations adds each rule's direct destination resource IDs to +// resourceIDs and returns the set of destination group IDs referenced. +func collectPolicyDestinations(resourceIDs map[string]struct{}, policies ...*types.Policy) map[string]struct{} { + destGroupSet := make(map[string]struct{}) for _, policy := range policies { if policy == nil { continue } for _, rule := range policy.Rules { - for _, gID := range rule.Destinations { - destGroupSet[gID] = struct{}{} - } + addAll(destGroupSet, rule.Destinations) if rule.DestinationResource.Type != types.ResourceTypePeer && rule.DestinationResource.ID != "" { resourceIDs[rule.DestinationResource.ID] = struct{}{} } } } + return destGroupSet +} - if len(destGroupSet) > 0 { - groups, err := s.GetGroupsByIDs(ctx, store.LockingStrengthNone, accountID, setToSlice(destGroupSet)) - if err != nil { - log.WithContext(ctx).Errorf("failed to get destination groups for resource router bridge: %v", err) - } else { - for _, group := range groups { - for _, res := range group.Resources { - if res.ID != "" { - resourceIDs[res.ID] = struct{}{} - } - } +// 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 _, res := range group.Resources { + if res.ID != "" { + resourceIDs[res.ID] = struct{}{} } } } - - return resourceIDs } func collectPolicyDirectPeers(policy *types.Policy, peerSet map[string]struct{}) { diff --git a/management/server/group.go b/management/server/group.go index aba4bf11a..a8c7593ff 100644 --- a/management/server/group.go +++ b/management/server/group.go @@ -152,18 +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) - - for _, peerID := range peersToAdd { - if err := transaction.AddPeerToGroup(ctx, accountID, peerID, newGroup.ID); err != nil { - return status.Errorf(status.Internal, "failed to add peer %s to group %s: %v", peerID, newGroup.ID, err) - } - } - for _, peerID := range peersToRemove { - if err := transaction.RemovePeerFromGroup(ctx, peerID, newGroup.ID); err != nil { - return status.Errorf(status.Internal, "failed to remove peer %s from group %s: %v", peerID, newGroup.ID, err) - } + if err = syncGroupMembership(ctx, transaction, accountID, newGroup.ID, util.Difference(newGroup.Peers, oldGroup.Peers), peersToRemove); err != nil { + return err } if err = transaction.UpdateGroup(ctx, newGroup); err != nil { @@ -196,6 +187,21 @@ func (am *DefaultAccountManager) UpdateGroup(ctx context.Context, accountID, use return nil } +// syncGroupMembership applies the peer membership delta for a group within a transaction. +func syncGroupMembership(ctx context.Context, transaction store.Store, accountID, groupID string, peersToAdd, peersToRemove []string) error { + for _, peerID := range peersToAdd { + if err := transaction.AddPeerToGroup(ctx, accountID, peerID, groupID); err != nil { + return status.Errorf(status.Internal, "failed to add peer %s to group %s: %v", peerID, groupID, err) + } + } + for _, peerID := range peersToRemove { + if err := transaction.RemovePeerFromGroup(ctx, peerID, groupID); err != nil { + return status.Errorf(status.Internal, "failed to remove peer %s from group %s: %v", peerID, groupID, err) + } + } + return nil +} + // CreateGroups adds new groups to the account. // Note: This function does not acquire the global lock. // It is the caller's responsibility to ensure proper locking is in place before invoking this method. @@ -443,20 +449,9 @@ func (am *DefaultAccountManager) DeleteGroups(ctx context.Context, accountID, us } err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { - for _, groupID := range groupIDs { - group, err := transaction.GetGroupByID(ctx, store.LockingStrengthNone, accountID, groupID) - if err != nil { - allErrors = errors.Join(allErrors, err) - continue - } - - if err = validateDeleteGroup(ctx, transaction, group, userID, extraSettings.FlowGroups); err != nil { - allErrors = errors.Join(allErrors, err) - continue - } - - groupIDsToDelete = append(groupIDsToDelete, groupID) - deletedGroups = append(deletedGroups, group) + deletedGroups, allErrors = collectDeletableGroups(ctx, transaction, accountID, userID, groupIDs, extraSettings.FlowGroups) + for _, group := range deletedGroups { + groupIDsToDelete = append(groupIDsToDelete, group.ID) } if len(groupIDsToDelete) == 0 { @@ -493,6 +488,26 @@ func (am *DefaultAccountManager) DeleteGroups(ctx context.Context, accountID, us return allErrors } +// collectDeletableGroups loads and validates each group for deletion, returning +// the groups that may be deleted and the joined validation errors for the rest. +func collectDeletableGroups(ctx context.Context, transaction store.Store, accountID, userID string, groupIDs, flowGroups []string) ([]*types.Group, error) { + var deletable []*types.Group + var allErrors error + for _, groupID := range groupIDs { + group, err := transaction.GetGroupByID(ctx, store.LockingStrengthNone, accountID, groupID) + if err != nil { + allErrors = errors.Join(allErrors, err) + continue + } + if err = validateDeleteGroup(ctx, transaction, group, userID, flowGroups); err != nil { + allErrors = errors.Join(allErrors, err) + continue + } + deletable = append(deletable, group) + } + return deletable, allErrors +} + // GroupAddPeer appends peer to the group func (am *DefaultAccountManager) GroupAddPeer(ctx context.Context, accountID, groupID, peerID string) error { var affectedPeerIDs []string