refactor(aws/recommendations): extract generic fetchWithRetry for CE page fetchers

CHANGELOG.md

@@ -133,6 +133,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/).
 
 ### Changed
 
+- **`aws/recommendations`:** the four Cost Explorer coverage and utilization page fetchers share one generic `fetchWithRetry` helper (rate limiter plus concurrency cap); behaviour and error messages are unchanged (#120).
 - **`aws/rds`:** the RDS purchase path uses the shared `purchaseguard` idempotency guard and AlreadyExists recovery, like ElastiCache. Behavior for valid purchases is unchanged (#10).
 - **`aws/memorydb`:** the MemoryDB purchase path uses the shared `purchaseguard` idempotency guard and AlreadyExists recovery, like ElastiCache. Behavior for valid purchases is unchanged; only the lookup-failure error wording changes to "before MemoryDB reservation purchase" (#10).
 - **`aws/opensearch`:** the OpenSearch purchase path uses the shared `purchaseguard` idempotency guard and AlreadyExists recovery, like ElastiCache. Behavior for valid purchases is unchanged (#10).

providers/aws/recommendations/coverage.go

@@ -12,7 +12,6 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/costexplorer/types"
 
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
-	"github.com/LeanerCloud/cloud-commitments-go/pkg/concurrency"
 )
 
 // coverageServiceFilters lists the Cost Explorer service-dimension values
@@ -384,23 +383,9 @@ func serviceRegionFilter(service, region string) *types.Expression {
 // 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) {
-	rateLimiter := c.rateLimiter.newOperation()
-	for {
-		if waitErr := 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 !rateLimiter.ShouldRetry(err) {
-			if err != nil {
-				return nil, fmt.Errorf("failed to get reservation coverage: %w", err)
-			}
-			return result, nil
-		}
-	}
+	return fetchWithRetry(ctx, c.rateLimiter, "failed to get reservation coverage", func() (*costexplorer.GetReservationCoverageOutput, error) {
+		return c.costExplorerClient.GetReservationCoverage(ctx, input)
+	})
 }
 
 // poolCoverageFromGroup pulls the (pct, avg) pair from a CE response

providers/aws/recommendations/fetch_retry.go

@@ -0,0 +1,33 @@
+package recommendations
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/concurrency"
+)
+
+// fetchWithRetry runs one Cost Explorer call under the shared rate limiter
+// and the global concurrency cap, retrying while the limiter classifies the
+// error as retryable. The semaphore is held only across the SDK call, so
+// back-off waits never occupy a slot. A terminal SDK error is wrapped with
+// errPrefix; limiter and semaphore failures carry their own messages.
+func fetchWithRetry[T any](ctx context.Context, limiter *RateLimiter, errPrefix string, call func() (*T, error)) (*T, error) {
+	op := limiter.newOperation()
+	for {
+		if waitErr := op.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 := call()
+		concurrency.Release(ctx)
+		if !op.ShouldRetry(err) {
+			if err != nil {
+				return nil, fmt.Errorf("%s: %w", errPrefix, err)
+			}
+			return result, nil
+		}
+	}
+}

providers/aws/recommendations/fetch_retry_test.go

@@ -0,0 +1,66 @@
+package recommendations
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func fastLimiter(maxRetries int) *RateLimiter {
+	return NewRateLimiterWithOptions(time.Millisecond, time.Millisecond, maxRetries)
+}
+
+func TestFetchWithRetry_RetriesThrottleThenSucceeds(t *testing.T) {
+	calls := 0
+	got, err := fetchWithRetry(context.Background(), fastLimiter(3), "boom", func() (*string, error) {
+		calls++
+		if calls < 3 {
+			return nil, newThrottleError()
+		}
+		v := "ok"
+		return &v, nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, "ok", *got)
+	assert.Equal(t, 3, calls)
+}
+
+func TestFetchWithRetry_WrapsTerminalError(t *testing.T) {
+	sentinel := errors.New("access denied")
+	calls := 0
+	got, err := fetchWithRetry(context.Background(), fastLimiter(3), "failed to get x", func() (*string, error) {
+		calls++
+		return nil, sentinel
+	})
+	require.Error(t, err)
+	assert.Nil(t, got)
+	assert.ErrorIs(t, err, sentinel)
+	assert.Contains(t, err.Error(), "failed to get x")
+	assert.Equal(t, 1, calls, "non-throttle errors must not be retried")
+}
+
+func TestFetchWithRetry_StopsAfterMaxRetries(t *testing.T) {
+	calls := 0
+	_, err := fetchWithRetry(context.Background(), fastLimiter(2), "p", func() (*string, error) {
+		calls++
+		return nil, newThrottleError()
+	})
+	require.Error(t, err)
+	assert.Equal(t, 3, calls)
+}
+
+func TestFetchWithRetry_CanceledContextFailsBeforeCalling(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	calls := 0
+	_, err := fetchWithRetry(ctx, fastLimiter(1), "p", func() (*string, error) {
+		calls++
+		return nil, nil
+	})
+	require.ErrorIs(t, err, context.Canceled)
+	assert.Zero(t, calls)
+}

providers/aws/recommendations/sp_coverage.go

@@ -11,8 +11,6 @@ import (
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/costexplorer"
 	"github.com/aws/aws-sdk-go-v2/service/costexplorer/types"
-
-	"github.com/LeanerCloud/cloud-commitments-go/pkg/concurrency"
 )
 
 // SPCoverageSummary is the Savings Plans coverage summary (optionally scoped
@@ -377,23 +375,9 @@ func (c *Client) GetSPCoverageSummary(ctx context.Context, region string, lookba
 // fetchSPCoveragePage calls GetSavingsPlansCoverage with rate-limit retry.
 // Mirrors fetchCoveragePage in coverage.go so both paths back off consistently.
 func (c *Client) fetchSPCoveragePage(ctx context.Context, input *costexplorer.GetSavingsPlansCoverageInput) (*costexplorer.GetSavingsPlansCoverageOutput, error) {
-	rateLimiter := c.rateLimiter.newOperation()
-	for {
-		if waitErr := 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.GetSavingsPlansCoverage(ctx, input)
-		concurrency.Release(ctx)
-		if !rateLimiter.ShouldRetry(err) {
-			if err != nil {
-				return nil, fmt.Errorf("failed to get SP coverage: %w", err)
-			}
-			return result, nil
-		}
-	}
+	return fetchWithRetry(ctx, c.rateLimiter, "failed to get SP coverage", func() (*costexplorer.GetSavingsPlansCoverageOutput, error) {
+		return c.costExplorerClient.GetSavingsPlansCoverage(ctx, input)
+	})
 }
 
 // GetSPUtilization fetches Savings Plans utilization for one plan type over
@@ -470,23 +454,9 @@ func (c *Client) GetSPUtilization(ctx context.Context, planType types.SupportedS
 // fetchSPUtilizationPage calls GetSavingsPlansUtilization with rate-limit retry.
 // Mirrors fetchUtilizationPage in utilization.go so both paths back off consistently.
 func (c *Client) fetchSPUtilizationPage(ctx context.Context, input *costexplorer.GetSavingsPlansUtilizationInput) (*costexplorer.GetSavingsPlansUtilizationOutput, error) {
-	rateLimiter := c.rateLimiter.newOperation()
-	for {
-		if waitErr := 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.GetSavingsPlansUtilization(ctx, input)
-		concurrency.Release(ctx)
-		if !rateLimiter.ShouldRetry(err) {
-			if err != nil {
-				return nil, fmt.Errorf("failed to get SP utilization: %w", err)
-			}
-			return result, nil
-		}
-	}
+	return fetchWithRetry(ctx, c.rateLimiter, "failed to get SP utilization", func() (*costexplorer.GetSavingsPlansUtilizationOutput, error) {
+		return c.costExplorerClient.GetSavingsPlansUtilization(ctx, input)
+	})
 }
 
 // buildSPUtilizationSummary converts a CE GetSavingsPlansUtilization response

providers/aws/recommendations/utilization.go

@@ -10,7 +10,6 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/costexplorer/types"
 
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
-	"github.com/LeanerCloud/cloud-commitments-go/pkg/concurrency"
 )
 
 // RIUtilization is an alias for common.RIUtilization so existing callers that
@@ -126,24 +125,9 @@ func buildUtilizations(agg map[string]*riAccumulator) []RIUtilization {
 // 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) {
-	rateLimiter := c.rateLimiter.newOperation()
-	for {
-		if waitErr := 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 !rateLimiter.ShouldRetry(err) {
-			if err != nil {
-				return nil, fmt.Errorf("failed to get reservation utilization: %w", err)
-			}
-			return result, nil
-		}
-	}
+	return fetchWithRetry(ctx, c.rateLimiter, "failed to get reservation utilization", func() (*costexplorer.GetReservationUtilizationOutput, error) {
+		return c.costExplorerClient.GetReservationUtilization(ctx, input)
+	})
 }
 
 // aggregateUtilizationDetail accumulates hours from a single utilization detail