feat(purchase): refuse a recent duplicate ElastiCache reservation before submission

internal/purchase/azure_redis_routing_test.go

@@ -24,6 +24,7 @@ func purchaseCacheRecThrough(t *testing.T, prov *MockProvider, providerName, reg
 	store.On("SavePurchaseHistory", ctx, mock.Anything).Return(nil)
 	factory.On("CreateAndValidateProvider", mock.Anything, providerName, mock.Anything).Return(prov, nil)
 	prov.On("GetServiceClient", mock.Anything, want, region).Return(provider.ServiceClient(svc), nil)
+	expectNoExistingCommitments(svc)
 	svc.On("PurchaseCommitment", mock.Anything, mock.MatchedBy(func(rec common.Recommendation) bool {
 		return rec.Service == common.ServiceCache
 	}), mock.Anything).Return(common.PurchaseResult{Success: true, CommitmentID: "order-1"}, nil)

internal/purchase/coverage_extra_test.go

@@ -535,6 +535,7 @@ func TestManager_ExecuteSinglePurchase_PurchaseNotSuccessful(t *testing.T) {
 	mockStore.On("GetPurchasePlan", ctx, "plan-not-success").Return(plan, nil)
 	mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProvider, nil)
 	mockProvider.On("GetServiceClient", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), common.ServiceElastiCache, "ap-southeast-1").Return(mockServiceClient, nil)
+	mockServiceClient.On("GetExistingCommitments", mock.Anything).Return([]common.Commitment(nil), nil)
 	mockServiceClient.On("PurchaseCommitment", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), mock.AnythingOfType("common.Recommendation"), mock.AnythingOfType("common.PurchaseOptions")).Return(
 		common.PurchaseResult{Success: false}, nil,
 	)
@@ -1006,6 +1007,7 @@ func TestManager_ExecuteSinglePurchase_DetailsByService(t *testing.T) {
 			mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil)
 			mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProvider, nil)
 			mockProvider.On("GetServiceClient", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), tc.serviceType, tc.region).Return(mockServiceClient, nil)
+			expectNoExistingCommitments(mockServiceClient)
 			mockServiceClient.On(
 				"PurchaseCommitment",
 				mock.MatchedBy(hasPerRecDeadline(30*time.Second)),
@@ -1145,6 +1147,7 @@ func TestManager_ExecuteSinglePurchase_LegacyEmptyDetails(t *testing.T) {
 			mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil)
 			mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProvider, nil)
 			mockProvider.On("GetServiceClient", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), tc.serviceType, tc.region).Return(mockServiceClient, nil)
+			expectNoExistingCommitments(mockServiceClient)
 			mockServiceClient.On(
 				"PurchaseCommitment",
 				mock.MatchedBy(hasPerRecDeadline(30*time.Second)),

internal/purchase/elasticache_recent_check.go

@@ -124,3 +124,40 @@ func elastiCacheRecentRefusal(ctx context.Context, client provider.ServiceClient
 		ctx, []common.Recommendation{elastiCacheDedupeCopy(rec)}, elastiCacheRecentDedupeClient{ServiceClient: client, own: ownIDs, self: self, sawSelf: &sawSelf})
 	return elastiCacheRefusalFor(passed, filtered, err, sawSelf, rec)
 }
+
+// purchaseOwnNames carries the commitment names this execution's selected recs
+// will create, per provider, so a retry is told apart from another lineage's
+// purchase. A nil pointer reads as no own names.
+type purchaseOwnNames struct {
+	gcp, elastiCache map[string]bool
+}
+
+func newPurchaseOwnNames(exec *config.PurchaseExecution, selected []int) *purchaseOwnNames {
+	return &purchaseOwnNames{
+		gcp:         gcpOwnCommitmentNames(exec, selected),
+		elastiCache: elastiCacheOwnReservationIDs(exec, selected),
+	}
+}
+
+func (o *purchaseOwnNames) gcpNames() map[string]bool {
+	if o == nil {
+		return nil
+	}
+	return o.gcp
+}
+
+func (o *purchaseOwnNames) elastiCacheNames() map[string]bool {
+	if o == nil {
+		return nil
+	}
+	return o.elastiCache
+}
+
+// ownElastiCacheID is the reservation ID this rec's purchase call would create,
+// or "" when the call carries no idempotency token.
+func ownElastiCacheID(opts common.PurchaseOptions) string {
+	if opts.IdempotencyToken == "" {
+		return ""
+	}
+	return common.IdempotentReservationID(elastiCacheReservationIDPrefix, opts.IdempotencyToken)
+}

internal/purchase/elasticache_recent_check_wiring_test.go

@@ -0,0 +1,155 @@
+package purchase
+
+// #1863 S2: the ElastiCache recent-reservation check wired into the real
+// executor path (the real elasticache.Client over the shared fake API).
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/provider"
+	"github.com/aws/aws-sdk-go-v2/service/elasticache/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+)
+
+// expectNoExistingCommitments is the shared expectation for tests that drive an
+// AWS ElastiCache rec through executeSinglePurchase with a MockServiceClient:
+// the check lists live reservations first, and none exist. Maybe, because the
+// same table also holds recs that never list.
+func expectNoExistingCommitments(svc *MockServiceClient) {
+	svc.On("GetExistingCommitments", mock.Anything).Return([]common.Commitment(nil), nil).Maybe()
+}
+
+// runECExecution runs one AWS ElastiCache rec through the real fan-out. The
+// client is returned for the rec's own region, and the rec's region is the one
+// recorded in ecRegion, so the listing and the recommendation agree.
+func runECExecution(t *testing.T, api *fakeElastiCacheAPI, recs ...config.RecommendationRecord) (*config.PurchaseExecution, []string, bool) {
+	t.Helper()
+	ctx := context.Background()
+	store, sender, factory, prov := new(MockConfigStore), new(MockEmailSender), new(MockProviderFactory), new(MockProvider)
+	assertMocksOnCleanup(t, store, sender, factory, prov)
+	store.SavePurchaseExecutionFn = func(context.Context, *config.PurchaseExecution) error { return nil }
+	store.On("SavePurchaseHistory", ctx, mock.Anything).Return(nil).Maybe()
+	factory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(prov, nil)
+	prov.On("GetServiceClient", mock.Anything, common.ServiceElastiCache, ecRegion).Return(provider.ServiceClient(newFakeElastiCacheClient(api)), nil)
+
+	if len(recs) == 0 {
+		recs = []config.RecommendationRecord{ecCacheRec(t, "elasticache")}
+	}
+	for i := range recs {
+		recs[i].ID = fmt.Sprintf("rec-1863-%d", i)
+	}
+	exec := &config.PurchaseExecution{ExecutionID: "exec-1863", IdempotencyKey: "lineage-1863-wiring", Recommendations: recs}
+	m := &Manager{config: store, email: sender, providerFactory: factory}
+	_, _, errs, allRefused, err := m.processPurchaseRecommendations(ctx, exec, &config.PurchasePlan{Name: "p"}, "acct", (*provider.Config)(nil), "")
+	require.NoError(t, err)
+	return exec, errs, allRefused
+}
+
+func TestElastiCacheWiring_RecentSamePoolReservationRefusesBeforePurchase(t *testing.T) {
+	api := &fakeElastiCacheAPI{held: []types.ReservedCacheNode{ecHeld("other-lineage", "redis", 2, time.Hour)}}
+
+	exec, errs, allRefused := runECExecution(t, api)
+
+	assert.Empty(t, api.purchases(), "the purchase call must never be made")
+	assert.True(t, allRefused, "the row ends as a before-submission refusal")
+	require.Len(t, errs, 1)
+	assert.Contains(t, errs[0], elastiCacheRecentReservationNote)
+	assert.False(t, exec.Recommendations[0].Purchased)
+}
+
+// Regression guard (passes on main too): the library's by-ID guard adopts the
+// execution's own listed reservation, so the check must not refuse it.
+func TestElastiCacheWiring_OwnListedReservationReachesPurchaseAndIsAdopted(t *testing.T) {
+	own := common.IdempotentReservationID(elastiCacheReservationIDPrefix, common.DeriveIdempotencyToken("lineage-1863-wiring", 0))
+	api := &fakeElastiCacheAPI{held: []types.ReservedCacheNode{ecHeld(own, "redis", 2, time.Minute)}}
+
+	exec, errs, allRefused := runECExecution(t, api)
+
+	assert.Empty(t, errs)
+	assert.False(t, allRefused)
+	assert.True(t, exec.Recommendations[0].Purchased, "the own reservation is adopted, not refused")
+}
+
+func ecOwnID(i int) string {
+	return common.IdempotentReservationID(elastiCacheReservationIDPrefix, common.DeriveIdempotencyToken("lineage-1863-wiring", i))
+}
+
+// Sibling recs of one execution never refuse each other: the sibling's listed
+// reservation is in the own-ID set.
+func TestElastiCacheWiring_SiblingOwnReservationDoesNotRefuse(t *testing.T) {
+	api := &fakeElastiCacheAPI{held: []types.ReservedCacheNode{ecHeld(ecOwnID(1), "redis", 2, time.Minute)}}
+	second := ecCacheRec(t, "elasticache")
+	second.ResourceType = "cache.r6g.xlarge"
+
+	_, errs, allRefused := runECExecution(t, api, ecCacheRec(t, "elasticache"), second)
+
+	assert.Empty(t, errs)
+	assert.False(t, allRefused)
+}
+
+// A lost-response retry beside another lineage's same-pool reservation: only the
+// self ID (sawSelf) rescues it from the refusal.
+func TestElastiCacheWiring_RetryWithForeignReservationIsAdopted(t *testing.T) {
+	api := &fakeElastiCacheAPI{held: []types.ReservedCacheNode{
+		ecHeld(ecOwnID(0), "redis", 2, time.Minute), ecHeld("other-lineage", "redis", 2, time.Hour)}}
+
+	exec, errs, allRefused := runECExecution(t, api)
+
+	assert.Empty(t, errs)
+	assert.False(t, allRefused)
+	assert.True(t, exec.Recommendations[0].Purchased)
+}
+
+func TestPreSubmitChecks_NilOwnNamesDoesNotPanic(t *testing.T) {
+	svc := new(MockServiceClient)
+	expectNoExistingCommitments(svc)
+	ecRecord := ecCacheRec(t, "cache")
+	gcpRecord := config.RecommendationRecord{Provider: "gcp", Service: "compute", Region: "us-central1", ResourceType: "n2-standard-4", Count: 1}
+	gcpRec := common.Recommendation{Provider: common.ProviderGCP, Service: common.ServiceCompute, Region: "us-central1", ResourceType: "n2-standard-4", Count: 1}
+
+	assert.NotPanics(t, func() {
+		_, err := preSubmitChecks(context.Background(), svc, gcpRecord, gcpRec, common.PurchaseOptions{}, nil)
+		assert.NoError(t, err)
+		_, err = preSubmitChecks(context.Background(), svc, ecRecord, ecRec("redis", 2), common.PurchaseOptions{}, nil)
+		assert.NoError(t, err)
+	})
+}
+
+func TestPreSubmitChecks_ElastiCacheRefusesBeforeDispatchTimeIsStamped(t *testing.T) {
+	held := newFakeElastiCacheClient(&fakeElastiCacheAPI{held: []types.ReservedCacheNode{ecHeld("other", "redis", 2, time.Hour)}})
+	clean := newFakeElastiCacheClient(&fakeElastiCacheAPI{})
+	record := ecCacheRec(t, "cache")
+
+	opts, err := preSubmitChecks(context.Background(), held, record, ecRec("redis", 2), common.PurchaseOptions{}, nil)
+	requireRefusal(t, err, elastiCacheRecentReservationNote)
+	assert.True(t, opts.RequestedAt.IsZero(), "a refusal returns before withDispatchTime")
+
+	opts, err = preSubmitChecks(context.Background(), clean, record, ecRec("redis", 2), common.PurchaseOptions{}, nil)
+	require.NoError(t, err)
+	assert.False(t, opts.RequestedAt.IsZero(), "a pass continues to withDispatchTime")
+}
+
+func TestPreSubmitChecks_OnlyAWSElastiCacheListsReservations(t *testing.T) {
+	cases := map[string]common.Recommendation{
+		"aws ec2":      {Provider: common.ProviderAWS, Service: common.ServiceCompute},
+		"aws rds":      {Provider: common.ProviderAWS, Service: common.ServiceRelationalDB},
+		"aws memorydb": {Provider: common.ProviderAWS, Service: common.ServiceMemoryDB},
+		"azure cache":  {Provider: common.ProviderAzure, Service: common.ServiceCache},
+	}
+	for name, rec := range cases {
+		t.Run(name, func(t *testing.T) {
+			rec.Count = 1
+			svc := new(MockServiceClient) // no GetExistingCommitments expectation: a call panics
+			_, err := preSubmitChecks(context.Background(), svc, config.RecommendationRecord{}, rec, common.PurchaseOptions{}, nil)
+			assert.NoError(t, err)
+		})
+	}
+}

internal/purchase/execution.go

@@ -794,7 +794,7 @@ func (m *Manager) processPurchaseRecommendations(ctx context.Context, exec *conf
 	// Build the list of selected indices once so the fan-out closure only
 	// has to look up rec[i] (no second pass over the full slice).
 	selected := selectedIndices(exec.Recommendations)
-	ownGCPNames := gcpOwnCommitmentNames(exec, selected)
+	ownNames := newPurchaseOwnNames(exec, selected)
 	if len(selected) == 0 {
 		logging.Infof("purchase[%s]: no selected recommendations, nothing to execute (account=%s plan=%q)",
 			exec.ExecutionID, accountID, plan.Name)
@@ -847,7 +847,7 @@ func (m *Manager) processPurchaseRecommendations(ctx context.Context, exec *conf
 			recOpts := opts
 			recOpts.IdempotencyToken = common.DeriveIdempotencyToken(idempotencyLineageKey(exec), i)
 			recOpts.AttemptJournal = m.attemptJournal
-			purchaseResult, err := m.executeSinglePurchase(ctx, rec, provCfg, recOpts, ownGCPNames)
+			purchaseResult, err := m.executeSinglePurchase(ctx, rec, provCfg, recOpts, ownNames)
 			return recPurchaseOutcome{index: i, purchase: purchaseResult, err: err}, nil
 		}, getMaxAccountParallelism())
 
@@ -1261,7 +1261,7 @@ func logRecCtxErr(executionID, recTuple string, elapsed time.Duration, recCtxErr
 // provCfg carries optional per-account credentials; pass nil to use ambient credentials.
 // opts carries execution-level metadata (the source surface) that providers stamp
 // onto the commitment they create.
-func (m *Manager) executeSinglePurchase(ctx context.Context, rec config.RecommendationRecord, provCfg *provider.Config, opts common.PurchaseOptions, ownGCPNames map[string]bool) (common.PurchaseResult, error) {
+func (m *Manager) executeSinglePurchase(ctx context.Context, rec config.RecommendationRecord, provCfg *provider.Config, opts common.PurchaseOptions, ownNames *purchaseOwnNames) (common.PurchaseResult, error) {
 	// Per-purchase Info logs tagged with the owning execution ID so a
 	// CloudWatch filter on the execution UUID surfaces every step of the
 	// purchase attempt -- provider construction, service-client lookup,
@@ -1359,7 +1359,7 @@ func (m *Manager) executeSinglePurchase(ctx context.Context, rec config.Recommen
 	// CloudWatch shows whether a failure came from the SDK call itself
 	// (which the AWS SDK retries up to 3× × 30s by default) versus
 	// something earlier in the flow — issue #667.
-	opts, err = preSubmitChecks(recCtx, serviceClient, rec, recommendation, opts, ownGCPNames)
+	opts, err = preSubmitChecks(recCtx, serviceClient, rec, recommendation, opts, ownNames)
 	if err != nil {
 		return common.PurchaseResult{}, err
 	}
@@ -1393,13 +1393,17 @@ func (m *Manager) executeSinglePurchase(ctx context.Context, rec config.Recommen
 	return result, nil
 }
 
-// preSubmitChecks runs what must hold before PurchaseCommitment: the GCP
+// preSubmitChecks runs what must hold before PurchaseCommitment: the AWS
+// ElastiCache recent-reservation refusal (platform#1863), the GCP
 // recent-commitment refusal (platform#682, a beforeSubmission error) and go's
 // Redshift dispatch time, which bounds its DescribeReservedNodes match window
 // by RequestedAt and must be the lineage's first dispatch time, never this
 // call's.
-func preSubmitChecks(ctx context.Context, client provider.ServiceClient, rec config.RecommendationRecord, recommendation common.Recommendation, opts common.PurchaseOptions, ownGCPNames map[string]bool) (common.PurchaseOptions, error) {
-	if err := gcpRecentCommitmentRefusal(ctx, client, recommendation, ownGCPNames, ownCommitmentName(opts)); err != nil {
+func preSubmitChecks(ctx context.Context, client provider.ServiceClient, rec config.RecommendationRecord, recommendation common.Recommendation, opts common.PurchaseOptions, ownNames *purchaseOwnNames) (common.PurchaseOptions, error) {
+	if err := elastiCacheRecentRefusal(ctx, client, recommendation, ownNames.elastiCacheNames(), ownElastiCacheID(opts)); err != nil {
+		return opts, err
+	}
+	if err := gcpRecentCommitmentRefusal(ctx, client, recommendation, ownNames.gcpNames(), ownCommitmentName(opts)); err != nil {
 		return opts, err
 	}
 	return withDispatchTime(rec, opts)