fix(aws/recs): cap coverage/utilization CE calls with shared semaphore

providers/aws/recommendations/concurrency_cap_test.go

@@ -0,0 +1,101 @@
+package recommendations
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/service/costexplorer"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/semaphore"
+
+	"github.com/LeanerCloud/CUDly/pkg/concurrency"
+)
+
+// semCountingCE counts coverage/utilization SDK calls so the tests below can
+// assert whether a CE call escaped the shared concurrency semaphore.
+type semCountingCE struct {
+	mockCostExplorerAPI
+	coverageCalls    atomic.Int32
+	utilizationCalls atomic.Int32
+}
+
+func (m *semCountingCE) GetReservationCoverage(ctx context.Context, params *costexplorer.GetReservationCoverageInput, optFns ...func(*costexplorer.Options)) (*costexplorer.GetReservationCoverageOutput, error) {
+	m.coverageCalls.Add(1)
+	return &costexplorer.GetReservationCoverageOutput{}, nil
+}
+
+func (m *semCountingCE) GetReservationUtilization(ctx context.Context, params *costexplorer.GetReservationUtilizationInput, optFns ...func(*costexplorer.Options)) (*costexplorer.GetReservationUtilizationOutput, error) {
+	m.utilizationCalls.Add(1)
+	return &costexplorer.GetReservationUtilizationOutput{}, nil
+}
+
+// fullSemaphoreCtx returns a cancelled context carrying a capacity-1 shared
+// semaphore whose only slot is already held, replicating the scheduler's
+// collection path when the CUDLY_MAX_PARALLELISM cap is saturated. A correct
+// leaf call must block on Acquire (and surface ctx cancellation) instead of
+// issuing the CE request; the pre-fix code escaped the cap and called CE
+// immediately (review finding PERF-02).
+func fullSemaphoreCtx(t *testing.T) context.Context {
+	t.Helper()
+	sem := semaphore.NewWeighted(1)
+	require.NoError(t, sem.Acquire(context.Background(), 1))
+	t.Cleanup(func() { sem.Release(1) })
+
+	ctx, cancel := context.WithCancel(concurrency.WithSharedSemaphore(context.Background(), sem))
+	cancel()
+	return ctx
+}
+
+// TestFetchCoveragePage_RespectsSharedSemaphore is the PERF-02 regression
+// test for the coverage path (scheduler's AttachDailyUsageHistory fan-out):
+// with the cap saturated, fetchCoveragePage must not reach the CE SDK.
+// Pre-fix this test fails: the call escaped the semaphore, hit the mock,
+// and returned success.
+func TestFetchCoveragePage_RespectsSharedSemaphore(t *testing.T) {
+	mock := &semCountingCE{}
+	client := NewClientWithAPI(mock, "us-east-1")
+
+	_, err := client.fetchCoveragePage(fullSemaphoreCtx(t), &costexplorer.GetReservationCoverageInput{})
+
+	require.Error(t, err, "fetch must fail when the cap is saturated and ctx is cancelled")
+	assert.ErrorIs(t, err, context.Canceled)
+	assert.Equal(t, int32(0), mock.coverageCalls.Load(),
+		"GetReservationCoverage escaped the shared concurrency semaphore")
+}
+
+// TestFetchUtilizationPage_RespectsSharedSemaphore is the PERF-02 regression
+// test for the utilization path: same invariant as the coverage test.
+func TestFetchUtilizationPage_RespectsSharedSemaphore(t *testing.T) {
+	mock := &semCountingCE{}
+	client := NewClientWithAPI(mock, "us-east-1")
+
+	_, err := client.fetchUtilizationPage(fullSemaphoreCtx(t), &costexplorer.GetReservationUtilizationInput{})
+
+	require.Error(t, err, "fetch must fail when the cap is saturated and ctx is cancelled")
+	assert.ErrorIs(t, err, context.Canceled)
+	assert.Equal(t, int32(0), mock.utilizationCalls.Load(),
+		"GetReservationUtilization escaped the shared concurrency semaphore")
+}
+
+// TestFetchPages_ReleaseSemaphoreSlot verifies the Acquire/Release pairing:
+// after a successful fetch under a free capacity-1 semaphore, the slot must
+// be available again (no leak that would starve later leaf calls).
+func TestFetchPages_ReleaseSemaphoreSlot(t *testing.T) {
+	mock := &semCountingCE{}
+	client := NewClientWithAPI(mock, "us-east-1")
+
+	sem := semaphore.NewWeighted(1)
+	ctx := concurrency.WithSharedSemaphore(context.Background(), sem)
+
+	_, err := client.fetchCoveragePage(ctx, &costexplorer.GetReservationCoverageInput{})
+	require.NoError(t, err)
+	_, err = client.fetchUtilizationPage(ctx, &costexplorer.GetReservationUtilizationInput{})
+	require.NoError(t, err)
+
+	assert.Equal(t, int32(1), mock.coverageCalls.Load())
+	assert.Equal(t, int32(1), mock.utilizationCalls.Load())
+	require.True(t, sem.TryAcquire(1), "semaphore slot leaked: fetch did not release after the SDK call")
+	sem.Release(1)
+}

providers/aws/recommendations/coverage.go

@@ -11,6 +11,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/costexplorer/types"
 
 	"github.com/LeanerCloud/CUDly/pkg/common"
+	"github.com/LeanerCloud/CUDly/pkg/concurrency"
 )
 
 // coverageServiceFilters lists the Cost Explorer service-dimension values
@@ -319,14 +320,20 @@ func serviceRegionFilter(service, region string) *types.Expression {
 
 // fetchCoveragePage calls the Cost Explorer API with rate-limit retry.
 // Mirrors fetchUtilizationPage so the two paths fail and back off the
-// same way.
+// same way. The shared concurrency semaphore (CUDLY_MAX_PARALLELISM) is
+// held only across the outbound SDK call, matching fetchRIPageWithRetry,
+// so the coverage-enrichment fan-out stays inside the global IO cap.
 func (c *Client) fetchCoveragePage(ctx context.Context, input *costexplorer.GetReservationCoverageInput) (*costexplorer.GetReservationCoverageOutput, error) {
 	c.rateLimiter.Reset()
 	for {
 		if waitErr := c.rateLimiter.Wait(ctx); waitErr != nil {
 			return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr)
 		}
+		if acqErr := concurrency.Acquire(ctx); acqErr != nil {
+			return nil, fmt.Errorf("concurrency acquire failed: %w", acqErr)
+		}
 		result, err := c.costExplorerClient.GetReservationCoverage(ctx, input)
+		concurrency.Release(ctx)
 		if !c.rateLimiter.ShouldRetry(err) {
 			if err != nil {
 				return nil, fmt.Errorf("failed to get reservation coverage: %w", err)

providers/aws/recommendations/utilization.go

@@ -12,6 +12,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/costexplorer/types"
 
 	"github.com/LeanerCloud/CUDly/pkg/common"
+	"github.com/LeanerCloud/CUDly/pkg/concurrency"
 )
 
 // RIUtilization is an alias for common.RIUtilization so existing callers that
@@ -97,14 +98,21 @@ func buildUtilizations(agg map[string]*riAccumulator) []RIUtilization {
 }
 
 // fetchUtilizationPage calls the Cost Explorer API with rate-limit retry.
+// The shared concurrency semaphore (CUDLY_MAX_PARALLELISM) is held only
+// across the outbound SDK call, matching fetchRIPageWithRetry, so callers
+// running under a semaphore-carrying context stay inside the global IO cap.
 func (c *Client) fetchUtilizationPage(ctx context.Context, input *costexplorer.GetReservationUtilizationInput) (*costexplorer.GetReservationUtilizationOutput, error) {
 	c.rateLimiter.Reset()
 	for {
 		if waitErr := c.rateLimiter.Wait(ctx); waitErr != nil {
 			return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr)
 		}
 
+		if acqErr := concurrency.Acquire(ctx); acqErr != nil {
+			return nil, fmt.Errorf("concurrency acquire failed: %w", acqErr)
+		}
 		result, err := c.costExplorerClient.GetReservationUtilization(ctx, input)
+		concurrency.Release(ctx)
 		if !c.rateLimiter.ShouldRetry(err) {
 			if err != nil {
 				return nil, fmt.Errorf("failed to get reservation utilization: %w", err)