From 6019153eb2eebf0741a5d35198642e75f9fea7cd Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Mon, 9 Nov 2026 21:43:31 +0100 Subject: [PATCH] feat(purchase): refuse a recent duplicate ElastiCache reservation before submission preSubmitChecks now runs the ElastiCache recent-reservation check for AWS cache recommendations, before the GCP check and before the dispatch time is stamped. The per-execution own-name sets are one pointer struct with nil-safe accessors, built once and passed through the fan-out, so the platform's own reservations (its siblings in the same execution and its lost-response retries) are adopted and never refused. The client and the recommendation both take their region from rec.Region. Tests that run AWS cache recommendations through the real executor now expect the reservation listing. Part of #1863 Co-Authored-By: Claude Sonnet 5.5 --- internal/purchase/azure_redis_routing_test.go | 1 + internal/purchase/coverage_extra_test.go | 3 + internal/purchase/elasticache_recent_check.go | 37 +++++ .../elasticache_recent_check_wiring_test.go | 155 ++++++++++++++++++ internal/purchase/execution.go | 18 +- 5 files changed, 207 insertions(+), 7 deletions(-) create mode 100644 internal/purchase/elasticache_recent_check_wiring_test.go diff --git a/internal/purchase/azure_redis_routing_test.go b/internal/purchase/azure_redis_routing_test.go index 9f2b6c867..bbde0f52b 100644 --- a/internal/purchase/azure_redis_routing_test.go +++ b/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) diff --git a/internal/purchase/coverage_extra_test.go b/internal/purchase/coverage_extra_test.go index 05ea829b9..a7d9f0caa 100644 --- a/internal/purchase/coverage_extra_test.go +++ b/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)), diff --git a/internal/purchase/elasticache_recent_check.go b/internal/purchase/elasticache_recent_check.go index a97bf9a1e..fb8c4b883 100644 --- a/internal/purchase/elasticache_recent_check.go +++ b/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) +} diff --git a/internal/purchase/elasticache_recent_check_wiring_test.go b/internal/purchase/elasticache_recent_check_wiring_test.go new file mode 100644 index 000000000..defdeaa2f --- /dev/null +++ b/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) + }) + } +} diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 5106db205..e2b3a333f 100644 --- a/internal/purchase/execution.go +++ b/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)