fix(azure): bound concurrency in the multi-subscription recommendation fan-out

providers/azure/recommendations_multi_subscription.go

@@ -12,6 +12,7 @@ import (
 	"golang.org/x/sync/errgroup"
 
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/concurrency"
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/logging"
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/provider"
 	azrecs "github.com/LeanerCloud/cloud-commitments-go/providers/azure/internal/recommendations"
@@ -213,10 +214,13 @@ func NewMultiSubscriptionRecommendationsClient(cred azcore.TokenCredential, acco
 // each per-subscription goroutine captures its own error and returns nil to
 // the group, so one subscription failing (e.g. the principal lost Reader
 // access mid-run, or a subscription-specific throttle) never cancels
-// sibling subscriptions. The semaphore that bounds aggregate concurrent ARM
-// calls is acquired inside each per-subscription client's own
-// GetRecommendations (around the outbound API calls, not around this
-// fan-out), so no additional semaphore is needed at this layer.
+// sibling subscriptions. The number of subscriptions in flight is capped with
+// g.SetLimit(concurrency.MaxParallelismFromEnv()) so a tenant with many
+// subscriptions cannot fan out unbounded. This is an errgroup limit, not a
+// second semaphore: the shared semaphore that bounds aggregate concurrent ARM
+// calls is acquired only inside each per-subscription client's leaf calls
+// (never held while waiting on this fan-out), so the two bounds cannot
+// deadlock against each other.
 //
 // After g.Wait(), ctx.Err() is checked explicitly: g.Wait() only reports
 // errors returned to the group, and every goroutine here returns nil, so a
@@ -253,6 +257,7 @@ func (m *MultiSubscriptionRecommendationsClient) GetRecommendations(ctx context.
 	errs := make([]error, len(targets))
 
 	g, gctx := errgroup.WithContext(ctx)
+	g.SetLimit(concurrency.MaxParallelismFromEnv())
 	for i, sub := range targets {
 		i, sub := i, sub
 		g.Go(func() error {

providers/azure/recommendations_multi_subscription_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/stretchr/testify/assert"
@@ -517,3 +518,62 @@ func TestNewMultiSubscriptionRecommendationsClient_DeduplicatesAccounts(t *testi
 		assert.Len(t, partial.Failed, 1)
 	})
 }
+
+// peakRecommendationsClient records the maximum number of concurrently
+// in-flight GetRecommendations calls across all instances sharing the same
+// counters.
+type peakRecommendationsClient struct {
+	inFlight *atomic.Int64
+	peak     *atomic.Int64
+}
+
+func (p *peakRecommendationsClient) GetRecommendations(ctx context.Context, _ *common.RecommendationParams) ([]common.Recommendation, error) {
+	cur := p.inFlight.Add(1)
+	defer p.inFlight.Add(-1)
+	for {
+		old := p.peak.Load()
+		if cur <= old || p.peak.CompareAndSwap(old, cur) {
+			break
+		}
+	}
+	select {
+	case <-time.After(5 * time.Millisecond):
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	return nil, nil
+}
+
+func (p *peakRecommendationsClient) GetRecommendationsForService(ctx context.Context, service common.ServiceType) ([]common.Recommendation, error) {
+	return p.GetRecommendations(ctx, &common.RecommendationParams{Service: service})
+}
+
+func (p *peakRecommendationsClient) GetAllRecommendations(ctx context.Context) ([]common.Recommendation, error) {
+	return p.GetRecommendations(ctx, &common.RecommendationParams{})
+}
+
+// The fan-out must not exceed CUDLY_MAX_PARALLELISM concurrent subscriptions,
+// otherwise a tenant with many subscriptions issues unbounded ARM calls.
+func TestMultiSubscriptionRecommendationsClient_GetRecommendations_BoundsConcurrency(t *testing.T) {
+	const limit, subs = 3, 24
+	t.Setenv("CUDLY_MAX_PARALLELISM", fmt.Sprint(limit))
+
+	var inFlight, peak atomic.Int64
+	orig := newSubscriptionRecommendationsClientFn
+	t.Cleanup(func() { newSubscriptionRecommendationsClientFn = orig })
+	newSubscriptionRecommendationsClientFn = func(azcore.TokenCredential, string) (provider.RecommendationsClient, error) {
+		return &peakRecommendationsClient{inFlight: &inFlight, peak: &peak}, nil
+	}
+
+	accounts := make([]common.Account, subs)
+	for i := range accounts {
+		accounts[i] = common.Account{Provider: common.ProviderAzure, ID: fmt.Sprintf("sub-%d", i)}
+	}
+	client, err := NewMultiSubscriptionRecommendationsClient(&mockAzureTokenCredential{}, accounts)
+	require.NoError(t, err)
+
+	_, err = client.GetAllRecommendations(context.Background())
+	require.NoError(t, err)
+	assert.LessOrEqual(t, peak.Load(), int64(limit))
+	assert.Greater(t, peak.Load(), int64(1), "fan-out should still run subscriptions concurrently")
+}