diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 6f3742ddb..4f19e8583 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -455,6 +455,10 @@ func (m *mockConfigStore) ClaimMarketplaceListingSlot(_ context.Context, _ strin return true, nil } +func (m *mockConfigStore) ClaimRIExchangeIdempotencyKey(_ context.Context, _ string, _ time.Duration) (bool, error) { + return true, nil +} + func (m *mockConfigStore) StampOfferingClass(_ context.Context, _, _ string) error { return nil } diff --git a/internal/api/handler_ri_exchange.go b/internal/api/handler_ri_exchange.go index 2065c539a..5bc838799 100644 --- a/internal/api/handler_ri_exchange.go +++ b/internal/api/handler_ri_exchange.go @@ -572,7 +572,10 @@ func validateAzureExchangeTargets(targets []AzureExchangeTargetBody, subscriptio } scope := azureBillingScopeID(subscriptionID) for i, t := range targets { - if t.SKU == "" { + // Trimmed, like location below: the #1642 idempotency fingerprint + // normalizes surrounding whitespace away, so a blank-but-not-empty + // SKU would reach it as "" and collapse onto other blank spellings. + if strings.TrimSpace(t.SKU) == "" { return NewClientError(400, fmt.Sprintf("targets[%d].sku is required", i)) } // Blank-but-not-empty is rejected too: targetLocations trims before @@ -1203,10 +1206,28 @@ func (h *Handler) executeAzureExchange(ctx context.Context, req *events.LambdaFu return nil, err } + // Submit-time idempotency (#1642). The re-quote above mints a FRESH + // session on every request, so Azure's own session-level replay + // protection cannot see two POSTs of one logical exchange as duplicates: + // without this claim a client that times out mid-LRO and retries commits + // the exchange twice, each half individually under the cap. Taken here, + // last, so no gate rejection ever leaves a claim behind. + err = h.claimExchangeSubmit(ctx, azureExchangeIdempotencyKey(body)) + if err != nil { + return nil, err + } + result, err := client.ExecuteExchange(ctx, preview.SessionID) if err != nil { logging.Errorf("azure exchange execution failed: %v", err) - return nil, mapAzureExchangeError("exchange execution failed", err) + // Non-4xx failures here are ambiguous: BeginPost may already have + // submitted the exchange, and a ctx cancellation mid-poll looks + // identical to one that never reached Azure. Say so rather than + // reporting a flat failure that invites the retry this claim now + // refuses (#1642). + return nil, mapAzureExchangeError( + "exchange execution failed and may already have been submitted to Azure; "+ + "verify the reservation state in the Azure portal before retrying", err) } logging.Infof("azure ri-exchange executed: subscription=%s session=%s status=%s", body.SubscriptionID, result.SessionID, result.Status) @@ -1616,8 +1637,15 @@ func firstNonEmptyCurrency(instances []ec2svc.ConvertibleRI) string { } // validateTargets checks each entry in targets for a non-empty, UUID-shaped -// offering_id. Extracted so both getExchangeQuote and validateExecuteExchangeBody -// share the same check without exceeding the gocyclo threshold. +// offering_id and a positive count. Extracted so both getExchangeQuote and +// validateExecuteExchangeBody share the same check without exceeding the +// gocyclo threshold. +// +// The count check mirrors pkg/exchange.validateTargets, which applies the same +// rule but only once ExecuteExchange is already running. Repeating it here +// moves the refusal ahead of the #1642 submit claim -- a request that can never +// commit must not leave a claim behind -- and turns what pkg/exchange would +// surface as an opaque 500 into a 400 naming the offending field. func validateTargets(targets []ExchangeTargetBody) error { for i, t := range targets { if t.OfferingID == "" { @@ -1630,6 +1658,25 @@ func validateTargets(targets []ExchangeTargetBody) error { "did you paste an instance type by mistake?", i, t.OfferingID)) } + if t.Count < 1 { + return NewClientError(400, fmt.Sprintf("targets[%d].count must be >= 1, got %d", i, t.Count)) + } + } + return nil +} + +// validateExchangeRIIDs checks the source list and each id in it. Each id +// individually, not just the list length: a blank entry reaches the #1642 +// submit fingerprint as an empty component, so two different blank spellings +// of one request would claim the same key. +func validateExchangeRIIDs(riIDs []string) error { + if len(riIDs) == 0 { + return NewClientError(400, "ri_ids is required") + } + for i, id := range riIDs { + if strings.TrimSpace(id) == "" { + return NewClientError(400, fmt.Sprintf("ri_ids[%d] is empty", i)) + } } return nil } @@ -1684,8 +1731,8 @@ func (h *Handler) getExchangeQuote(ctx context.Context, req *events.LambdaFuncti // cyclomatic-complexity threshold; every branch here becomes a // separate test case so the logic stays inspectable. func validateExecuteExchangeBody(body ExchangeExecuteRequestBody) error { - if len(body.RIIDs) == 0 { - return NewClientError(400, "ri_ids is required") + if err := validateExchangeRIIDs(body.RIIDs); err != nil { + return err } if len(body.Targets) == 0 && body.TargetOfferingID == "" { return NewClientError(400, "either targets[] or target_offering_id is required") @@ -1693,6 +1740,12 @@ func validateExecuteExchangeBody(body ExchangeExecuteRequestBody) error { if err := validateTargets(body.Targets); err != nil { return err } + // The legacy singleton's count, which validateTargets above does not see. + // Same reasoning as the targets[] count: refuse before the submit claim + // rather than inside ExecuteExchange, after it. + if len(body.Targets) == 0 && body.TargetCount < 1 { + return NewClientError(400, fmt.Sprintf("target_count must be >= 1, got %d", body.TargetCount)) + } if body.MaxPaymentDueUSD == "" { return NewClientError(400, "max_payment_due_usd is required as a safety guardrail") } @@ -1761,6 +1814,16 @@ func (h *Handler) executeExchange(ctx context.Context, req *events.LambdaFunctio return nil, err } + // Submit-time idempotency (#1642). AcceptReservedInstancesExchangeQuote + // carries no ClientToken, so AWS will happily accept the same exchange + // twice; a client that times out and retries otherwise double-spends with + // each half individually under the cap. Taken last, after every gate, so + // no rejected request leaves a claim behind. + err = h.claimExchangeSubmit(ctx, awsExchangeIdempotencyKey(cloudAccountID, body)) + if err != nil { + return nil, err + } + exchangeID, quote, err := exchange.ExecuteExchange(ctx, exchange.ExchangeExecuteRequest{ Region: region, ReservedIDs: body.RIIDs, diff --git a/internal/api/handler_ri_exchange_test.go b/internal/api/handler_ri_exchange_test.go index 3720a879d..1bef1549c 100644 --- a/internal/api/handler_ri_exchange_test.go +++ b/internal/api/handler_ri_exchange_test.go @@ -1522,7 +1522,7 @@ func TestExecuteExchange_EmptyRegionReturns400(t *testing.T) { _, err := h.executeExchange(context.Background(), &events.LambdaFunctionURLRequest{ Headers: map[string]string{"authorization": "Bearer test-token"}, - Body: `{"ri_ids":["ri-123"],"target_offering_id":"off-1","max_payment_due_usd":"10.00"}`, + Body: `{"ri_ids":["ri-123"],"target_offering_id":"off-1","target_count":1,"max_payment_due_usd":"10.00"}`, }) require.Error(t, err) ce, ok := IsClientError(err) @@ -1571,7 +1571,7 @@ func TestExecuteExchange_PermissionConstraintsDenied(t *testing.T) { } _, err := h.executeExchange(ctx, &events.LambdaFunctionURLRequest{ Headers: map[string]string{"authorization": "Bearer exchange-token"}, - Body: `{"ri_ids":["ri-123"],"target_offering_id":"off-1","max_payment_due_usd":"250.50","region":"eu-central-1"}`, + Body: `{"ri_ids":["ri-123"],"target_offering_id":"off-1","target_count":1,"max_payment_due_usd":"250.50","region":"eu-central-1"}`, }) require.Error(t, err) ce, ok := IsClientError(err) @@ -1606,7 +1606,7 @@ func TestExecuteExchange_AccountResolutionErrorFailsClosed(t *testing.T) { } _, err := h.executeExchange(ctx, &events.LambdaFunctionURLRequest{ Headers: map[string]string{"authorization": "Bearer exchange-token"}, - Body: `{"ri_ids":["ri-123"],"target_offering_id":"off-1","max_payment_due_usd":"250.50","region":"eu-central-1"}`, + Body: `{"ri_ids":["ri-123"],"target_offering_id":"off-1","target_count":1,"max_payment_due_usd":"250.50","region":"eu-central-1"}`, }) require.Error(t, err) assert.Contains(t, err.Error(), "resolve cloud account scope") @@ -1643,7 +1643,7 @@ func TestExecuteExchange_UnattributedAccountStillConstrained(t *testing.T) { } _, err := h.executeExchange(ctx, &events.LambdaFunctionURLRequest{ Headers: map[string]string{"authorization": "Bearer exchange-token"}, - Body: `{"ri_ids":["ri-123"],"target_offering_id":"off-1","max_payment_due_usd":"250.50","region":"eu-central-1"}`, + Body: `{"ri_ids":["ri-123"],"target_offering_id":"off-1","target_count":1,"max_payment_due_usd":"250.50","region":"eu-central-1"}`, }) require.Error(t, err) ce, ok := IsClientError(err) diff --git a/internal/api/openapi.yaml b/internal/api/openapi.yaml index 36809a380..ab1b5aa40 100644 --- a/internal/api/openapi.yaml +++ b/internal/api/openapi.yaml @@ -775,6 +775,13 @@ paths: the fresh quote carries policy errors, omits a net payable amount, is denominated in a different currency than requested, or exceeds `max_payment_due`. + Submits are deduplicated for 15 minutes on a fingerprint of + `subscription_id` plus the sources and targets with their quantities, + so a retry after a client timeout returns 409 rather than committing + the exchange a second time. The spend cap and currency are not part of + that fingerprint: raising the cap does not make it a different + purchase. See the 409 response: it does not assert that the earlier + submit succeeded, only that it claimed the window. parameters: - $ref: '#/components/parameters/CSRFToken' requestBody: @@ -820,6 +827,8 @@ paths: $ref: '#/components/responses/Forbidden' '404': $ref: '#/components/responses/NotFound' + '409': + $ref: '#/components/responses/Conflict' '422': $ref: '#/components/responses/UnprocessableEntity' @@ -917,6 +926,13 @@ paths: exchanges submitted to AWS cannot be rolled back. Non-admin users must be explicitly granted `execute:ri-exchange` via a custom group; there is no default user-role grant. + Submits are deduplicated for 15 minutes on a fingerprint of the + deployment's cloud account and region plus the source RIs and targets + with their counts, so a retry after a client timeout returns 409 rather + than committing the exchange a second time. `max_payment_due_usd` is + not part of that fingerprint: raising the cap does not make it a + different purchase. See the 409 response: it does not assert that the + earlier submit succeeded, only that it claimed the window. parameters: - $ref: '#/components/parameters/CSRFToken' requestBody: @@ -951,6 +967,8 @@ paths: $ref: '#/components/responses/Unauthorized' '403': $ref: '#/components/responses/Forbidden' + '409': + $ref: '#/components/responses/Conflict' /api/ri-exchange/config: get: @@ -1881,6 +1899,21 @@ components: application/json: schema: $ref: '#/components/schemas/Error' + Conflict: + description: > + An identical submit already holds the idempotency claim, so THIS + request was not executed. The claim is retained unconditionally once + taken, including when the earlier submit failed, so this response + asserts nothing about that submit's outcome: it may still be running, + it may have committed, its outcome may be unresolved (the provider + call can fail after the operation was already submitted), or it may + have failed without committing anything. Do not read a 409 as + confirmation that the earlier submit succeeded. Verify its outcome + with the provider before resubmitting. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' RateLimited: description: Rate limit exceeded content: diff --git a/internal/api/ri_exchange_idempotency.go b/internal/api/ri_exchange_idempotency.go new file mode 100644 index 000000000..307ef3627 --- /dev/null +++ b/internal/api/ri_exchange_idempotency.go @@ -0,0 +1,197 @@ +package api + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "sort" + "strings" + "time" +) + +// riExchangeIdempotencyWindow is how long a committed RI exchange submit +// "absorbs" an identical resubmission (issue #1642). +// +// It is deliberately much longer than the purchase path's two-minute +// purchaseIdempotencyWindow. An exchange's provider call is a long-running +// operation polled to completion INSIDE the request: Azure's ExecuteExchange +// blocks on PollUntilDone, and the AWS path quotes, re-quotes and accepts in +// sequence. The double-spend this guards against is a client that gives up on +// that still-running request and retries, so the window has to outlast a full +// request lifetime plus the client's own timeout and the operator's retry, not +// merely a double-click. +// +// The cost of the window being generous is bounded and visible: a caller who +// genuinely wants to repeat the very same exchange sooner gets an explanatory +// 409 telling them to wait. The cost of it being too short is an irreversible +// second commitment, so it is sized towards the former. +const riExchangeIdempotencyWindow = 15 * time.Minute + +// Provider scope literals for the idempotency fingerprint. They keep an AWS +// exchange and an Azure exchange from ever colliding on one key, whatever the +// rest of the fingerprint happens to hash to. +const ( + azureExchangeIdempotencyScope = "azure-ri-exchange" + awsExchangeIdempotencyScope = "aws-ri-exchange" +) + +// writeLengthPrefixed feeds variable-length components into a hash as +// ":", so no value can be crafted to shift a field +// boundary and make two genuinely different requests fingerprint alike. +// (A plain separator-joined encoding would let a reservation ID carrying the +// separator do exactly that, silently suppressing a distinct exchange.) +func writeLengthPrefixed(w io.Writer, parts ...string) { + for _, p := range parts { + fmt.Fprintf(w, "%d:%s", len(p), p) + } +} + +// exchangeIdempotencyKey hashes a canonical description of what an RI exchange +// submit buys: the provider scope, the account scope the money lands in, and +// the exchange's sources and targets. Both lists are sorted, so the ordering a +// client happens to send cannot change the fingerprint, and the element counts +// are hashed too, so a source can never be read as a target. +func exchangeIdempotencyKey(scope, accountScope string, sources, targets []string) string { + src := append([]string(nil), sources...) + sort.Strings(src) + tgt := append([]string(nil), targets...) + sort.Strings(tgt) + + h := sha256.New() + writeLengthPrefixed(h, scope, accountScope) + fmt.Fprintf(h, "|s%d|", len(src)) + writeLengthPrefixed(h, src...) + fmt.Fprintf(h, "|t%d|", len(tgt)) + writeLengthPrefixed(h, tgt...) + return hex.EncodeToString(h.Sum(nil)) +} + +// azureExchangeIdempotencyKey fingerprints an Azure execute request. +// +// INCLUDED, and why omitting any of them would let one token stand for two +// different purchases (silently dropping the second): +// +// - subscription_id, per the #1495 precedent. ListExchangeableReservations +// is TENANT-wide, so a scope-blind token aliases across subscriptions. +// - every source's reservation id AND its quantity. +// - every target's sku, location, term AND quantity. +// +// EXCLUDED, and why including any of them would let two tokens stand for one +// purchase -- which is the double-spend this whole guard exists to stop: +// +// - max_payment_due: a spend CAP, not what is bought. A retry that merely +// raises the cap must not mint a fresh key and commit a second exchange. +// Two submits differing only in cap buy exactly the same thing. +// - currency: likewise a guardrail. It is checked against the fresh quote +// before the claim is ever made and does not change what is bought. +// - targets[].billing_scope_id: validated to be either absent or the +// subscription's own derived scope, so it never changes what is charged. +// Including it would fingerprint the omitted and the spelled-out form of +// the SAME purchase differently. +// - the submitting user: the money moves identically whoever sends it, and +// two users exchanging the same reservation ids IS the double spend. The +// claim is taken after every authorization gate, so a cross-user 409 +// discloses nothing the caller was not already entitled to see. +// +// Case and surrounding whitespace are normalized away on the ARM identifiers, +// which are case-insensitive: two spellings of one subscription, reservation +// or SKU are one purchase and must fingerprint alike. Every field read here is +// already validated non-empty by validateAzureExecuteBody, so the fingerprint +// can never collapse onto a blank field. +func azureExchangeIdempotencyKey(body AzureExecuteExchangeRequestBody) string { + sources := make([]string, 0, len(body.Sources)) + for _, s := range body.Sources { + id := strings.ToLower(strings.TrimSpace(s.ReservationID)) + sources = append(sources, fmt.Sprintf("%d:%s|%d", len(id), id, s.Quantity)) + } + targets := make([]string, 0, len(body.Targets)) + for _, t := range body.Targets { + sku := strings.ToLower(strings.TrimSpace(t.SKU)) + loc := strings.ToLower(strings.TrimSpace(t.Location)) + term := strings.ToUpper(strings.TrimSpace(t.Term)) + targets = append(targets, fmt.Sprintf("%d:%s|%d:%s|%d:%s|%d", + len(sku), sku, len(loc), loc, len(term), term, t.Quantity)) + } + return exchangeIdempotencyKey( + azureExchangeIdempotencyScope, + strings.ToLower(strings.TrimSpace(body.SubscriptionID)), + sources, targets) +} + +// awsExchangeIdempotencyKey fingerprints an AWS execute request. The same +// inclusion/exclusion reasoning as azureExchangeIdempotencyKey applies: +// max_payment_due_usd is excluded as a cap rather than a purchase attribute. +// +// cloudAccountID is the account the exchange's RIs live in, as already +// resolved by the caller (unattributedAccountConstraint when the deployment +// maps to no registered account -- never empty), and region is part of the +// account scope because RI exchanges are region-scoped. +// +// The target list is canonicalized with exactly the precedence +// pkg/exchange.buildTargetConfigs applies when it builds the actual AWS +// request: `targets[]` when non-empty, otherwise the single legacy +// target_offering_id/target_count pair. Diverging from it would let the two +// request spellings of one purchase fingerprint differently and double-spend. +func awsExchangeIdempotencyKey(cloudAccountID string, body ExchangeExecuteRequestBody) string { + sources := make([]string, 0, len(body.RIIDs)) + for _, raw := range body.RIIDs { + id := strings.ToLower(strings.TrimSpace(raw)) + sources = append(sources, fmt.Sprintf("%d:%s", len(id), id)) + } + + targetTuple := func(offeringID string, count int32) string { + id := strings.ToLower(strings.TrimSpace(offeringID)) + return fmt.Sprintf("%d:%s|%d", len(id), id, count) + } + var targets []string + if len(body.Targets) > 0 { + targets = make([]string, 0, len(body.Targets)) + for _, t := range body.Targets { + targets = append(targets, targetTuple(t.OfferingID, t.Count)) + } + } else { + targets = []string{targetTuple(body.TargetOfferingID, body.TargetCount)} + } + + region := strings.ToLower(strings.TrimSpace(body.Region)) + accountScope := fmt.Sprintf("%d:%s|%d:%s", len(cloudAccountID), cloudAccountID, len(region), region) + return exchangeIdempotencyKey(awsExchangeIdempotencyScope, accountScope, sources, targets) +} + +// claimExchangeSubmit takes the submit-time idempotency claim for key, and is +// called immediately before the irreversible provider commit -- after every +// authorization gate and every money guardrail have passed. +// +// Claiming last is what makes a release path unnecessary: no request that +// failed a gate ever holds a claim, so no legitimate retry is turned away with +// a spurious 409. Conversely, once taken the claim is never released, for the +// rest of the window, WHATEVER the commit does. That is unconditional and not +// only for the ambiguous failures: past the commit call the provider may +// already have accepted the exchange, and a handler that released the claim on +// the errors it believes to be definite rejections would be guessing about the +// one thing it cannot observe. +// +// The cost of that choice is what the 409 below must not paper over. A caller +// receiving it knows only that an identical submit claimed the window; it may +// be running, may have committed, may have failed after reaching the provider, +// or may have failed without committing anything. The message therefore +// asserts none of those and tells the caller to go and look. +// +// A store failure refuses the exchange rather than proceeding unguarded: an +// idempotency guard that cannot be evaluated is not a guard. +func (h *Handler) claimExchangeSubmit(ctx context.Context, key string) error { + claimed, err := h.config.ClaimRIExchangeIdempotencyKey(ctx, key, riExchangeIdempotencyWindow) + if err != nil { + return fmt.Errorf("failed to claim the RI exchange idempotency key: %w", err) + } + if !claimed { + return NewClientError(409, fmt.Sprintf( + "an identical RI exchange claimed this submit within the last %s, so this request was "+ + "not executed; that earlier submit may be running, may have committed, or may have "+ + "failed, so verify its outcome before resubmitting rather than risking a second commitment", + riExchangeIdempotencyWindow)) + } + return nil +} diff --git a/internal/api/ri_exchange_idempotency_guards_test.go b/internal/api/ri_exchange_idempotency_guards_test.go new file mode 100644 index 000000000..8918a5dac --- /dev/null +++ b/internal/api/ri_exchange_idempotency_guards_test.go @@ -0,0 +1,116 @@ +package api + +// ri_exchange_idempotency_guards_test.go -- the input checks that must refuse +// a request BEFORE it can take a submit claim (issue #1642). Split from +// ri_exchange_idempotency_test.go to keep each file under the project's +// 500-line limit; exchangeClaimLedger is defined there. + +import ( + "context" + "testing" + + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// --- refusals that must precede the claim --- + +// TestValidateExecuteExchangeBody_RefusesUncommittableRequests covers the +// fields the AWS execute body used to leave to pkg/exchange, which validates +// them only once ExecuteExchange is already running -- i.e. after the #1642 +// claim. A request that can never commit must not leave a claim behind, and +// these also make the fingerprint's "no component is blank" property true. +func TestValidateExecuteExchangeBody_RefusesUncommittableRequests(t *testing.T) { + const offeringID = "4b2293b4-5fbc-4017-9c75-d5a9d3aa8c91" + base := func() ExchangeExecuteRequestBody { + return ExchangeExecuteRequestBody{ + RIIDs: []string{"ri-123"}, + Targets: []ExchangeTargetBody{{OfferingID: offeringID, Count: 2}}, + MaxPaymentDueUSD: "250.50", + Region: "eu-central-1", + } + } + cases := map[string]struct { + mutate func(*ExchangeExecuteRequestBody) + want string + }{ + "an empty source id": { + func(b *ExchangeExecuteRequestBody) { b.RIIDs = []string{"ri-123", ""} }, + "ri_ids[1] is empty", + }, + "a whitespace-only source id": { + func(b *ExchangeExecuteRequestBody) { b.RIIDs = []string{" "} }, + "ri_ids[0] is empty", + }, + "a zero target count": { + func(b *ExchangeExecuteRequestBody) { b.Targets[0].Count = 0 }, + "targets[0].count must be >= 1", + }, + "a negative target count": { + func(b *ExchangeExecuteRequestBody) { b.Targets[0].Count = -3 }, + "targets[0].count must be >= 1", + }, + "a zero legacy target_count": { + func(b *ExchangeExecuteRequestBody) { + b.Targets = nil + b.TargetOfferingID = offeringID + }, + "target_count must be >= 1", + }, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + b := base() + tc.mutate(&b) + err := validateExecuteExchangeBody(b) + require.Error(t, err) + ce, ok := IsClientError(err) + require.True(t, ok, "expected a ClientError, got: %v", err) + assert.Equal(t, 400, ce.code) + assert.Contains(t, err.Error(), tc.want) + }) + } + + assert.NoError(t, validateExecuteExchangeBody(base()), "the valid body must still pass") +} + +// TestExecuteExchange_InvalidBodyNeverClaims is the ordering half: the refusal +// above has to happen before the ledger is touched, or a request that can +// never commit would hold a claim for the whole window. +func TestExecuteExchange_InvalidBodyNeverClaims(t *testing.T) { + ctx := context.Background() + ledger := newExchangeClaimLedger() + t.Cleanup(func() { ledger.AssertExpectations(t) }) + + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", ctx, "tok").Return(&Session{UserID: "user-1"}, nil) + mockAuth.On("HasPermissionAPI", ctx, "user-1", "execute", "ri-exchange").Return(true, nil) + // Maybe(): the validation below must return before this is reached. It is + // registered anyway so that a regression letting the request through fails + // on the assertions rather than panicking on an unexpected mock call. + mockAuth.On("HasPermissionForConstraintsAPI", ctx, "user-1", "execute", "ri-exchange", mock.Anything). + Return(false, nil).Maybe() + t.Cleanup(func() { mockAuth.AssertExpectations(t) }) + + h := &Handler{ + auth: mockAuth, + config: ledger, + reshapeAccountResolver: func(_ context.Context) (string, error) { + return "11111111-2222-3333-4444-555555555555", nil + }, + } + _, err := h.executeExchange(ctx, &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"authorization": "Bearer tok"}, + Body: `{"ri_ids":["ri-123"],"targets":[{"offering_id":"4b2293b4-5fbc-4017-9c75-d5a9d3aa8c91","count":0}],` + + `"max_payment_due_usd":"250.50","region":"eu-central-1"}`, + }) + require.Error(t, err) + ce, ok := IsClientError(err) + require.True(t, ok, "expected a ClientError, got: %v", err) + assert.Equal(t, 400, ce.code) + assert.Empty(t, ledger.claimedKeys(), "a request refused at validation must not hold a claim") + mockAuth.AssertNotCalled(t, "HasPermissionForConstraintsAPI", + mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything) +} diff --git a/internal/api/ri_exchange_idempotency_key_test.go b/internal/api/ri_exchange_idempotency_key_test.go new file mode 100644 index 000000000..6b573b5c2 --- /dev/null +++ b/internal/api/ri_exchange_idempotency_key_test.go @@ -0,0 +1,287 @@ +package api + +// ri_exchange_idempotency_key_test.go -- what the RI exchange submit +// fingerprint is made of (issue #1642). Split from +// ri_exchange_idempotency_test.go to keep each file under the project's +// 500-line limit; the handler-level tests that consume these keys live there. + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/assert" +) + +// --- key composition --- +// +// The fingerprint has to survive aliasing in BOTH directions, and the two +// directions fail differently: +// +// - two tokens standing for one purchase lets a retry through and spends +// twice (the #1642 double spend); +// - one token standing for two purchases silently swallows a genuinely +// distinct second exchange behind a 409. + +func azureKeyBody() AzureExecuteExchangeRequestBody { + return AzureExecuteExchangeRequestBody{ + SubscriptionID: "sub-1", + Sources: []AzureExchangeSourceBody{{ReservationID: "res-1", Quantity: 4}}, + Targets: []AzureExchangeTargetBody{{SKU: "Standard_D8s_v3", Location: "eastus", Term: "P1Y", Quantity: 4}}, + MaxPaymentDue: "500.00", + Currency: "USD", + } +} + +// TestAzureExchangeIdempotencyKey_SubscriptionScoped is the #1495 precedent +// applied here: ListExchangeableReservations is TENANT-wide, so two +// subscriptions can legitimately present the same reservation and target +// shapes. A scope-blind fingerprint would let one subscription's exchange +// suppress another's. +func TestAzureExchangeIdempotencyKey_SubscriptionScoped(t *testing.T) { + a := azureKeyBody() + b := azureKeyBody() + b.SubscriptionID = "sub-2" + assert.NotEqual(t, azureExchangeIdempotencyKey(a), azureExchangeIdempotencyKey(b), + "a fingerprint blind to subscription_id aliases across the tenant") +} + +// TestAzureExchangeIdempotencyKey_StableAcrossNonPurchaseFields pins the other +// direction: everything that does not change WHAT is bought must leave the +// fingerprint alone, or a retry mints a fresh key and commits a second time. +func TestAzureExchangeIdempotencyKey_StableAcrossNonPurchaseFields(t *testing.T) { + base := azureKeyBody() + base.Sources = append(base.Sources, AzureExchangeSourceBody{ReservationID: "res-2", Quantity: 1}) + base.Targets = append(base.Targets, AzureExchangeTargetBody{SKU: "Standard_D2s_v3", Location: "eastus", Term: "P3Y", Quantity: 1}) + want := azureExchangeIdempotencyKey(base) + + variants := map[string]func(*AzureExecuteExchangeRequestBody){ + "a raised spend cap": func(b *AzureExecuteExchangeRequestBody) { + b.MaxPaymentDue = "9999.00" + }, + "a differently spelled currency": func(b *AzureExecuteExchangeRequestBody) { + b.Currency = "usd" + }, + "the billing scope spelled out rather than omitted": func(b *AzureExecuteExchangeRequestBody) { + for i := range b.Targets { + b.Targets[i].BillingScopeID = "/subscriptions/sub-1" + } + }, + "ARM identifiers in a different case": func(b *AzureExecuteExchangeRequestBody) { + b.SubscriptionID = strings.ToUpper(b.SubscriptionID) + b.Sources[0].ReservationID = strings.ToUpper(b.Sources[0].ReservationID) + b.Targets[0].SKU = strings.ToUpper(b.Targets[0].SKU) + b.Targets[0].Location = "EastUS" + b.Targets[0].Term = "p1y" + }, + "surrounding whitespace": func(b *AzureExecuteExchangeRequestBody) { + b.SubscriptionID = " sub-1 " + b.Sources[0].ReservationID = "res-1 " + b.Targets[0].SKU = " Standard_D8s_v3" + }, + "sources and targets sent in the opposite order": func(b *AzureExecuteExchangeRequestBody) { + b.Sources[0], b.Sources[1] = b.Sources[1], b.Sources[0] + b.Targets[0], b.Targets[1] = b.Targets[1], b.Targets[0] + }, + } + for name, mutate := range variants { + t.Run(name, func(t *testing.T) { + b := azureKeyBody() + b.Sources = append(b.Sources, AzureExchangeSourceBody{ReservationID: "res-2", Quantity: 1}) + b.Targets = append(b.Targets, AzureExchangeTargetBody{SKU: "Standard_D2s_v3", Location: "eastus", Term: "P3Y", Quantity: 1}) + mutate(&b) + assert.Equal(t, want, azureExchangeIdempotencyKey(b), + "%s does not change what the exchange buys, so it must not mint a fresh key", name) + }) + } +} + +// TestAzureExchangeIdempotencyKey_DistinguishesEveryPurchaseField walks every +// field that DOES change what is bought and requires each to move the +// fingerprint. A field missing here would let a distinct exchange be swallowed +// by an earlier one's claim. +func TestAzureExchangeIdempotencyKey_DistinguishesEveryPurchaseField(t *testing.T) { + // Keyed by fingerprint so a collision with ANY earlier variant is caught, + // not just with the baseline. + seen := map[string]string{azureExchangeIdempotencyKey(azureKeyBody()): "the baseline exchange"} + + variants := map[string]func(*AzureExecuteExchangeRequestBody){ + "a different source reservation": func(b *AzureExecuteExchangeRequestBody) { + b.Sources[0].ReservationID = "res-9" + }, + "a different source quantity": func(b *AzureExecuteExchangeRequestBody) { + b.Sources[0].Quantity = 5 + }, + "an extra source": func(b *AzureExecuteExchangeRequestBody) { + b.Sources = append(b.Sources, AzureExchangeSourceBody{ReservationID: "res-2", Quantity: 1}) + }, + "a different target SKU": func(b *AzureExecuteExchangeRequestBody) { + b.Targets[0].SKU = "Standard_D16s_v3" + }, + "a different target location": func(b *AzureExecuteExchangeRequestBody) { + b.Targets[0].Location = "westeurope" + }, + "a different target term": func(b *AzureExecuteExchangeRequestBody) { + b.Targets[0].Term = "P3Y" + }, + "a different target quantity": func(b *AzureExecuteExchangeRequestBody) { + b.Targets[0].Quantity = 8 + }, + "an extra target": func(b *AzureExecuteExchangeRequestBody) { + b.Targets = append(b.Targets, AzureExchangeTargetBody{SKU: "Standard_D2s_v3", Location: "eastus", Term: "P1Y", Quantity: 1}) + }, + } + for name, mutate := range variants { + b := azureKeyBody() + mutate(&b) + key := azureExchangeIdempotencyKey(b) + if other, clash := seen[key]; clash { + t.Errorf("%q fingerprints the same as %q; the second exchange would be silently refused", name, other) + continue + } + seen[key] = name + } +} + +func awsKeyBody() ExchangeExecuteRequestBody { + return ExchangeExecuteRequestBody{ + RIIDs: []string{"ri-123"}, + Targets: []ExchangeTargetBody{{OfferingID: "off-1", Count: 2}}, + MaxPaymentDueUSD: "250.50", + Region: "eu-central-1", + } +} + +// TestAwsExchangeIdempotencyKey_ScopedToAccountAndRegion pins the AWS scope +// dimensions. RI exchanges are region-scoped and act on the RIs of whichever +// account the deployment resolves to, so both belong in the fingerprint: two +// accounts, or two regions of one account, can hold same-named RI ids. +func TestAwsExchangeIdempotencyKey_ScopedToAccountAndRegion(t *testing.T) { + body := awsKeyBody() + base := awsExchangeIdempotencyKey("acct-1", body) + + otherRegion := body + otherRegion.Region = "us-east-1" + + assert.NotEqual(t, base, awsExchangeIdempotencyKey("acct-2", body), + "a fingerprint blind to the cloud account aliases across accounts") + assert.NotEqual(t, base, awsExchangeIdempotencyKey("acct-1", otherRegion), + "a fingerprint blind to the region aliases across regions") + assert.NotEqual(t, base, awsExchangeIdempotencyKey(unattributedAccountConstraint, body), + "the unattributed sentinel is its own scope, not a wildcard") +} + +// TestAwsExchangeIdempotencyKey_StableAcrossNonPurchaseFields mirrors the +// Azure stability test. The legacy/array equivalence matters most: the two +// spellings build the identical AWS request (pkg/exchange.buildTargetConfigs), +// so a retry that switches spelling must not mint a fresh key. +func TestAwsExchangeIdempotencyKey_StableAcrossNonPurchaseFields(t *testing.T) { + want := awsExchangeIdempotencyKey("acct-1", awsKeyBody()) + + variants := map[string]func(*ExchangeExecuteRequestBody){ + "a raised spend cap": func(b *ExchangeExecuteRequestBody) { + b.MaxPaymentDueUSD = "9999.00" + }, + "the legacy singleton spelling of the same target": func(b *ExchangeExecuteRequestBody) { + b.Targets = nil + b.TargetOfferingID = "off-1" + b.TargetCount = 2 + }, + "legacy fields shadowed by an equivalent targets[]": func(b *ExchangeExecuteRequestBody) { + b.TargetOfferingID = "off-ignored" + b.TargetCount = 99 + }, + "identifiers in a different case, with whitespace": func(b *ExchangeExecuteRequestBody) { + b.RIIDs = []string{" RI-123"} + b.Targets = []ExchangeTargetBody{{OfferingID: "OFF-1 ", Count: 2}} + b.Region = "EU-Central-1" + }, + } + for name, mutate := range variants { + t.Run(name, func(t *testing.T) { + b := awsKeyBody() + mutate(&b) + assert.Equal(t, want, awsExchangeIdempotencyKey("acct-1", b), + "%s does not change what the exchange buys, so it must not mint a fresh key", name) + }) + } +} + +// TestAwsExchangeIdempotencyKey_DistinguishesEveryPurchaseField is the +// no-swallowing direction for AWS. +func TestAwsExchangeIdempotencyKey_DistinguishesEveryPurchaseField(t *testing.T) { + seen := map[string]string{awsExchangeIdempotencyKey("acct-1", awsKeyBody()): "the baseline exchange"} + + variants := map[string]func(*ExchangeExecuteRequestBody){ + "a different source RI": func(b *ExchangeExecuteRequestBody) { + b.RIIDs = []string{"ri-999"} + }, + "an extra source RI": func(b *ExchangeExecuteRequestBody) { + b.RIIDs = append(b.RIIDs, "ri-456") + }, + "the same source RI listed twice": func(b *ExchangeExecuteRequestBody) { + b.RIIDs = append(b.RIIDs, "ri-123") + }, + "a different target offering": func(b *ExchangeExecuteRequestBody) { + b.Targets[0].OfferingID = "off-9" + }, + "a different target count": func(b *ExchangeExecuteRequestBody) { + b.Targets[0].Count = 3 + }, + "an extra target": func(b *ExchangeExecuteRequestBody) { + b.Targets = append(b.Targets, ExchangeTargetBody{OfferingID: "off-2", Count: 1}) + }, + } + for name, mutate := range variants { + b := awsKeyBody() + mutate(&b) + key := awsExchangeIdempotencyKey("acct-1", b) + if other, clash := seen[key]; clash { + t.Errorf("%q fingerprints the same as %q; the second exchange would be silently refused", name, other) + continue + } + seen[key] = name + } +} + +// TestExchangeIdempotencyKey_ProviderScopesNeverCollide feeds the AWS and +// Azure derivations component lists that are identical by construction. The +// provider scope is what keeps them apart, and it must, because the two share +// one claim table. +func TestExchangeIdempotencyKey_ProviderScopesNeverCollide(t *testing.T) { + sources := []string{"same-source"} + targets := []string{"same-target"} + assert.NotEqual(t, + exchangeIdempotencyKey(azureExchangeIdempotencyScope, "same-scope", sources, targets), + exchangeIdempotencyKey(awsExchangeIdempotencyScope, "same-scope", sources, targets)) +} + +// TestExchangeIdempotencyKey_FieldBoundariesAreUnambiguous pins the +// length-prefixed encoding. Without it a value carrying the separator could be +// crafted to shift a field boundary, making two genuinely different exchanges +// fingerprint alike -- which suppresses the second one. +func TestExchangeIdempotencyKey_FieldBoundariesAreUnambiguous(t *testing.T) { + cases := map[string][2]string{ + "a character moved across the scope boundary": { + exchangeIdempotencyKey("ab", "c", nil, nil), + exchangeIdempotencyKey("a", "bc", nil, nil), + }, + "an element moved from sources to targets": { + exchangeIdempotencyKey("s", "a", []string{"x", "y"}, nil), + exchangeIdempotencyKey("s", "a", []string{"x"}, []string{"y"}), + }, + "an element carrying the separator": { + exchangeIdempotencyKey("s", "a", []string{"x|y"}, nil), + exchangeIdempotencyKey("s", "a", []string{"x", "y"}, nil), + }, + "an empty element versus no element": { + exchangeIdempotencyKey("s", "a", []string{""}, nil), + exchangeIdempotencyKey("s", "a", nil, nil), + }, + } + for name, pair := range cases { + assert.NotEqual(t, pair[0], pair[1], "%s must change the fingerprint", name) + } +} + +// TestClaimExchangeSubmit_Outcomes covers the three answers the ledger can +// give, independently of either handler. diff --git a/internal/api/ri_exchange_idempotency_test.go b/internal/api/ri_exchange_idempotency_test.go new file mode 100644 index 000000000..55364a342 --- /dev/null +++ b/internal/api/ri_exchange_idempotency_test.go @@ -0,0 +1,312 @@ +package api + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/CUDly/internal/config" + azurecompute "github.com/LeanerCloud/CUDly/providers/azure/services/compute" + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// --- submit-time idempotency for the RI exchange execute endpoints (#1642) --- + +// exchangeClaimLedger is an in-memory stand-in for the ri_exchange_idempotency +// table, implementing the same contract PostgresStore.ClaimRIExchangeIdempotencyKey +// does: the first claimant of a fingerprint wins, every later claimant inside +// the window loses. +// +// It is a real implementation rather than a canned mock sequence on purpose. +// The double-spend in #1642 is two POSTs deduping against each other, so the +// test has to let the handler's own key derivation decide whether the second +// submit collides. A mock returning true-then-false would pass even if the +// handler fingerprinted the two requests differently, which is exactly the bug +// class this guards. +type exchangeClaimLedger struct { + *MockConfigStore + + mu sync.Mutex + held map[string]bool + keys []string + windows []time.Duration + + // holdAll makes every claim lose, standing for "an identical submit is + // already in flight" without the test having to derive the key itself. + holdAll bool + // err, when set, fails every claim: an idempotency guard that cannot be + // evaluated must refuse the exchange rather than wave it through. + err error +} + +func newExchangeClaimLedger() *exchangeClaimLedger { + return &exchangeClaimLedger{MockConfigStore: &MockConfigStore{}, held: map[string]bool{}} +} + +func (l *exchangeClaimLedger) ClaimRIExchangeIdempotencyKey(_ context.Context, key string, window time.Duration) (bool, error) { + l.mu.Lock() + defer l.mu.Unlock() + l.keys = append(l.keys, key) + l.windows = append(l.windows, window) + if l.err != nil { + return false, l.err + } + if l.holdAll || l.held[key] { + return false, nil + } + l.held[key] = true + return true, nil +} + +// claimedKeys returns the fingerprints the handler asked to claim, in order. +func (l *exchangeClaimLedger) claimedKeys() []string { + l.mu.Lock() + defer l.mu.Unlock() + return append([]string(nil), l.keys...) +} + +// azureDoubleSpendHandler wires an Azure execute handler whose every gate +// passes, so the only thing standing between two identical POSTs and two +// commitments is the submit-time claim. committed accumulates the session IDs +// ExecuteExchange was actually called with, which is the count of real money +// movements. +// +// CalculateExchange deliberately hands back a DIFFERENT session on each call, +// reproducing the property that makes #1642 possible: the handler re-quotes +// before every commit, so Azure's own session-level replay protection never +// sees two POSTs of one logical exchange as the same operation. +func azureDoubleSpendHandler(t *testing.T, ledger *exchangeClaimLedger, committed *[]string) *Handler { + t.Helper() + ctx := context.Background() + + opsClient := new(mockAzureExchangeOpsClient) + // The #1642 scenario's source: a reservation holding 10, of which the + // request below exchanges 4. + opsClient.On("ListExchangeableReservations", mock.Anything).Return([]azurecompute.ExchangeableReservation{ + {ReservationID: "res-1", BillingScopeID: "/subscriptions/sub-1", Region: "eastus", Quantity: 10}, + }, nil) + opsClient.On("CalculateExchange", ctx, mock.Anything, mock.Anything).Return( + &azurecompute.ExchangePreview{SessionID: "sess-1", NetPayable: toPtr(400.00), NetPayableCurrency: "USD"}, + []azurecompute.CompatibleOffering{}, nil, + ).Once() + // Maybe(): the store-failure test below re-quotes only once, and an unmet + // required expectation there would fail for the wrong reason. + opsClient.On("CalculateExchange", ctx, mock.Anything, mock.Anything).Return( + &azurecompute.ExchangePreview{SessionID: "sess-2", NetPayable: toPtr(400.00), NetPayableCurrency: "USD"}, + []azurecompute.CompatibleOffering{}, nil, + ).Maybe() + // Registered for any session and without Once() so that a regression + // committing twice fails by ASSERTION on the recorded sessions below, + // rather than panicking the test binary on an unexpected mock call. + opsClient.On("ExecuteExchange", ctx, mock.Anything).Run(func(args mock.Arguments) { + *committed = append(*committed, args.String(1)) + }).Return(&azurecompute.ExchangeResult{SessionID: "sess-1", Status: "Succeeded"}, nil).Maybe() + t.Cleanup(func() { opsClient.AssertExpectations(t) }) + + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", ctx, "tok").Return(&Session{UserID: "user-1"}, nil) + mockAuth.On("HasPermissionAPI", ctx, "user-1", "execute", "ri-exchange").Return(true, nil) + allowAnyAccountScope(mockAuth) + mockAuth.On("HasPermissionForConstraintsAPI", ctx, "user-1", "execute", "ri-exchange", mock.Anything).Return(true, nil) + t.Cleanup(func() { mockAuth.AssertExpectations(t) }) + + ledger.GetCloudAccountByExternalIDFn = func(_ context.Context, _, _ string) (*config.CloudAccount, error) { + return &config.CloudAccount{ID: "acct-mine", Name: "My Team"}, nil + } + t.Cleanup(func() { ledger.AssertExpectations(t) }) + + return &Handler{ + auth: mockAuth, + config: ledger, + azureExchangeFactory: func(_ string) azureExchangeClient { return opsClient }, + } +} + +// azureDoubleSpendBody is the #1642 request: exchange 4 of the 10 units held +// by res-1 in sub-1 for 4 units of a different SKU. +const azureDoubleSpendBody = `{ + "subscription_id": "sub-1", + "sources": [{"reservation_id": "res-1", "quantity": 4}], + "targets": [{"sku": "Standard_D8s_v3", "location": "eastus", "term": "P1Y", "quantity": 4}], + "max_payment_due": "500.00", + "currency": "USD" +}` + +func azureExecuteRequest(body string) *events.LambdaFunctionURLRequest { + return &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"authorization": "Bearer tok"}, + Body: body, + } +} + +// TestExecuteAzureExchange_RetryOfInFlightSubmitDoesNotDoubleSpend is the +// regression test for #1642, replaying the issue's scenario exactly: a caller +// exchanges 4 units of a reservation holding 10, the long-running operation +// outlives the client's patience, and the client re-POSTs the identical body. +// +// Before the fix the second POST re-quoted against the now-smaller reservation, +// cleared every gate (ownership, scope, constraints, currency, cap) because +// each half is individually under the cap, and committed a SECOND exchange: +// two calls to ExecuteExchange, roughly twice the intended spend. +func TestExecuteAzureExchange_RetryOfInFlightSubmitDoesNotDoubleSpend(t *testing.T) { + ctx := context.Background() + ledger := newExchangeClaimLedger() + var committed []string + h := azureDoubleSpendHandler(t, ledger, &committed) + + _, err := h.executeAzureExchange(ctx, azureExecuteRequest(azureDoubleSpendBody)) + require.NoError(t, err, "the first submit of a valid exchange must go through") + + // The client timed out on the still-polling first request and retried it + // verbatim. This is the double spend. + _, err = h.executeAzureExchange(ctx, azureExecuteRequest(azureDoubleSpendBody)) + require.Error(t, err, "an identical resubmit must be refused, not committed a second time") + ce, ok := IsClientError(err) + require.True(t, ok, "expected a ClientError, got: %v", err) + assert.Equal(t, 409, ce.code) + assert.Contains(t, err.Error(), "identical RI exchange claimed this submit") + // The refusal must not promise the caller that the earlier submit + // committed: the claim is retained even when the provider call failed, so + // a client reading 409 as "it already succeeded" would skip verifying a + // purchase that may never have happened. + assert.Contains(t, err.Error(), "verify its outcome before resubmitting") + + assert.Equal(t, []string{"sess-1"}, committed, + "exactly one exchange may reach Azure; a second commit is the #1642 double spend") + + keys := ledger.claimedKeys() + require.Len(t, keys, 2, "both submits must consult the claim ledger") + assert.Equal(t, keys[0], keys[1], + "two POSTs of one logical exchange must fingerprint identically, or the guard never fires") + assert.Equal(t, []time.Duration{riExchangeIdempotencyWindow, riExchangeIdempotencyWindow}, ledger.windows) +} + +// TestExecuteAzureExchange_ClaimStoreFailureRefusesCommit pins the fail-loud +// half: a claim that cannot be evaluated is not a claim, so the exchange is +// refused rather than committed unguarded. +func TestExecuteAzureExchange_ClaimStoreFailureRefusesCommit(t *testing.T) { + ctx := context.Background() + ledger := newExchangeClaimLedger() + ledger.err = fmt.Errorf("connection refused") + var committed []string + h := azureDoubleSpendHandler(t, ledger, &committed) + + _, err := h.executeAzureExchange(ctx, azureExecuteRequest(azureDoubleSpendBody)) + require.Error(t, err) + assert.Contains(t, err.Error(), "claim the RI exchange idempotency key") + assert.Empty(t, committed, "a store failure must not fall through to the irreversible commit") +} + +// TestExecuteExchange_DuplicateSubmitRefusedBeforeAWS is the AWS half of the +// same gap (the issue's scope note): AcceptReservedInstancesExchangeQuote +// carries no ClientToken, so nothing downstream deduplicates a retry. +// +// The ledger reports every fingerprint as already held, standing for a first +// submit still in flight. Both requests describe the SAME purchase in the two +// spellings the endpoint accepts -- the legacy target_offering_id/target_count +// singleton and the targets[] array -- so the test also pins that the handler +// fingerprints them alike, matching pkg/exchange.buildTargetConfigs. Diverging +// there would let one request shape retry as the other and double-spend. +// +// A 409 also proves the claim precedes the AWS call: reaching +// exchange.ExecuteExchange without credentials would surface as a 5xx. +func TestExecuteExchange_DuplicateSubmitRefusedBeforeAWS(t *testing.T) { + ctx := context.Background() + const deploymentAccountID = "11111111-2222-3333-4444-555555555555" + + ledger := newExchangeClaimLedger() + ledger.holdAll = true + t.Cleanup(func() { ledger.AssertExpectations(t) }) + + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", ctx, "tok").Return(&Session{UserID: "user-1"}, nil) + mockAuth.On("HasPermissionAPI", ctx, "user-1", "execute", "ri-exchange").Return(true, nil) + mockAuth.On("HasPermissionForConstraintsAPI", ctx, "user-1", "execute", "ri-exchange", mock.Anything).Return(true, nil) + t.Cleanup(func() { mockAuth.AssertExpectations(t) }) + + h := &Handler{ + auth: mockAuth, + config: ledger, + reshapeAccountResolver: func(_ context.Context) (string, error) { return deploymentAccountID, nil }, + } + + // targets[] entries are checked against offeringIDPattern, so the shared + // offering must be a real AWS offering UUID for the two spellings to be + // comparable at all. + const offeringID = "4b2293b4-5fbc-4017-9c75-d5a9d3aa8c91" + bodies := map[string]string{ + "legacy singleton": `{"ri_ids":["ri-123"],"target_offering_id":"` + offeringID + `","target_count":2,` + + `"max_payment_due_usd":"250.50","region":"eu-central-1"}`, + "targets array": `{"ri_ids":["ri-123"],"targets":[{"offering_id":"` + offeringID + `","count":2}],` + + `"max_payment_due_usd":"250.50","region":"eu-central-1"}`, + } + for name, body := range bodies { + t.Run(name, func(t *testing.T) { + _, err := h.executeExchange(ctx, &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"authorization": "Bearer tok"}, + Body: body, + }) + require.Error(t, err) + ce, ok := IsClientError(err) + require.True(t, ok, "expected a ClientError, got: %v", err) + assert.Equal(t, 409, ce.code, "a held claim must refuse the submit before AWS is called") + }) + } + + keys := ledger.claimedKeys() + require.Len(t, keys, 2) + assert.Equal(t, keys[0], keys[1], + "the legacy and targets[] spellings of one purchase must fingerprint alike") +} + +func TestClaimExchangeSubmit_Outcomes(t *testing.T) { + ctx := context.Background() + + t.Run("won", func(t *testing.T) { + ledger := newExchangeClaimLedger() + t.Cleanup(func() { ledger.AssertExpectations(t) }) + h := &Handler{config: ledger} + require.NoError(t, h.claimExchangeSubmit(ctx, "key-1")) + }) + + t.Run("lost", func(t *testing.T) { + ledger := newExchangeClaimLedger() + ledger.holdAll = true + t.Cleanup(func() { ledger.AssertExpectations(t) }) + h := &Handler{config: ledger} + err := h.claimExchangeSubmit(ctx, "key-1") + require.Error(t, err) + ce, ok := IsClientError(err) + require.True(t, ok, "expected a ClientError, got: %v", err) + assert.Equal(t, 409, ce.code) + assert.Contains(t, err.Error(), riExchangeIdempotencyWindow.String(), + "the refusal must tell the caller how long the claim holds") + + // The claim is retained even when the provider call FAILED, so the + // refusal must leave the earlier submit's outcome open. A client that + // reads 409 as "the earlier one succeeded" skips verification and + // assumes a purchase that may never have happened. + for _, outcome := range []string{"may be running", "may have committed", "may have failed"} { + assert.Contains(t, err.Error(), outcome, + "the refusal must not assert which outcome the earlier submit had") + } + assert.Contains(t, err.Error(), "verify its outcome before resubmitting") + }) + + t.Run("store failure refuses rather than proceeding unguarded", func(t *testing.T) { + ledger := newExchangeClaimLedger() + ledger.err = fmt.Errorf("connection refused") + t.Cleanup(func() { ledger.AssertExpectations(t) }) + h := &Handler{config: ledger} + err := h.claimExchangeSubmit(ctx, "key-1") + require.Error(t, err) + _, isClient := IsClientError(err) + assert.False(t, isClient, "a store outage is not the caller's fault; it must not surface as a 4xx") + assert.Contains(t, err.Error(), "connection refused") + }) +} diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 5af8b3b6d..b20d06c25 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -265,6 +265,26 @@ type StoreInterface interface { // FlipPurchaseRevocationInFlight. ClaimMarketplaceListingSlot(ctx context.Context, purchaseID string) (bool, error) + // ClaimRIExchangeIdempotencyKey atomically claims key for an RI exchange + // submit, so a client that retries a timed-out execute request cannot + // commit the same exchange twice (issue #1642). key is a fingerprint of + // what the request buys; the caller derives it and claims it immediately + // before the irreversible provider call. + // + // Returns (true, nil) when this call won the claim and may proceed to + // commit, and (false, nil) when another submit of the same fingerprint + // claimed it less than window ago (the caller maps this to a 409). + // A claim older than window is reclaimable, so a genuine intentional + // repeat of the same exchange is not blocked forever. + // + // There is no release: once taken, a claim stands for the rest of the + // window whatever the provider call does, including when it fails. Past + // the point of submission the outcome is ambiguous, and holding the claim + // is the fail-closed choice. The consequence for the caller's 409 is that + // it cannot promise the earlier submit committed, only that it claimed the + // window. + ClaimRIExchangeIdempotencyKey(ctx context.Context, key string, window time.Duration) (bool, error) + // RI Exchange history SaveRIExchangeRecord(ctx context.Context, record *RIExchangeRecord) error GetRIExchangeRecord(ctx context.Context, id string) (*RIExchangeRecord, error) diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 9e8d8e514..a2849b4e6 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -2498,6 +2498,40 @@ func (s *PostgresStore) GetPurchaseHistoryInFlight(ctx context.Context) ([]*Purc // RI EXCHANGE HISTORY // ========================================== +// ClaimRIExchangeIdempotencyKey atomically claims an RI exchange submit +// fingerprint (issue #1642). See the interface doc for the contract. +// +// The whole decision is one statement, so two concurrent submits of the same +// fingerprint cannot both win: the INSERT and the conditional takeover of an +// expired row are the same atomic row operation, with no read-then-write +// window between them. RowsAffected() is 1 only when this call inserted the +// row or took over an expired one; the ON CONFLICT ... WHERE predicate +// evaluating false leaves it at 0. +// +// Both the write and the expiry comparison use now(), the DATABASE clock, so +// clock skew between concurrent application instances cannot distort the +// window. +func (s *PostgresStore) ClaimRIExchangeIdempotencyKey(ctx context.Context, key string, window time.Duration) (bool, error) { + if key == "" { + return false, fmt.Errorf("refusing to claim an empty RI exchange idempotency key") + } + if window <= 0 { + return false, fmt.Errorf("RI exchange idempotency window must be positive, got %s", window) + } + query := ` + INSERT INTO ri_exchange_idempotency (idempotency_key, claimed_at) + VALUES ($1, now()) + ON CONFLICT (idempotency_key) DO UPDATE + SET claimed_at = now() + WHERE ri_exchange_idempotency.claimed_at < now() - make_interval(secs => $2) + ` + tag, err := s.db.Exec(ctx, query, key, window.Seconds()) + if err != nil { + return false, fmt.Errorf("failed to claim RI exchange idempotency key: %w", err) + } + return tag.RowsAffected() == 1, nil +} + // SaveRIExchangeRecord saves an RI exchange record. func (s *PostgresStore) SaveRIExchangeRecord(ctx context.Context, record *RIExchangeRecord) error { if record.ID == "" { diff --git a/internal/config/store_postgres_db_test.go b/internal/config/store_postgres_db_test.go index 1bdc06243..c7311979a 100644 --- a/internal/config/store_postgres_db_test.go +++ b/internal/config/store_postgres_db_test.go @@ -1437,3 +1437,111 @@ func TestPostgresStoreDB_CleanupOldExecutions_RetainsCanceledInsideHealthWindow( } assert.Equal(t, wantDeleted, deleted, "RowsAffected must match the rows actually purged") } + +// TestPostgresStoreDB_ClaimRIExchangeIdempotencyKey is the live-database half +// of the #1642 guard. It runs against a real PostgreSQL rather than pgxmock +// because what is under test is what the STATEMENT means, not how it is +// spelled: whether a single INSERT ... ON CONFLICT DO UPDATE ... WHERE really +// lets exactly one of several racing submits through, and whether the window +// predicate is evaluated against the database clock. +// +// A pgxmock test can only assert that RowsAffected is read correctly; it would +// pass just as happily against a statement that let both submits win. +func TestPostgresStoreDB_ClaimRIExchangeIdempotencyKey(t *testing.T) { + conn := setupTestContainerDB(t) + if conn == nil { + return + } + store := NewPostgresStore(conn) + ctx := context.Background() + + const window = 15 * time.Minute + _, err := conn.Exec(ctx, "DELETE FROM ri_exchange_idempotency") + require.NoError(t, err) + + t.Run("first submit wins and the identical retry loses", func(t *testing.T) { + const key = "fingerprint-sequential" + claimed, err := store.ClaimRIExchangeIdempotencyKey(ctx, key, window) + require.NoError(t, err) + require.True(t, claimed, "the first submit of a fingerprint must win the claim") + + claimed, err = store.ClaimRIExchangeIdempotencyKey(ctx, key, window) + require.NoError(t, err) + assert.False(t, claimed, "the retry that would double-spend must lose") + }) + + t.Run("a different fingerprint is unaffected", func(t *testing.T) { + claimed, err := store.ClaimRIExchangeIdempotencyKey(ctx, "fingerprint-other", window) + require.NoError(t, err) + assert.True(t, claimed, "one exchange's claim must not suppress a genuinely different one") + }) + + t.Run("a submit racing an uncommitted claim loses", func(t *testing.T) { + const key = "fingerprint-concurrent" + + // Hold a claim on key from a separate, still-open transaction. This + // is the race the single-statement claim exists to win: a claimant + // that READ the ledger and then wrote would see nothing (the other + // transaction has not committed), conclude the fingerprint is free, + // and let a second exchange commit. + // + // Racing N goroutines instead would not test this. They serialize on + // connection acquisition, so the first claim commits before the next + // one starts and a read-then-write implementation passes too -- + // verified by mutating this store method into exactly that shape. + tx, err := conn.Begin(ctx) + require.NoError(t, err) + _, err = tx.Exec(ctx, + `INSERT INTO ri_exchange_idempotency (idempotency_key, claimed_at) VALUES ($1, now())`, key) + require.NoError(t, err) + + type claimResult struct { + claimed bool + err error + } + done := make(chan claimResult, 1) + go func() { + claimed, err := store.ClaimRIExchangeIdempotencyKey(ctx, key, window) + done <- claimResult{claimed, err} + }() + + // The claim must contend for the row rather than answer from a read, + // so it cannot return while the conflicting write is uncommitted. The + // wait is the only way to observe "still blocked"; it is a lower bound + // on the observation, not a sleep the assertion depends on. + select { + case r := <-done: + _ = tx.Rollback(ctx) + t.Fatalf("the claim answered %v while a conflicting claim was still uncommitted; "+ + "it read the ledger instead of contending for the row", r.claimed) + case <-time.After(300 * time.Millisecond): + } + + require.NoError(t, tx.Commit(ctx)) + r := <-done + require.NoError(t, r.err) + assert.False(t, r.claimed, "the loser of the race must not also claim the submit") + }) + + t.Run("a claim older than the window is reclaimable", func(t *testing.T) { + const key = "fingerprint-expired" + claimed, err := store.ClaimRIExchangeIdempotencyKey(ctx, key, window) + require.NoError(t, err) + require.True(t, claimed) + + // Back-date the claim past the window rather than sleeping: an + // intentional repeat of the same exchange must not be blocked forever. + _, err = conn.Exec(ctx, + `UPDATE ri_exchange_idempotency SET claimed_at = now() - INTERVAL '16 minutes' WHERE idempotency_key = $1`, key) + require.NoError(t, err) + + claimed, err = store.ClaimRIExchangeIdempotencyKey(ctx, key, window) + require.NoError(t, err) + assert.True(t, claimed, "an expired claim must be reclaimable") + + // ...and taking it over restarts the window. + claimed, err = store.ClaimRIExchangeIdempotencyKey(ctx, key, window) + require.NoError(t, err) + assert.False(t, claimed, "the takeover must refresh claimed_at, not leave the row expired") + }) +} diff --git a/internal/config/store_postgres_ri_exchange_idempotency_test.go b/internal/config/store_postgres_ri_exchange_idempotency_test.go new file mode 100644 index 000000000..e93ff24e8 --- /dev/null +++ b/internal/config/store_postgres_ri_exchange_idempotency_test.go @@ -0,0 +1,101 @@ +package config + +// store_postgres_ri_exchange_idempotency_test.go — pgxmock coverage for the +// RI exchange submit-time claim (issue #1642). The claim decides whether an +// irreversible exchange commits, so both of its answers and its failure mode +// are pinned here; the real concurrency and window semantics are exercised +// against a live Postgres in store_postgres_db_test.go. + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/pashagolub/pgxmock/v4" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestPGXMock_ClaimRIExchangeIdempotencyKey_WonWhenRowWritten pins the winning +// answer AND the shape of the statement that produces it: the whole decision +// is one Exec, and the window travels as seconds into make_interval so the +// expiry is compared against the DATABASE clock rather than the caller's. +func TestPGXMock_ClaimRIExchangeIdempotencyKey_WonWhenRowWritten(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + + mock.ExpectExec("INSERT INTO ri_exchange_idempotency"). + WithArgs("fingerprint-1", 900.0). + WillReturnResult(pgxmock.NewResult("INSERT", 1)) + + claimed, err := store.ClaimRIExchangeIdempotencyKey(context.Background(), "fingerprint-1", 15*time.Minute) + require.NoError(t, err) + assert.True(t, claimed, "an inserted row means this caller owns the submit") + assert.NoError(t, mock.ExpectationsWereMet()) +} + +// TestPGXMock_ClaimRIExchangeIdempotencyKey_LostWhenConflictHeld is the branch +// that stops the double spend: the ON CONFLICT predicate finds the existing +// claim still inside its window, updates nothing, and the caller must read +// that zero-row result as "someone else already submitted this". +func TestPGXMock_ClaimRIExchangeIdempotencyKey_LostWhenConflictHeld(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + + mock.ExpectExec("INSERT INTO ri_exchange_idempotency"). + WithArgs("fingerprint-1", 900.0). + WillReturnResult(pgxmock.NewResult("INSERT", 0)) + + claimed, err := store.ClaimRIExchangeIdempotencyKey(context.Background(), "fingerprint-1", 15*time.Minute) + require.NoError(t, err) + assert.False(t, claimed, "a live claim held by an earlier submit must not be taken over") + assert.NoError(t, mock.ExpectationsWereMet()) +} + +// TestPGXMock_ClaimRIExchangeIdempotencyKey_ErrorIsNotAWonClaim: a failed Exec +// must surface as an error, never as a claim the caller can act on. +func TestPGXMock_ClaimRIExchangeIdempotencyKey_ErrorIsNotAWonClaim(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + + mock.ExpectExec("INSERT INTO ri_exchange_idempotency"). + WithArgs("fingerprint-1", 900.0). + WillReturnError(errors.New("connection refused")) + + claimed, err := store.ClaimRIExchangeIdempotencyKey(context.Background(), "fingerprint-1", 15*time.Minute) + require.Error(t, err) + assert.False(t, claimed) + assert.Contains(t, err.Error(), "connection refused") + assert.NoError(t, mock.ExpectationsWereMet()) +} + +// TestPGXMock_ClaimRIExchangeIdempotencyKey_RejectsUnusableArguments pins the +// fail-loud guards. An empty key would collapse every distinct exchange onto +// one row, and a non-positive window would make make_interval(secs => 0) treat +// every existing claim as expired -- both turn the guard into a no-op that +// still reports success, which is worse than no guard at all. +func TestPGXMock_ClaimRIExchangeIdempotencyKey_RejectsUnusableArguments(t *testing.T) { + cases := map[string]struct { + key string + window time.Duration + want string + }{ + "empty key": {"", 15 * time.Minute, "empty RI exchange idempotency key"}, + "zero window": {"fingerprint-1", 0, "window must be positive"}, + "negative window": {"fingerprint-1", -time.Second, "window must be positive"}, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + // No ExpectExec: the guard must refuse before touching the database. + + claimed, err := store.ClaimRIExchangeIdempotencyKey(context.Background(), tc.key, tc.window) + require.Error(t, err) + assert.False(t, claimed) + assert.Contains(t, err.Error(), tc.want) + assert.NoError(t, mock.ExpectationsWereMet()) + }) + } +} diff --git a/internal/database/postgres/migrations/000097_ri_exchange_idempotency.down.sql b/internal/database/postgres/migrations/000097_ri_exchange_idempotency.down.sql new file mode 100644 index 000000000..5793cb6c2 --- /dev/null +++ b/internal/database/postgres/migrations/000097_ri_exchange_idempotency.down.sql @@ -0,0 +1,2 @@ +-- Reverse 000097: drop the RI exchange submit-time idempotency ledger. +DROP TABLE IF EXISTS ri_exchange_idempotency; diff --git a/internal/database/postgres/migrations/000097_ri_exchange_idempotency.up.sql b/internal/database/postgres/migrations/000097_ri_exchange_idempotency.up.sql new file mode 100644 index 000000000..f6a0d5587 --- /dev/null +++ b/internal/database/postgres/migrations/000097_ri_exchange_idempotency.up.sql @@ -0,0 +1,30 @@ +-- Submit-time idempotency for the RI exchange execute endpoints (issue #1642). +-- +-- POST /api/ri-exchange/azure-instances/exchange and its AWS sibling commit an +-- irreversible exchange with no request-level dedupe. The Azure handler mints a +-- FRESH CalculateExchange session on every request (deliberately, so a stale or +-- client-supplied session can never bypass the guardrails), which means Azure's +-- own session-level replay protection never fires across two POSTs of the same +-- logical exchange. AWS's AcceptReservedInstancesExchangeQuote has no +-- ClientToken at all. A client that times out while the long-running operation +-- is still polling and retries therefore commits the exchange twice, spending +-- roughly twice the intended amount while each half stays under the cap. +-- +-- This table is the claim ledger. The handler derives a fingerprint of WHAT the +-- request buys (account scope + sources + targets; see the exchangeIdempotency* +-- helpers in internal/api/handler_ri_exchange.go) and claims it in a single +-- atomic statement immediately before the commit call. Exactly one claimant wins +-- inside the window; the losers get a 409. +-- +-- Rows are not garbage-collected on a schedule: a re-claim of the same +-- fingerprint after the window expires overwrites claimed_at in place, so the +-- table's size is bounded by the number of DISTINCT exchange shapes ever +-- submitted, which is tiny. +CREATE TABLE IF NOT EXISTS ri_exchange_idempotency ( + -- sha256 hex of the request fingerprint. Opaque to the database. + idempotency_key TEXT PRIMARY KEY, + -- When the current holder claimed it. Always written from the DATABASE + -- clock (now()), never the application's, so skew between concurrent + -- application instances cannot widen or shrink the window. + claimed_at TIMESTAMPTZ NOT NULL DEFAULT now() +); diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 1b75dd8e1..2129e9364 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -533,6 +533,20 @@ func (m *MockConfigStore) ClaimMarketplaceListingSlot(ctx context.Context, purch return args.Bool(0), args.Error(1) } +// ClaimRIExchangeIdempotencyKey mocks the atomic RI exchange submit claim +// (issue #1642). With no registered expectation it defaults to "this call won +// the claim" so the many pre-existing execute-path tests, whose subject is a +// gate upstream of the claim, keep exercising the commit call. Tests whose +// subject IS the claim register an explicit expectation and assert on it. +func (m *MockConfigStore) ClaimRIExchangeIdempotencyKey(ctx context.Context, key string, window time.Duration) (bool, error) { + m.record("ClaimRIExchangeIdempotencyKey", ctx, key, window) + if !isExpected(&m.Mock, "ClaimRIExchangeIdempotencyKey") { + return true, nil + } + args := m.Called(ctx, key, window) + return args.Bool(0), args.Error(1) +} + func (m *MockConfigStore) SaveRIExchangeRecord(ctx context.Context, record *config.RIExchangeRecord) error { m.record("SaveRIExchangeRecord", ctx, record) args := m.Called(ctx, record) diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 36ac77414..f1ee2611f 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -386,6 +386,10 @@ func (m *mockConfigStoreForHealth) ClaimMarketplaceListingSlot(_ context.Context return true, nil } +func (m *mockConfigStoreForHealth) ClaimRIExchangeIdempotencyKey(_ context.Context, _ string, _ time.Duration) (bool, error) { + return true, nil +} + func (m *mockConfigStoreForHealth) StampOfferingClass(_ context.Context, _, _ string) error { return nil }