fix(aws): paginate CE recommendations + RDS engine versions (closes #692)

cmd/multi_service_engine_versions.go

@@ -14,6 +14,7 @@ import (
 	awsec2 "github.com/aws/aws-sdk-go-v2/service/ec2"
 	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
 	awsrds "github.com/aws/aws-sdk-go-v2/service/rds"
+	rdstypes "github.com/aws/aws-sdk-go-v2/service/rds/types"
 )
 
 // InstanceEngineVersion stores engine version information for an instance
@@ -87,6 +88,16 @@ func getAWSRegions(ctx context.Context, awsCfg aws.Config) ([]ec2types.Region, e
 // maxConcurrentRegionQueries limits the number of concurrent AWS API calls across regions
 const maxConcurrentRegionQueries = 10
 
+// maxEngineVersionPages caps DescribeDBMajorEngineVersions pagination per engine.
+// 20 pages x ~100 records/page = ~2000 records, enough for any engine list (issue #692).
+const maxEngineVersionPages = 20
+
+// RDSMajorVersionsClient is the subset of the RDS API needed by
+// queryMajorEngineVersionsWithClient, extracted so tests can inject a mock.
+type RDSMajorVersionsClient interface {
+	DescribeDBMajorEngineVersions(ctx context.Context, params *awsrds.DescribeDBMajorEngineVersionsInput, optFns ...func(*awsrds.Options)) (*awsrds.DescribeDBMajorEngineVersionsOutput, error)
+}
+
 // queryRDSInstancesInRegions queries RDS instances in all regions concurrently
 func queryRDSInstancesInRegions(ctx context.Context, awsCfg aws.Config, regions []ec2types.Region) (map[string][]InstanceEngineVersion, error) {
 	instanceVersions := make(map[string][]InstanceEngineVersion)
@@ -186,51 +197,93 @@ func queryMajorEngineVersions(ctx context.Context, cfg Config) (map[string]Major
 		return nil, fmt.Errorf("failed to load AWS config: %w", err)
 	}
 
-	rdsClient := awsrds.NewFromConfig(awsCfg)
+	return queryMajorEngineVersionsWithClient(ctx, awsrds.NewFromConfig(awsCfg))
+}
 
+// queryMajorEngineVersionsWithClient is the testable core of queryMajorEngineVersions.
+// It accepts a RDSMajorVersionsClient so tests can inject a mock without real AWS creds.
+func queryMajorEngineVersionsWithClient(ctx context.Context, rdsClient RDSMajorVersionsClient) (map[string]MajorEngineVersionInfo, error) {
 	// Map of "engine:majorVersion" -> MajorEngineVersionInfo
 	versionInfo := make(map[string]MajorEngineVersionInfo)
 
 	// Query all engine types we care about
 	engines := []string{"mysql", "postgres", "aurora-mysql", "aurora-postgresql"}
 
 	for _, engine := range engines {
+		if err := fetchMajorEngineVersionsForEngine(ctx, rdsClient, engine, versionInfo); err != nil {
+			log.Printf("Warning: Failed to describe major engine versions for %s: %v", engine, err)
+		}
+	}
+
+	return versionInfo, nil
+}
+
+// fetchMajorEngineVersionsForEngine fetches all pages of major engine version
+// info for a single engine and merges results into versionInfo. Returns an error
+// only on API failure or pagination cap exceeded (issue #692).
+func fetchMajorEngineVersionsForEngine(ctx context.Context, rdsClient RDSMajorVersionsClient, engine string, versionInfo map[string]MajorEngineVersionInfo) error {
+	var marker *string
+
+	for pageIdx := 0; ; pageIdx++ {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		if pageIdx >= maxEngineVersionPages {
+			return fmt.Errorf(
+				"pagination cap reached after %d pages for engine %s (issue #692)",
+				maxEngineVersionPages, engine,
+			)
+		}
+
 		output, err := rdsClient.DescribeDBMajorEngineVersions(ctx, &awsrds.DescribeDBMajorEngineVersionsInput{
 			Engine: aws.String(engine),
+			Marker: marker,
 		})
 		if err != nil {
-			log.Printf("⚠️  Warning: Failed to describe major engine versions for %s: %v", engine, err)
-			continue
+			return err
 		}
 
 		for _, version := range output.DBMajorEngineVersions {
-			info := MajorEngineVersionInfo{
-				Engine:             aws.ToString(version.Engine),
-				MajorEngineVersion: aws.ToString(version.MajorEngineVersion),
-			}
+			info := parseDBMajorEngineVersion(version)
+			key := fmt.Sprintf("%s:%s", info.Engine, info.MajorEngineVersion)
+			versionInfo[key] = info
+		}
 
-			// Parse lifecycle support dates
-			for _, lifecycle := range version.SupportedEngineLifecycles {
-				lifecycleInfo := EngineLifecycleInfo{
-					LifecycleSupportName: string(lifecycle.LifecycleSupportName),
-				}
+		if output.Marker == nil || aws.ToString(output.Marker) == "" {
+			break
+		}
+		marker = output.Marker
+	}
 
-				if lifecycle.LifecycleSupportStartDate != nil {
-					lifecycleInfo.LifecycleSupportStartDate = *lifecycle.LifecycleSupportStartDate
-				}
-				if lifecycle.LifecycleSupportEndDate != nil {
-					lifecycleInfo.LifecycleSupportEndDate = *lifecycle.LifecycleSupportEndDate
-				}
+	return nil
+}
 
-				info.SupportedEngineLifecycles = append(info.SupportedEngineLifecycles, lifecycleInfo)
-			}
+// parseDBMajorEngineVersion converts an RDS DBMajorEngineVersion into a
+// MajorEngineVersionInfo, extracting lifecycle support dates. Extracted from
+// fetchMajorEngineVersionsForEngine to keep its cyclomatic complexity below
+// the gocyclo cap.
+func parseDBMajorEngineVersion(version rdstypes.DBMajorEngineVersion) MajorEngineVersionInfo {
+	info := MajorEngineVersionInfo{
+		Engine:             aws.ToString(version.Engine),
+		MajorEngineVersion: aws.ToString(version.MajorEngineVersion),
+	}
 
-			key := fmt.Sprintf("%s:%s", info.Engine, info.MajorEngineVersion)
-			versionInfo[key] = info
+	for _, lifecycle := range version.SupportedEngineLifecycles {
+		lifecycleInfo := EngineLifecycleInfo{
+			LifecycleSupportName: string(lifecycle.LifecycleSupportName),
+		}
+
+		if lifecycle.LifecycleSupportStartDate != nil {
+			lifecycleInfo.LifecycleSupportStartDate = *lifecycle.LifecycleSupportStartDate
+		}
+		if lifecycle.LifecycleSupportEndDate != nil {
+			lifecycleInfo.LifecycleSupportEndDate = *lifecycle.LifecycleSupportEndDate
 		}
+
+		info.SupportedEngineLifecycles = append(info.SupportedEngineLifecycles, lifecycleInfo)
 	}
 
-	return versionInfo, nil
+	return info
 }
 
 // extractMajorVersion extracts the major version from a full engine version string

cmd/multi_service_engine_versions_paginate_test.go

@@ -0,0 +1,153 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awsrds "github.com/aws/aws-sdk-go-v2/service/rds"
+	rdstypes "github.com/aws/aws-sdk-go-v2/service/rds/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// multiPageRDSMajorVersionsMock implements RDSMajorVersionsClient and returns
+// distinct pages based on the Marker in the incoming request.
+type multiPageRDSMajorVersionsMock struct {
+	pages  []*awsrds.DescribeDBMajorEngineVersionsOutput
+	tokens []string // tokens[i] triggers pages[i+1]; first call has empty marker
+	calls  int
+}
+
+func (m *multiPageRDSMajorVersionsMock) DescribeDBMajorEngineVersions(
+	_ context.Context,
+	params *awsrds.DescribeDBMajorEngineVersionsInput,
+	_ ...func(*awsrds.Options),
+) (*awsrds.DescribeDBMajorEngineVersionsOutput, error) {
+	idx := 0
+	incoming := aws.ToString(params.Marker)
+	for i, tok := range m.tokens {
+		if tok == incoming {
+			idx = i + 1
+			break
+		}
+	}
+	if incoming == "" {
+		idx = 0
+	}
+	m.calls++
+	if idx >= len(m.pages) {
+		return nil, fmt.Errorf("unexpected RDS Marker %q", incoming)
+	}
+	return m.pages[idx], nil
+}
+
+// rdsMajorVersion builds a minimal DBMajorEngineVersion for tests.
+func rdsMajorVersion(engine, major string) rdstypes.DBMajorEngineVersion {
+	return rdstypes.DBMajorEngineVersion{
+		Engine:             aws.String(engine),
+		MajorEngineVersion: aws.String(major),
+		SupportedEngineLifecycles: []rdstypes.SupportedEngineLifecycle{
+			{
+				LifecycleSupportName:      "open-source-rds-extended-support",
+				LifecycleSupportStartDate: aws.Time(time.Now().AddDate(-1, 0, 0)),
+				LifecycleSupportEndDate:   aws.Time(time.Now().AddDate(2, 0, 0)),
+			},
+		},
+	}
+}
+
+// TestFetchMajorEngineVersionsForEngine_Paginates asserts that all pages are
+// fetched and results accumulated (issue #692 regression test).
+func TestFetchMajorEngineVersionsForEngine_Paginates(t *testing.T) {
+	mock := &multiPageRDSMajorVersionsMock{
+		pages: []*awsrds.DescribeDBMajorEngineVersionsOutput{
+			{
+				DBMajorEngineVersions: []rdstypes.DBMajorEngineVersion{
+					rdsMajorVersion("mysql", "5.7"),
+					rdsMajorVersion("mysql", "8.0"),
+				},
+				Marker: aws.String("tok1"),
+			},
+			{
+				DBMajorEngineVersions: []rdstypes.DBMajorEngineVersion{
+					rdsMajorVersion("mysql", "8.4"),
+					rdsMajorVersion("mysql", "9.0"),
+				},
+				Marker: aws.String("tok2"),
+			},
+			{
+				DBMajorEngineVersions: []rdstypes.DBMajorEngineVersion{
+					rdsMajorVersion("mysql", "9.1"),
+				},
+				Marker: nil,
+			},
+		},
+		tokens: []string{"tok1", "tok2"},
+	}
+
+	versionInfo := make(map[string]MajorEngineVersionInfo)
+	err := fetchMajorEngineVersionsForEngine(context.Background(), mock, "mysql", versionInfo)
+	require.NoError(t, err)
+	// 2 + 2 + 1 = 5 versions across 3 pages
+	assert.Len(t, versionInfo, 5, "must accumulate all versions across pages")
+	assert.Equal(t, 3, mock.calls, "must call API once per page")
+	assert.Contains(t, versionInfo, "mysql:5.7")
+	assert.Contains(t, versionInfo, "mysql:9.1")
+}
+
+// TestFetchMajorEngineVersionsForEngine_EmptyMarkerTerminates asserts that an
+// empty-string Marker is treated as terminal (parity with PR #690).
+func TestFetchMajorEngineVersionsForEngine_EmptyMarkerTerminates(t *testing.T) {
+	mock := &multiPageRDSMajorVersionsMock{
+		pages: []*awsrds.DescribeDBMajorEngineVersionsOutput{
+			{
+				DBMajorEngineVersions: []rdstypes.DBMajorEngineVersion{
+					rdsMajorVersion("mysql", "8.0"),
+				},
+				Marker: aws.String(""), // empty string -- must terminate
+			},
+		},
+		tokens: []string{},
+	}
+
+	versionInfo := make(map[string]MajorEngineVersionInfo)
+	err := fetchMajorEngineVersionsForEngine(context.Background(), mock, "mysql", versionInfo)
+	require.NoError(t, err)
+	assert.Len(t, versionInfo, 1)
+	assert.Equal(t, 1, mock.calls, "empty-string Marker must terminate after page 1")
+}
+
+// alwaysNextPageRDSMock returns pages each carrying a non-nil non-empty Marker.
+type alwaysNextPageRDSMock struct {
+	calls int
+}
+
+func (m *alwaysNextPageRDSMock) DescribeDBMajorEngineVersions(
+	_ context.Context,
+	_ *awsrds.DescribeDBMajorEngineVersionsInput,
+	_ ...func(*awsrds.Options),
+) (*awsrds.DescribeDBMajorEngineVersionsOutput, error) {
+	m.calls++
+	return &awsrds.DescribeDBMajorEngineVersionsOutput{
+		DBMajorEngineVersions: []rdstypes.DBMajorEngineVersion{
+			rdsMajorVersion("mysql", fmt.Sprintf("5.%d", m.calls)),
+		},
+		Marker: aws.String(fmt.Sprintf("tok%d", m.calls)),
+	}, nil
+}
+
+// TestFetchMajorEngineVersionsForEngine_PaginationCapError asserts that
+// exceeding maxEngineVersionPages returns a diagnostic error (issue #692).
+func TestFetchMajorEngineVersionsForEngine_PaginationCapError(t *testing.T) {
+	mock := &alwaysNextPageRDSMock{}
+	versionInfo := make(map[string]MajorEngineVersionInfo)
+
+	err := fetchMajorEngineVersionsForEngine(context.Background(), mock, "mysql", versionInfo)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "pagination cap reached")
+	assert.Equal(t, maxEngineVersionPages, mock.calls,
+		"must stop exactly at the cap")
+}

providers/aws/recommendations/client.go

@@ -15,6 +15,12 @@ import (
 	"github.com/LeanerCloud/CUDly/pkg/logging"
 )
 
+// maxRecommendationPages caps the number of pages fetched per Cost Explorer
+// GetReservationPurchaseRecommendation or GetSavingsPlansPurchaseRecommendation
+// call. 20 pages x ~100 items/page = ~2000 items, enough headroom for any
+// payer org we have seen. Exceeding the cap returns a diagnostic error (issue #692).
+const maxRecommendationPages = 20
+
 // CostExplorerAPI defines the interface for Cost Explorer operations
 type CostExplorerAPI interface {
 	GetReservationPurchaseRecommendation(ctx context.Context, params *costexplorer.GetReservationPurchaseRecommendationInput, optFns ...func(*costexplorer.Options)) (*costexplorer.GetReservationPurchaseRecommendationOutput, error)
@@ -72,16 +78,65 @@ func (c *Client) GetRecommendations(ctx context.Context, params common.Recommend
 		AccountScope:         types.AccountScopeLinked,
 	}
 
-	// Implement rate limiting with exponential backoff. The shared semaphore
-	// (if any) on ctx bounds aggregate concurrent Cost Explorer requests; we
-	// Acquire/Release around the SDK call itself rather than around the whole
-	// service sweep so a goroutine waiting on rate-limiter backoff or
-	// processing a response doesn't monopolise a permit while no request is
-	// in flight. See pkg/concurrency.
+	allRecs, err := c.fetchRIAllPages(ctx, input, params.Service)
+	if err != nil {
+		return nil, err
+	}
+
+	return c.parseRecommendations(allRecs, params)
+}
+
+// fetchRIAllPages paginates over all pages of RI recommendations for a single
+// (service, term, payment) combination. ctx.Err() is checked at the top of
+// each iteration so cancellation is terminal (per feedback_ctx_cancel_terminal.md,
+// issue #692).
+func (c *Client) fetchRIAllPages(
+	ctx context.Context,
+	input *costexplorer.GetReservationPurchaseRecommendationInput,
+	service common.ServiceType,
+) ([]types.ReservationPurchaseRecommendation, error) {
+	var allRecs []types.ReservationPurchaseRecommendation
+	var nextPageToken *string
+
+	for pageIdx := 0; ; pageIdx++ {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		if pageIdx >= maxRecommendationPages {
+			return nil, fmt.Errorf(
+				"pagination cap reached after %d pages for RI %s (issue #692)",
+				maxRecommendationPages, service,
+			)
+		}
+		input.NextPageToken = nextPageToken
+
+		result, err := c.fetchRIPageWithRetry(ctx, input)
+		if err != nil {
+			return nil, err
+		}
+
+		allRecs = append(allRecs, result.Recommendations...)
+
+		if result.NextPageToken == nil || aws.ToString(result.NextPageToken) == "" {
+			break
+		}
+		nextPageToken = result.NextPageToken
+	}
+
+	return allRecs, nil
+}
+
+// fetchRIPageWithRetry executes a single GetReservationPurchaseRecommendation
+// call with rate-limiter exponential back-off. Extracted so the pagination loop
+// in fetchRIAllPages stays below the gocyclo cap.
+func (c *Client) fetchRIPageWithRetry(
+	ctx context.Context,
+	input *costexplorer.GetReservationPurchaseRecommendationInput,
+) (*costexplorer.GetReservationPurchaseRecommendationOutput, error) {
+	c.rateLimiter.Reset()
 	var result *costexplorer.GetReservationPurchaseRecommendationOutput
 	var err error
 
-	c.rateLimiter.Reset()
 	for {
 		if waitErr := c.rateLimiter.Wait(ctx); waitErr != nil {
 			return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr)
@@ -101,7 +156,7 @@ func (c *Client) GetRecommendations(ctx context.Context, params common.Recommend
 		return nil, fmt.Errorf("failed to get RI recommendations after %d retries: %w", c.rateLimiter.GetRetryCount(), err)
 	}
 
-	return c.parseRecommendations(result.Recommendations, params)
+	return result, nil
 }
 
 // defaultDiscoveryTerms enumerates the term lengths the discovery flow