From 532712ee8bcbaec32ffe2bed0a3f8e78286d203d Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 13 Aug 2026 19:16:27 +0200 Subject: [PATCH 1/4] fix(api): submit-time idempotency for RI exchange execute (#1642) Both RI exchange execute endpoints commit an irreversible exchange with no request-level dedupe, so a client that times out while the provider call is still running and retries spends roughly twice the intended amount, with each half individually under MaxPurchaseAmount. Azure's own replay protection cannot cover this: executeAzureExchange mints a FRESH CalculateExchange session on every request, deliberately, so that a stale or client-supplied session can never bypass the guardrails. Two POSTs of one logical exchange therefore carry two different session IDs. On AWS, AcceptReservedInstancesExchangeQuote carries no ClientToken at all. Each handler now claims a fingerprint of what the request buys immediately before the commit, in one atomic INSERT ... ON CONFLICT DO UPDATE ... WHERE against a new ri_exchange_idempotency ledger. Exactly one claimant wins inside the window; the losers get a 409 pointing at the in-flight submit. The fingerprint folds in the account scope (subscription_id on Azure, per the #1495 precedent that its reservation listing is tenant-wide; cloud account plus region on AWS) and every source and target with its quantity. It deliberately excludes the spend cap, the currency and the derived billing scope: none of them change what is bought, so including them would let a retry that merely raises the cap mint a fresh key and commit a second time. The claim is taken after every authorization gate and money guardrail, so no rejected request leaves a claim behind, and it is not released when the commit fails: past submission the outcome is ambiguous, and holding it is the fail-closed choice. The Azure failure message now says so rather than reporting a flat failure that invites the very retry the claim refuses. Three input checks move ahead of the claim so that a request which can never commit cannot hold one, and so that no fingerprint component can be blank: per-id non-emptiness of ri_ids, targets[].count >= 1, and target_count >= 1 on the legacy singleton spelling. pkg/exchange already enforced the two count rules, but only once ExecuteExchange was running, which is after the claim and surfaced as an opaque 500. targets[].sku is likewise now rejected when blank-but-not-empty, matching location: the fingerprint trims surrounding whitespace, so a whitespace-only SKU would otherwise reach it empty. Closes #1642 --- internal/analytics/collector_test.go | 4 + internal/api/handler_ri_exchange.go | 75 +- internal/api/handler_ri_exchange_test.go | 8 +- internal/api/openapi.yaml | 25 + internal/api/ri_exchange_idempotency.go | 188 +++++ internal/api/ri_exchange_idempotency_test.go | 669 ++++++++++++++++++ internal/config/interfaces.go | 17 + internal/config/store_postgres.go | 34 + internal/config/store_postgres_db_test.go | 108 +++ ...e_postgres_ri_exchange_idempotency_test.go | 101 +++ .../000097_ri_exchange_idempotency.down.sql | 2 + .../000097_ri_exchange_idempotency.up.sql | 30 + internal/mocks/stores.go | 14 + internal/server/test_helpers_test.go | 4 + 14 files changed, 1269 insertions(+), 10 deletions(-) create mode 100644 internal/api/ri_exchange_idempotency.go create mode 100644 internal/api/ri_exchange_idempotency_test.go create mode 100644 internal/config/store_postgres_ri_exchange_idempotency_test.go create mode 100644 internal/database/postgres/migrations/000097_ri_exchange_idempotency.down.sql create mode 100644 internal/database/postgres/migrations/000097_ri_exchange_idempotency.up.sql 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..80ae91a0e 100644 --- a/internal/api/openapi.yaml +++ b/internal/api/openapi.yaml @@ -775,6 +775,12 @@ 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. parameters: - $ref: '#/components/parameters/CSRFToken' requestBody: @@ -820,6 +826,8 @@ paths: $ref: '#/components/responses/Forbidden' '404': $ref: '#/components/responses/NotFound' + '409': + $ref: '#/components/responses/Conflict' '422': $ref: '#/components/responses/UnprocessableEntity' @@ -917,6 +925,12 @@ 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. parameters: - $ref: '#/components/parameters/CSRFToken' requestBody: @@ -951,6 +965,8 @@ paths: $ref: '#/components/responses/Unauthorized' '403': $ref: '#/components/responses/Forbidden' + '409': + $ref: '#/components/responses/Conflict' /api/ri-exchange/config: get: @@ -1881,6 +1897,15 @@ components: application/json: schema: $ref: '#/components/schemas/Error' + Conflict: + description: > + An identical submit is already in flight or recently committed. The + request was NOT executed a second time; check the outcome of the + earlier one 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..374fcac42 --- /dev/null +++ b/internal/api/ri_exchange_idempotency.go @@ -0,0 +1,188 @@ +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 the claim is deliberately NOT released when the +// commit itself fails, because past that point the provider may already have +// accepted the exchange; holding the claim for the rest of the window is the +// fail-closed choice. +// +// 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 was submitted within the last %s and may still be settling; "+ + "check the exchange's outcome before resubmitting rather than committing it twice", + riExchangeIdempotencyWindow)) + } + return nil +} diff --git a/internal/api/ri_exchange_idempotency_test.go b/internal/api/ri_exchange_idempotency_test.go new file mode 100644 index 000000000..5d9c5a6ff --- /dev/null +++ b/internal/api/ri_exchange_idempotency_test.go @@ -0,0 +1,669 @@ +package api + +import ( + "context" + "fmt" + "strings" + "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 was submitted") + + 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") +} + +// --- 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. +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") + }) + + 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") + }) +} + +// --- 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") +} diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 5af8b3b6d..4c98a620d 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -265,6 +265,23 @@ 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. + // + // The claim is deliberately NOT released when the provider call fails: + // past the point of submission the outcome is ambiguous, and holding the + // claim for the rest of the window is the fail-closed choice. + 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 } From bd50d2a9a24ed67b3a4905d65819ba2393e32ab3 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 13 Aug 2026 20:34:52 +0200 Subject: [PATCH 2/4] docs(api): a 409 from RI exchange execute asserts no outcome (#1642) The shared Conflict response said an identical submit was "already in flight or recently committed". Both of those are resolved states, and the claim is retained unconditionally once taken -- including when the provider call failed -- so a duplicate can receive a 409 when the earlier submit has no confirmed commit at all. A client reading that as "the earlier one succeeded" skips verification and assumes a purchase that may never have happened. The claim genuinely is unconditional, and deliberately so: past the commit call the provider may already have accepted the exchange, and releasing the claim on the errors the handler believes to be definite rejections would be guessing about the one thing it cannot observe. So the contract now says what it actually offers, in the same terms everywhere it is stated: a 409 means an identical submit claimed the window, and that submit may be running, may have committed, may have failed after reaching the provider, or may have failed without committing anything. Verify before resubmitting. Corrected in all four places it appears rather than only the one the review anchored on: the Conflict response, both execute endpoint descriptions (which pointed at the 409 without qualifying it), the runtime 409 message, and the StoreInterface doc for ClaimRIExchangeIdempotencyKey. The PR description carried the same claim and was corrected too. TestClaimExchangeSubmit_Outcomes now pins the message against overclaiming: it must name all three possible outcomes and tell the caller to verify. Mutation-verified by restoring the previous wording, which the test catches by assertion. --- internal/api/openapi.yaml | 18 ++++++++++++----- internal/api/ri_exchange_idempotency.go | 21 ++++++++++++++------ internal/api/ri_exchange_idempotency_test.go | 17 +++++++++++++++- internal/config/interfaces.go | 9 ++++++--- 4 files changed, 50 insertions(+), 15 deletions(-) diff --git a/internal/api/openapi.yaml b/internal/api/openapi.yaml index 80ae91a0e..ab1b5aa40 100644 --- a/internal/api/openapi.yaml +++ b/internal/api/openapi.yaml @@ -780,7 +780,8 @@ paths: 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. + 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: @@ -930,7 +931,8 @@ paths: 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. + 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: @@ -1899,9 +1901,15 @@ components: $ref: '#/components/schemas/Error' Conflict: description: > - An identical submit is already in flight or recently committed. The - request was NOT executed a second time; check the outcome of the - earlier one before resubmitting. + 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: diff --git a/internal/api/ri_exchange_idempotency.go b/internal/api/ri_exchange_idempotency.go index 374fcac42..307ef3627 100644 --- a/internal/api/ri_exchange_idempotency.go +++ b/internal/api/ri_exchange_idempotency.go @@ -166,10 +166,18 @@ func awsExchangeIdempotencyKey(cloudAccountID string, body ExchangeExecuteReques // // 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 the claim is deliberately NOT released when the -// commit itself fails, because past that point the provider may already have -// accepted the exchange; holding the claim for the rest of the window is the -// fail-closed choice. +// 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. @@ -180,8 +188,9 @@ func (h *Handler) claimExchangeSubmit(ctx context.Context, key string) error { } if !claimed { return NewClientError(409, fmt.Sprintf( - "an identical RI exchange was submitted within the last %s and may still be settling; "+ - "check the exchange's outcome before resubmitting rather than committing it twice", + "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_test.go b/internal/api/ri_exchange_idempotency_test.go index 5d9c5a6ff..20ff00e39 100644 --- a/internal/api/ri_exchange_idempotency_test.go +++ b/internal/api/ri_exchange_idempotency_test.go @@ -170,7 +170,12 @@ func TestExecuteAzureExchange_RetryOfInFlightSubmitDoesNotDoubleSpend(t *testing 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 was submitted") + 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") @@ -555,6 +560,16 @@ func TestClaimExchangeSubmit_Outcomes(t *testing.T) { 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) { diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 4c98a620d..b20d06c25 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -277,9 +277,12 @@ type StoreInterface interface { // A claim older than window is reclaimable, so a genuine intentional // repeat of the same exchange is not blocked forever. // - // The claim is deliberately NOT released when the provider call fails: - // past the point of submission the outcome is ambiguous, and holding the - // claim for the rest of the window is the fail-closed choice. + // 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 From c273e0739d34cb2716bf447a35fa920231542f86 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 13 Aug 2026 20:49:46 +0200 Subject: [PATCH 3/4] test(api): split the RI exchange idempotency tests under the file-size limit (#1642) ri_exchange_idempotency_test.go had grown to 684 lines, over the 500-line limit in CLAUDE.md. The sibling handler test files in this package are far worse (2013 and 2225 lines), but pre-existing debt does not license adding more, so the new file is split along the seams it already had: ri_exchange_idempotency_test.go the exchangeClaimLedger harness, the handler double-spend tests and the claim-outcome contract ri_exchange_idempotency_key_test.go what the fingerprint is made of ri_exchange_idempotency_guards_test.go the input checks that must refuse a request before it can claim A pure relocation: comparing the sorted lines of the old file against the three new ones shows only additions, all of them the new file headers. No test body, assertion or expectation changed, and the set of declared functions is identical. --- .../ri_exchange_idempotency_guards_test.go | 114 ++++++ .../api/ri_exchange_idempotency_key_test.go | 287 ++++++++++++++ internal/api/ri_exchange_idempotency_test.go | 372 ------------------ 3 files changed, 401 insertions(+), 372 deletions(-) create mode 100644 internal/api/ri_exchange_idempotency_guards_test.go create mode 100644 internal/api/ri_exchange_idempotency_key_test.go 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..960286ecf --- /dev/null +++ b/internal/api/ri_exchange_idempotency_guards_test.go @@ -0,0 +1,114 @@ +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") +} 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 index 20ff00e39..55364a342 100644 --- a/internal/api/ri_exchange_idempotency_test.go +++ b/internal/api/ri_exchange_idempotency_test.go @@ -3,7 +3,6 @@ package api import ( "context" "fmt" - "strings" "sync" "testing" "time" @@ -265,279 +264,6 @@ func TestExecuteExchange_DuplicateSubmitRefusedBeforeAWS(t *testing.T) { "the legacy and targets[] spellings of one purchase must fingerprint alike") } -// --- 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. func TestClaimExchangeSubmit_Outcomes(t *testing.T) { ctx := context.Background() @@ -584,101 +310,3 @@ func TestClaimExchangeSubmit_Outcomes(t *testing.T) { assert.Contains(t, err.Error(), "connection refused") }) } - -// --- 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") -} From f0ddfef076c5a2ba9334619f48dea07927764eb9 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 13 Aug 2026 23:29:22 +0200 Subject: [PATCH 4/4] test(api): assert an invalid exchange body never reaches the constraint check (#1642) --- internal/api/ri_exchange_idempotency_guards_test.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/internal/api/ri_exchange_idempotency_guards_test.go b/internal/api/ri_exchange_idempotency_guards_test.go index 960286ecf..8918a5dac 100644 --- a/internal/api/ri_exchange_idempotency_guards_test.go +++ b/internal/api/ri_exchange_idempotency_guards_test.go @@ -111,4 +111,6 @@ func TestExecuteExchange_InvalidBodyNeverClaims(t *testing.T) { 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) }