From fbdbde6bc0145fa11301fce79ab8192b64def4c9 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 6 Oct 2026 21:16:27 +0200 Subject: [PATCH 1/2] fix(marketplace): persist the listing ClientToken before the AWS call The ClientToken was derived from the request, so a retry across a month boundary (the default price schedule depends on the remaining months) produced a new token and AWS silently created a second listing. A request that read the row before another one claimed it could also reuse the token of a canceled listing. Persist a random token and the price schedule in the UPDATE that claims the slot, before the AWS call (migration 000105). A retry resends the stored token and schedule. A pending row that holds a token can be claimed again; the retry first asks AWS (DescribeReservedInstancesListings) whether the earlier attempt created the listing and records it instead of creating another. Recording a listing, or a definitive AWS rejection, drops the token; an ambiguous error keeps it. Refs #525 --- internal/analytics/collector_test.go | 8 +- internal/api/handler.go | 4 + internal/api/handler_marketplace.go | 141 +----- internal/api/handler_marketplace_attempt.go | 249 ++++++++++ .../api/handler_marketplace_attempt_test.go | 438 ++++++++++++++++++ .../handler_marketplace_credentials_test.go | 5 +- internal/api/handler_marketplace_test.go | 186 ++------ internal/api/handler_purchase_action_test.go | 2 +- internal/api/nullable_purchase_cost_test.go | 4 +- internal/config/interfaces.go | 27 +- internal/config/store_postgres.go | 63 ++- ...gres_marketplace_claim_integration_test.go | 113 +++++ internal/config/types.go | 18 + ..._purchase_history_listing_attempt.down.sql | 4 + ...05_purchase_history_listing_attempt.up.sql | 17 + ...5_purchase_history_listing_attempt_test.go | 53 +++ internal/mocks/stores.go | 16 +- internal/server/test_helpers_test.go | 8 +- 18 files changed, 1026 insertions(+), 330 deletions(-) create mode 100644 internal/api/handler_marketplace_attempt.go create mode 100644 internal/api/handler_marketplace_attempt_test.go create mode 100644 internal/config/store_postgres_marketplace_claim_integration_test.go create mode 100644 internal/database/postgres/migrations/000105_purchase_history_listing_attempt.down.sql create mode 100644 internal/database/postgres/migrations/000105_purchase_history_listing_attempt.up.sql create mode 100644 internal/database/postgres/migrations/000105_purchase_history_listing_attempt_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 6870feb9e..da30c67ca 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -475,8 +475,12 @@ func (m *mockConfigStore) UpdatePurchaseHistoryListing(_ context.Context, _, _, return nil } -func (m *mockConfigStore) ClaimMarketplaceListingSlot(_ context.Context, _ string) (bool, error) { - return true, nil +func (m *mockConfigStore) ClaimMarketplaceListingSlot(_ context.Context, _, _ string, _ []byte) (*config.MarketplaceListingClaim, error) { + return &config.MarketplaceListingClaim{}, nil +} + +func (m *mockConfigStore) ReleaseMarketplaceListingClaim(_ context.Context, _, _ string, _ bool) error { + return nil } func (m *mockConfigStore) ClaimRIExchangeIdempotencyKey(_ context.Context, _ string, _ time.Duration) (bool, error) { diff --git a/internal/api/handler.go b/internal/api/handler.go index cb4ad18c1..aca78168e 100644 --- a/internal/api/handler.go +++ b/internal/api/handler.go @@ -91,6 +91,10 @@ type Handler struct { // awsprovider.NewEC2ClientDirect. marketplaceEC2Factory func(aws.Config) marketplaceEC2Client + // Optional listing-finder injected by tests. When nil, the marketplace + // reconcile uses the AWS SDK directly (see awsMarketplaceListingFinder). + marketplaceFinderFactory func(aws.Config) marketplaceListingFinder + // Optional account-resolver injection point used by the reshape // handler integration test. When nil (the production default), the // handler calls h.resolveAWSCloudAccountID which in turn invokes diff --git a/internal/api/handler_marketplace.go b/internal/api/handler_marketplace.go index ea802669e..8a2fd7905 100644 --- a/internal/api/handler_marketplace.go +++ b/internal/api/handler_marketplace.go @@ -32,7 +32,6 @@ import ( ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types" "github.com/aws/aws-sdk-go-v2/service/sts" smithy "github.com/aws/smithy-go" - "github.com/google/uuid" ) // marketplaceEC2Client is the narrow EC2 interface the marketplace handlers @@ -226,15 +225,7 @@ func (h *Handler) marketplaceList(ctx context.Context, req *events.LambdaFunctio return nil, NewClientError(400, "only Standard Reserved Instances can be listed on the AWS Marketplace; this purchase has offering_class="+offeringClass) } - awsSchedule := make([]ec2svc.MarketplacePriceTier, 0, len(schedule)) - for _, t := range schedule { - awsSchedule = append(awsSchedule, ec2svc.MarketplacePriceTier{ - Term: t.TermMonths, - Price: t.Price, - }) - } - - result, err := h.reserveAndCreateListing(ctx, purchaseID, row, ec2Client, awsSchedule) + result, usedSchedule, err := h.reserveAndCreateListing(ctx, purchaseID, row, cfg, ec2Client, schedule) if err != nil { return nil, err } @@ -242,7 +233,7 @@ func (h *Handler) marketplaceList(ctx context.Context, req *events.LambdaFunctio return &MarketplaceListResponse{ ListingID: result.ListingID, ListingState: result.State, - PriceSchedule: schedule, + PriceSchedule: usedSchedule, AWSFeePercent: awsMarketplaceFeePercent, Note: fmt.Sprintf("AWS charges a %d%% transaction fee on the listing proceeds. Net proceeds = ListingPrice * %.2f.", awsMarketplaceFeePercent, awsMarketplaceNetFactor), }, nil @@ -252,122 +243,6 @@ func (h *Handler) marketplaceList(ctx context.Context, req *events.LambdaFunctio // run after a created listing failed to persist. const marketplaceCompensationTimeout = 10 * time.Second -// reserveAndCreateListing atomically claims the marketplace-listing slot for the -// row, creates the AWS listing, and persists it, releasing the claim on every -// failure path that leaves no live listing so a failed attempt does not leave -// the row stuck in the transient pending state. Extracted from marketplaceList -// to keep that handler under the gocyclo budget. Returns the persisted listing -// result on success. -func (h *Handler) reserveAndCreateListing(ctx context.Context, purchaseID string, row *config.PurchaseHistoryRecord, ec2Client marketplaceEC2Client, awsSchedule []ec2svc.MarketplacePriceTier) (ec2svc.MarketplaceListingResult, error) { - // List every RI in the row: a row of N Standard RIs must list all N, not a - // single unit (issue #292 multi-count fix). Floor at 1 for legacy rows that - // somehow recorded a non-positive count so a valid Standard RI still lists, - // and reject an implausibly large count rather than silently truncating it - // into int32 (which could list the wrong number of RIs on the money path). - instanceCount := int32(1) - switch { - case row.Count > math.MaxInt32: - return ec2svc.MarketplaceListingResult{}, NewClientError(500, fmt.Sprintf("purchase count %d exceeds the marketplace listing limit", row.Count)) - case row.Count > 1: - instanceCount = int32(row.Count) - } - - // Atomically reserve the listing slot before calling AWS. The read-only - // listing_state check in validateMarketplaceListRequest is not enough on its - // own: two concurrent requests can both pass it, and AWS assigns a fresh - // ClientToken per call so the provider will not dedup them, leaving two live - // listings for one RI. The claim serializes concurrent creates for the same - // purchase_id (issue #292 CR concurrent-create guard). - claimed, err := h.config.ClaimMarketplaceListingSlot(ctx, purchaseID) - if err != nil { - return ec2svc.MarketplaceListingResult{}, fmt.Errorf("failed to reserve marketplace listing slot: %w", err) - } - if !claimed { - return ec2svc.MarketplaceListingResult{}, NewClientError(409, "a marketplace listing is already active or in progress for this RI; cancel it first") - } - - result, err := ec2Client.CreateMarketplaceListing(ctx, ec2svc.MarketplaceListingRequest{ - ReservedInstancesID: purchaseID, - ClientToken: marketplaceClientToken(purchaseID, row.ListingID, instanceCount, awsSchedule), - PriceSchedule: awsSchedule, - InstanceCount: instanceCount, - }) - if err != nil { - // AWS created nothing: release the claim so a retry is not blocked. - h.releaseMarketplaceClaim(ctx, purchaseID, row) - logging.Warnf("marketplace: CreateReservedInstancesListing for purchase %s failed: %v", purchaseID, err) - return ec2svc.MarketplaceListingResult{}, mapAWSMarketplaceError("AWS marketplace listing failed", err) - } - - if isDeadListingState(result.State) { - return ec2svc.MarketplaceListingResult{}, h.recordDeadListing(ctx, purchaseID, row, result) - } - - // Persist the listing ID and state. On DB failure, attempt a compensating - // rollback (cancel the just-created listing) to avoid a desync where the - // user sees success but the listing is invisible in subsequent renders, then - // release the claim so the row does not stay stuck in the pending state. - if dbErr := h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, result.ListingID, result.State); dbErr != nil { - logging.Errorf("marketplace: listing created (%s / %s) but DB update failed: %v; attempting rollback", result.ListingID, result.State, dbErr) - compCtx, cancelComp := context.WithTimeout(context.WithoutCancel(ctx), marketplaceCompensationTimeout) - defer cancelComp() - if _, rollbackErr := ec2Client.CancelMarketplaceListing(compCtx, result.ListingID); rollbackErr != nil { - return ec2svc.MarketplaceListingResult{}, h.keepUncanceledListing(compCtx, purchaseID, result, rollbackErr) - } - logging.Warnf("marketplace: listing %s rolled back (canceled) after DB failure", result.ListingID) - h.recordRolledBackListing(compCtx, ctx, purchaseID, row, result.ListingID) - return ec2svc.MarketplaceListingResult{}, fmt.Errorf("listing created but could not be persisted; listing has been rolled back: %w", dbErr) - } - - return result, nil -} - -// marketplaceClientToken derives the CreateReservedInstancesListing ClientToken -// from the request instead of drawing a random one, so a retry of the same -// attempt (for example after a timeout that hid AWS's success) carries the same -// token and AWS returns the existing listing rather than creating a second one. -// The previously recorded listing id is part of the input: every path that -// ends a listing (cancel, rollback, dead state) records that listing's id, so -// the next attempt gets a new token instead of AWS replaying the dead listing. -// The default schedule depends on the remaining months, so a retry across a -// month boundary also gets a new token; persisting the token (issue #525) -// closes that. The count and schedule are included so a retry -// with different parameters never trips AWS's idempotent-parameter-mismatch -// error. -func marketplaceClientToken(purchaseID, priorListingID string, count int32, schedule []ec2svc.MarketplacePriceTier) string { - var b strings.Builder - fmt.Fprintf(&b, "%s|%s|%d", purchaseID, priorListingID, count) - for _, t := range schedule { - fmt.Fprintf(&b, "|%d:%v", t.Term, t.Price) - } - return uuid.NewSHA1(uuid.NameSpaceOID, []byte(b.String())).String() -} - -func isDeadListingState(state string) bool { - return strings.EqualFold(state, config.ListingStateCancelled) || strings.EqualFold(state, config.ListingStateClosed) -} - -// recordDeadListing handles a create response in a canceled or closed state -// (AWS replays a dead listing for a reused ClientToken): it records the listing -// so the slot is free with a new token for the next attempt, and reports it. -func (h *Handler) recordDeadListing(ctx context.Context, purchaseID string, row *config.PurchaseHistoryRecord, listing ec2svc.MarketplaceListingResult) error { - if err := h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, listing.ListingID, listing.State); err != nil { - logging.Errorf("marketplace: failed to record dead listing %s for purchase %s: %v", listing.ListingID, purchaseID, err) - h.releaseMarketplaceClaim(ctx, purchaseID, row) - } - return NewClientError(502, fmt.Sprintf("AWS returned listing %s in state %s instead of a live listing; retry to create a new one", listing.ListingID, listing.State)) -} - -// recordRolledBackListing records the canceled listing instead of restoring the -// old listing id, so the next attempt derives a different ClientToken; AWS -// would otherwise replay the canceled listing for the same token. -func (h *Handler) recordRolledBackListing(compCtx, ctx context.Context, purchaseID string, row *config.PurchaseHistoryRecord, listingID string) { - if err := h.config.UpdatePurchaseHistoryListing(compCtx, purchaseID, listingID, config.ListingStateCancelled); err != nil { - logging.Errorf("marketplace: failed to record rolled-back listing %s for purchase %s: %v", listingID, purchaseID, err) - h.releaseMarketplaceClaim(ctx, purchaseID, row) - } -} - // keepUncanceledListing records a listing that could be neither persisted nor // canceled; releasing the claim instead would hide it from the cancel endpoint. func (h *Handler) keepUncanceledListing(ctx context.Context, purchaseID string, listing ec2svc.MarketplaceListingResult, cancelErr error) error { @@ -414,18 +289,6 @@ func (h *Handler) populateOfferingClass(ctx context.Context, purchaseID string, return class, nil } -// releaseMarketplaceClaim restores a purchase_history row's listing fields to -// the state captured before ClaimMarketplaceListingSlot reserved the slot. It -// runs on every failure path after a successful claim so a failed listing -// attempt does not leave the row stuck in the transient pending state (which -// would block future list attempts). Best-effort: on error it logs loudly -// because the row may stay pending until the #292 status poller reconciles it. -func (h *Handler) releaseMarketplaceClaim(ctx context.Context, purchaseID string, row *config.PurchaseHistoryRecord) { - if err := h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, row.ListingID, row.ListingState); err != nil { - logging.Errorf("marketplace: failed to release listing claim for purchase %s (row may be stuck in %q): %v", purchaseID, config.ListingStatePending, err) - } -} - // marketplaceCancel handles POST /api/purchases/{id}/marketplace-cancel. func (h *Handler) marketplaceCancel(ctx context.Context, req *events.LambdaFunctionURLRequest, purchaseID string) (any, error) { session, action, err := h.requireSessionPurchaseAction(ctx, req, auth.ActionSellAny, auth.ActionSellOwn) diff --git a/internal/api/handler_marketplace_attempt.go b/internal/api/handler_marketplace_attempt.go new file mode 100644 index 000000000..352a59030 --- /dev/null +++ b/internal/api/handler_marketplace_attempt.go @@ -0,0 +1,249 @@ +package api + +// handler_marketplace_attempt.go holds the create-listing attempt of the +// marketplace-list flow (issue #525). The attempt's ClientToken and price +// schedule are persisted in the statement that claims the listing slot, before +// the AWS call. A retry after an ambiguous AWS error or a crash resends the +// stored token and schedule, so AWS returns the existing listing instead of +// creating a second one, and first asks AWS whether the earlier attempt +// already created the listing (reconcile) and records it if so. + +import ( + "context" + "encoding/json" + "fmt" + "math" + "strings" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + ec2svc "github.com/LeanerCloud/cloud-commitments-go/providers/aws/services/ec2" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/aws/aws-sdk-go-v2/aws" + sdkec2 "github.com/aws/aws-sdk-go-v2/service/ec2" + "github.com/google/uuid" +) + +// marketplaceListingFinder looks up a listing created by an earlier attempt +// through DescribeReservedInstancesListings. The ec2 service wrapper has no +// lookup by ClientToken, so this is a separate narrow interface. +type marketplaceListingFinder interface { + FindMarketplaceListingByToken(ctx context.Context, reservedInstancesID, clientToken string) (ec2svc.MarketplaceListingResult, bool, error) +} + +type describeListingsAPI interface { + DescribeReservedInstancesListings(ctx context.Context, in *sdkec2.DescribeReservedInstancesListingsInput, opts ...func(*sdkec2.Options)) (*sdkec2.DescribeReservedInstancesListingsOutput, error) +} + +type awsMarketplaceListingFinder struct{ api describeListingsAPI } + +func (f awsMarketplaceListingFinder) FindMarketplaceListingByToken(ctx context.Context, reservedInstancesID, clientToken string) (ec2svc.MarketplaceListingResult, bool, error) { + out, err := f.api.DescribeReservedInstancesListings(ctx, &sdkec2.DescribeReservedInstancesListingsInput{ + ReservedInstancesId: aws.String(reservedInstancesID), + }) + if err != nil { + return ec2svc.MarketplaceListingResult{}, false, fmt.Errorf("DescribeReservedInstancesListings failed: %w", err) + } + for _, l := range out.ReservedInstancesListings { + if aws.ToString(l.ClientToken) == clientToken { + return ec2svc.MarketplaceListingResult{ListingID: aws.ToString(l.ReservedInstancesListingId), State: string(l.Status)}, true, nil + } + } + return ec2svc.MarketplaceListingResult{}, false, nil +} + +func (h *Handler) buildMarketplaceFinder(cfg aws.Config) marketplaceListingFinder { + if h.marketplaceFinderFactory != nil { + return h.marketplaceFinderFactory(cfg) + } + return awsMarketplaceListingFinder{api: sdkec2.NewFromConfig(cfg)} +} + +// marketplaceInstanceCount returns how many RIs of the row to list. A row of N +// Standard RIs must list all N, not a single unit (issue #292 multi-count fix). +// It floors at 1 for legacy rows that recorded a non-positive count so a valid +// Standard RI still lists, and rejects an implausibly large count rather than +// silently truncating it into int32 (which could list the wrong number of RIs +// on the money path). +func marketplaceInstanceCount(row *config.PurchaseHistoryRecord) (int32, error) { + switch { + case row.Count > math.MaxInt32: + return 0, NewClientError(500, fmt.Sprintf("purchase count %d exceeds the marketplace listing limit", row.Count)) + case row.Count > 1: + return int32(row.Count), nil + } + return 1, nil +} + +func toAWSPriceSchedule(schedule []MarketplacePriceTier) []ec2svc.MarketplacePriceTier { + out := make([]ec2svc.MarketplacePriceTier, 0, len(schedule)) + for _, t := range schedule { + out = append(out, ec2svc.MarketplacePriceTier{Term: t.TermMonths, Price: t.Price}) + } + return out +} + +// reserveAndCreateListing atomically claims the marketplace-listing slot for the +// row (persisting the attempt's token and schedule), reconciles an earlier +// unresolved attempt, creates the AWS listing, and persists it, releasing the +// claim on every failure path that leaves no live listing so a failed attempt +// does not leave the row stuck in the transient pending state. Returns the +// persisted listing and the schedule it was created with, which on a retry is +// the stored one, not the one passed in. +func (h *Handler) reserveAndCreateListing(ctx context.Context, purchaseID string, row *config.PurchaseHistoryRecord, cfg aws.Config, ec2Client marketplaceEC2Client, schedule []MarketplacePriceTier) (ec2svc.MarketplaceListingResult, []MarketplacePriceTier, error) { + instanceCount, err := marketplaceInstanceCount(row) + if err != nil { + return ec2svc.MarketplaceListingResult{}, nil, err + } + claim, used, err := h.claimListingSlot(ctx, purchaseID, schedule) + if err != nil { + return ec2svc.MarketplaceListingResult{}, nil, err + } + if claim.Resumed { + result, found, recErr := h.reconcileAttempt(ctx, purchaseID, claim, h.buildMarketplaceFinder(cfg)) + if recErr != nil || found { + return result, used, recErr + } + } + result, err := h.createAndPersistListing(ctx, purchaseID, claim, ec2Client, instanceCount, toAWSPriceSchedule(used)) + return result, used, err +} + +// claimListingSlot claims the row with a fresh random token. The token is +// persisted by the claim itself and only the persisted copy is used, so it +// never depends on a row read before the claim or on the clock. +func (h *Handler) claimListingSlot(ctx context.Context, purchaseID string, schedule []MarketplacePriceTier) (*config.MarketplaceListingClaim, []MarketplacePriceTier, error) { + encoded, err := json.Marshal(schedule) + if err != nil { + return nil, nil, fmt.Errorf("failed to encode marketplace price schedule: %w", err) + } + claim, err := h.config.ClaimMarketplaceListingSlot(ctx, purchaseID, uuid.NewString(), encoded) + if err != nil { + return nil, nil, fmt.Errorf("failed to reserve marketplace listing slot: %w", err) + } + if claim == nil { + return nil, nil, NewClientError(409, "a marketplace listing is already active or in progress for this RI; cancel it first") + } + if !claim.Resumed { + return claim, schedule, nil + } + var stored []MarketplacePriceTier + if err = json.Unmarshal(claim.PriceSchedule, &stored); err != nil { + h.releaseMarketplaceClaim(ctx, purchaseID, claim, true) + return nil, nil, fmt.Errorf("stored marketplace price schedule for purchase %s is unusable: %w", purchaseID, err) + } + if len(stored) == 0 { + h.releaseMarketplaceClaim(ctx, purchaseID, claim, true) + return nil, nil, fmt.Errorf("stored marketplace price schedule for purchase %s is empty", purchaseID) + } + return claim, stored, nil +} + +// reconcileAttempt asks AWS whether the earlier attempt that persisted +// claim.ClientToken created its listing. found reports that the attempt is +// resolved: result holds the recorded listing, or err why it ended in a dead +// one. Not found means AWS has nothing for the token, so the caller creates +// the listing with that same token. +func (h *Handler) reconcileAttempt(ctx context.Context, purchaseID string, claim *config.MarketplaceListingClaim, finder marketplaceListingFinder) (ec2svc.MarketplaceListingResult, bool, error) { + listing, found, err := finder.FindMarketplaceListingByToken(ctx, purchaseID, claim.ClientToken) + if err != nil { + h.releaseMarketplaceClaim(ctx, purchaseID, claim, true) + return ec2svc.MarketplaceListingResult{}, false, mapAWSMarketplaceError("could not reconcile the earlier listing attempt on AWS", err) + } + if !found { + return ec2svc.MarketplaceListingResult{}, false, nil + } + logging.Warnf("marketplace: purchase %s: earlier attempt already created listing %s (%s); recording it", purchaseID, listing.ListingID, listing.State) + if isDeadListingState(listing.State) { + return ec2svc.MarketplaceListingResult{}, true, h.recordDeadListing(ctx, purchaseID, claim, listing) + } + if err = h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, listing.ListingID, listing.State); err != nil { + h.releaseMarketplaceClaim(ctx, purchaseID, claim, true) + return ec2svc.MarketplaceListingResult{}, true, fmt.Errorf("listing %s exists on AWS but could not be recorded: %w", listing.ListingID, err) + } + return listing, true, nil +} + +func (h *Handler) createAndPersistListing(ctx context.Context, purchaseID string, claim *config.MarketplaceListingClaim, ec2Client marketplaceEC2Client, instanceCount int32, awsSchedule []ec2svc.MarketplacePriceTier) (ec2svc.MarketplaceListingResult, error) { + result, err := ec2Client.CreateMarketplaceListing(ctx, ec2svc.MarketplaceListingRequest{ + ReservedInstancesID: purchaseID, + ClientToken: claim.ClientToken, + PriceSchedule: awsSchedule, + InstanceCount: instanceCount, + }) + if err != nil { + mapped := mapAWSMarketplaceError("AWS marketplace listing failed", err) + // Release the claim so a retry is not blocked. AWS may still have created + // the listing unless it rejected the request as a client fault, so only + // then is the token dropped; otherwise the retry resends it. + h.releaseMarketplaceClaim(ctx, purchaseID, claim, !isClientFaultError(mapped)) + logging.Warnf("marketplace: CreateReservedInstancesListing for purchase %s failed: %v", purchaseID, err) + return ec2svc.MarketplaceListingResult{}, mapped + } + + if isDeadListingState(result.State) { + return ec2svc.MarketplaceListingResult{}, h.recordDeadListing(ctx, purchaseID, claim, result) + } + + // Persist the listing ID and state. On DB failure, attempt a compensating + // rollback (cancel the just-created listing) to avoid a desync where the + // user sees success but the listing is invisible in subsequent renders, then + // release the claim so the row does not stay stuck in the pending state. + if dbErr := h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, result.ListingID, result.State); dbErr != nil { + logging.Errorf("marketplace: listing created (%s / %s) but DB update failed: %v; attempting rollback", result.ListingID, result.State, dbErr) + compCtx, cancelComp := context.WithTimeout(context.WithoutCancel(ctx), marketplaceCompensationTimeout) + defer cancelComp() + if _, rollbackErr := ec2Client.CancelMarketplaceListing(compCtx, result.ListingID); rollbackErr != nil { + return ec2svc.MarketplaceListingResult{}, h.keepUncanceledListing(compCtx, purchaseID, result, rollbackErr) + } + logging.Warnf("marketplace: listing %s rolled back (canceled) after DB failure", result.ListingID) + h.recordRolledBackListing(compCtx, ctx, purchaseID, claim, result.ListingID) + return ec2svc.MarketplaceListingResult{}, fmt.Errorf("listing created but could not be persisted; listing has been rolled back: %w", dbErr) + } + + return result, nil +} + +func isClientFaultError(err error) bool { + ce, ok := IsClientError(err) + return ok && ce.code == 400 +} + +func isDeadListingState(state string) bool { + return strings.EqualFold(state, config.ListingStateCancelled) || strings.EqualFold(state, config.ListingStateClosed) +} + +// recordDeadListing handles a listing of this attempt that is canceled or +// closed on AWS: it records the listing, which also drops the attempt's token, +// so the next attempt starts fresh instead of AWS replaying the dead listing. +func (h *Handler) recordDeadListing(ctx context.Context, purchaseID string, claim *config.MarketplaceListingClaim, listing ec2svc.MarketplaceListingResult) error { + if err := h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, listing.ListingID, listing.State); err != nil { + logging.Errorf("marketplace: failed to record dead listing %s for purchase %s: %v", listing.ListingID, purchaseID, err) + h.releaseMarketplaceClaim(ctx, purchaseID, claim, false) + } + return NewClientError(502, fmt.Sprintf("AWS returned listing %s in state %s instead of a live listing; retry to create a new one", listing.ListingID, listing.State)) +} + +// recordRolledBackListing records the canceled listing and with it drops the +// attempt's token, so the next attempt does not replay the canceled listing. +func (h *Handler) recordRolledBackListing(compCtx, ctx context.Context, purchaseID string, claim *config.MarketplaceListingClaim, listingID string) { + if err := h.config.UpdatePurchaseHistoryListing(compCtx, purchaseID, listingID, config.ListingStateCancelled); err != nil { + logging.Errorf("marketplace: failed to record rolled-back listing %s for purchase %s: %v", listingID, purchaseID, err) + h.releaseMarketplaceClaim(ctx, purchaseID, claim, false) + } +} + +// releaseMarketplaceClaim moves the row from pending back to the state it had +// when claimed. It runs on every failure path after a successful claim so a +// failed attempt does not leave the row stuck in the transient pending state +// (which would block future list attempts unless a token resumes it). +// keepAttempt keeps the persisted token and schedule for the retry. +// Best-effort: on error it logs loudly because the row may stay pending. +func (h *Handler) releaseMarketplaceClaim(ctx context.Context, purchaseID string, claim *config.MarketplaceListingClaim, keepAttempt bool) { + prior := claim.PriorState + if strings.EqualFold(prior, config.ListingStatePending) { + prior = "" + } + if err := h.config.ReleaseMarketplaceListingClaim(ctx, purchaseID, prior, keepAttempt); err != nil { + logging.Errorf("marketplace: failed to release listing claim for purchase %s (row may be stuck in %q): %v", purchaseID, config.ListingStatePending, err) + } +} diff --git a/internal/api/handler_marketplace_attempt_test.go b/internal/api/handler_marketplace_attempt_test.go new file mode 100644 index 000000000..d3a124091 --- /dev/null +++ b/internal/api/handler_marketplace_attempt_test.go @@ -0,0 +1,438 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "testing" + "time" + + ec2svc "github.com/LeanerCloud/cloud-commitments-go/providers/aws/services/ec2" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/aws/aws-sdk-go-v2/aws" + sdkec2 "github.com/aws/aws-sdk-go-v2/service/ec2" + sdkec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types" + smithy "github.com/aws/smithy-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// fakeListingDB is an in-memory purchase_history row that follows the claim, +// release and record semantics of the Postgres store, so the tests below drive +// the handler through several requests against one evolving row. +type fakeListingDB struct { + *MockConfigStore + row config.PurchaseHistoryRecord + token string + schedule []byte + // failRecords makes the next N UpdatePurchaseHistoryListing calls fail. + failRecords int + // staleOnce, when set, is returned by the next GetPurchaseHistoryByPurchaseID. + staleOnce *config.PurchaseHistoryRecord +} + +func newFakeListingDB() *fakeListingDB { + return &fakeListingDB{MockConfigStore: &MockConfigStore{}, row: *standardRow()} +} + +func (f *fakeListingDB) GetPurchaseHistoryByPurchaseID(_ context.Context, _ string) (*config.PurchaseHistoryRecord, error) { + if f.staleOnce != nil { + stale := *f.staleOnce + f.staleOnce = nil + return &stale, nil + } + cp := f.row + return &cp, nil +} + +func (f *fakeListingDB) ClaimMarketplaceListingSlot(_ context.Context, _, clientToken string, priceSchedule []byte) (*config.MarketplaceListingClaim, error) { + state := strings.ToLower(f.row.ListingState) + if state == config.ListingStateActive || (state == config.ListingStatePending && f.token == "") { + return nil, nil + } + resumed := f.token != "" + if !resumed { + f.token, f.schedule = clientToken, priceSchedule + } + prior := f.row.ListingState + f.row.ListingState = config.ListingStatePending + return &config.MarketplaceListingClaim{ListingID: f.row.ListingID, PriorState: prior, ClientToken: f.token, PriceSchedule: f.schedule, Resumed: resumed}, nil +} + +func (f *fakeListingDB) ReleaseMarketplaceListingClaim(_ context.Context, _, priorState string, keepAttempt bool) error { + if strings.EqualFold(f.row.ListingState, config.ListingStatePending) { + f.row.ListingState = priorState + if !keepAttempt { + f.token, f.schedule = "", nil + } + } + return nil +} + +func (f *fakeListingDB) UpdatePurchaseHistoryListing(_ context.Context, _, listingID, state string) error { + if f.failRecords > 0 { + f.failRecords-- + return errors.New("db down") + } + f.row.ListingID, f.row.ListingState = listingID, state + f.token, f.schedule = "", nil + return nil +} + +// recordCanceled marks a listing canceled in the row, like the cancel handler. +func (f *fakeListingDB) recordCanceled(listingID string) { + f.row.ListingID, f.row.ListingState = listingID, config.ListingStateCancelled +} + +type fakeAWSListing struct { + id, state, token string + count int32 + schedule []ec2svc.MarketplacePriceTier +} + +// fakeMarketplaceAWS dedups CreateReservedInstancesListing on ClientToken like +// the real API: a repeat returns the existing listing (even a canceled one) and +// a repeat with different parameters is IdempotentParameterMismatch. +type fakeMarketplaceAWS struct { + listings []*fakeAWSListing + // loseResponses makes the next N creates succeed on AWS but fail for the caller. + loseResponses int + failCancel bool + creates []ec2svc.MarketplaceListingRequest + finds int +} + +func (a *fakeMarketplaceAWS) byToken(token string) *fakeAWSListing { + for _, l := range a.listings { + if l.token == token { + return l + } + } + return nil +} + +func (a *fakeMarketplaceAWS) create(_ context.Context, req ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { + a.creates = append(a.creates, req) + l := a.byToken(req.ClientToken) + switch { + case l == nil: + l = &fakeAWSListing{id: fmt.Sprintf("ril-%d", len(a.listings)+1), state: config.ListingStateActive, token: req.ClientToken, count: req.InstanceCount, schedule: req.PriceSchedule} + a.listings = append(a.listings, l) + case l.count != req.InstanceCount || fmt.Sprint(l.schedule) != fmt.Sprint(req.PriceSchedule): + return ec2svc.MarketplaceListingResult{}, errors.New("IdempotentParameterMismatch") + } + if a.loseResponses > 0 { + a.loseResponses-- + return ec2svc.MarketplaceListingResult{}, errors.New("RequestTimeout: response lost") + } + return ec2svc.MarketplaceListingResult{ListingID: l.id, State: l.state}, nil +} + +func (a *fakeMarketplaceAWS) cancel(_ context.Context, id string) (ec2svc.MarketplaceListingResult, error) { + if a.failCancel { + return ec2svc.MarketplaceListingResult{}, errors.New("RequestLimitExceeded") + } + for _, l := range a.listings { + if l.id == id { + l.state = config.ListingStateCancelled + } + } + return ec2svc.MarketplaceListingResult{ListingID: id, State: config.ListingStateCancelled}, nil +} + +func (a *fakeMarketplaceAWS) FindMarketplaceListingByToken(_ context.Context, _, token string) (ec2svc.MarketplaceListingResult, bool, error) { + a.finds++ + l := a.byToken(token) + if l == nil { + return ec2svc.MarketplaceListingResult{}, false, nil + } + return ec2svc.MarketplaceListingResult{ListingID: l.id, State: l.state}, true, nil +} + +func (a *fakeMarketplaceAWS) live() int { + n := 0 + for _, l := range a.listings { + if l.state == config.ListingStateActive { + n++ + } + } + return n +} + +type marketplaceFixture struct { + h *Handler + db *fakeListingDB + aws *fakeMarketplaceAWS +} + +func newMarketplaceFixture(t *testing.T) *marketplaceFixture { + t.Helper() + db := newFakeListingDB() + authSvc := &MockAuthService{} + adminSession(authSvc) + authSvc.On("ValidateCSRFToken", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() + fake := &fakeMarketplaceAWS{} + // The token and schedule must be on the row before AWS sees the request. + create := func(ctx context.Context, req ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { + require.Equal(t, db.token, req.ClientToken, "the token must be persisted before the AWS call") + var stored []MarketplacePriceTier + require.NoError(t, json.Unmarshal(db.schedule, &stored), "the schedule must be persisted before the AWS call") + require.Equal(t, toAWSPriceSchedule(stored), req.PriceSchedule) + return fake.create(ctx, req) + } + stub := &stubMarketplaceEC2{createFn: create, cancelFn: fake.cancel} + h := newMarketplaceHandler(db.MockConfigStore, authSvc, stub) + h.config = db + h.marketplaceFinderFactory = func(aws.Config) marketplaceListingFinder { return fake } + return &marketplaceFixture{h: h, db: db, aws: fake} +} + +// list drives POST /marketplace-list through HandleRequest and returns the status. +func (f *marketplaceFixture) list(t *testing.T) int { + t.Helper() + resp, err := f.h.HandleRequest(context.Background(), marketplaceListHTTPRequest()) + require.NoError(t, err) + return resp.StatusCode +} + +// monthsLater moves the purchase back one month and a bit so the remaining +// months, and with them the default price schedule, change on the next request. +func (f *marketplaceFixture) monthsLater(t *testing.T) { + t.Helper() + before, err := computeRemainingMonths(f.db.row.Timestamp, f.db.row.Term*12, time.Now()) + require.NoError(t, err) + f.db.row.Timestamp = f.db.row.Timestamp.Add(-731 * time.Hour) + after, err := computeRemainingMonths(f.db.row.Timestamp, f.db.row.Term*12, time.Now()) + require.NoError(t, err) + require.NotEqual(t, before, after, "the clock must cross a month boundary") +} + +// Issue #525: the first attempt created the listing on AWS but the response was +// lost; the operator retries after the remaining months changed (a month +// boundary), so the default schedule differs. The retry must not create a +// second listing. +func TestMarketplaceListHTTP_RetryAcrossMonthBoundaryDoesNotDuplicate(t *testing.T) { + f := newMarketplaceFixture(t) + f.aws.loseResponses = 1 + + assert.Equal(t, 502, f.list(t), "the lost response is reported") + require.Len(t, f.aws.listings, 1) + firstSchedule := f.aws.creates[0].PriceSchedule + f.monthsLater(t) + + assert.Equal(t, 200, f.list(t), "the retry reconciles the listing the first attempt created") + assert.Len(t, f.aws.listings, 1, "AWS must hold exactly one listing") + assert.Equal(t, 1, f.aws.live()) + assert.Equal(t, "ril-1", f.db.row.ListingID) + assert.Equal(t, config.ListingStateActive, f.db.row.ListingState) + assert.Empty(t, f.db.token, "recording the listing resolves the attempt") + assert.Equal(t, 1, f.aws.finds) + assert.Equal(t, firstSchedule, f.aws.creates[0].PriceSchedule) +} + +// The same retry when AWS has no listing for the token (the first create never +// reached AWS): the retry creates it with the persisted token and schedule, not +// ones re-derived from the clock. +func TestMarketplaceListHTTP_RetryReusesPersistedTokenAndSchedule(t *testing.T) { + f := newMarketplaceFixture(t) + // The first create never reaches AWS: fail before the fake records a listing. + calls := 0 + f.h.marketplaceEC2Factory = func(aws.Config) marketplaceEC2Client { + return &stubMarketplaceEC2{createFn: func(ctx context.Context, req ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { + calls++ + if calls == 1 { + f.aws.creates = append(f.aws.creates, req) + return ec2svc.MarketplaceListingResult{}, errors.New("connection reset") + } + return f.aws.create(ctx, req) + }} + } + + assert.Equal(t, 502, f.list(t)) + require.Len(t, f.aws.creates, 1) + f.monthsLater(t) + + assert.Equal(t, 200, f.list(t)) + require.Len(t, f.aws.creates, 2) + assert.Equal(t, f.aws.creates[0].ClientToken, f.aws.creates[1].ClientToken, "the retry must resend the persisted token") + assert.Equal(t, f.aws.creates[0].PriceSchedule, f.aws.creates[1].PriceSchedule, "the retry must resend the persisted schedule, not one re-derived from the clock") + assert.Len(t, f.aws.listings, 1) +} + +// A crash after the AWS create leaves the row pending with the token. The +// persist step fails and so does the compensating cancel (and the record of the +// uncanceled listing), as in a process that dies right there. The next request +// resumes the pending row, finds the listing on AWS and records it. +func TestMarketplaceListHTTP_CrashBetweenCreateAndPersistIsReconciled(t *testing.T) { + f := newMarketplaceFixture(t) + f.db.failRecords = 2 + f.aws.failCancel = true + + assert.Equal(t, 502, f.list(t)) + assert.Equal(t, config.ListingStatePending, f.db.row.ListingState, "the crashed attempt leaves the claim in place") + assert.NotEmpty(t, f.db.token) + f.monthsLater(t) + + assert.Equal(t, 200, f.list(t)) + assert.Len(t, f.aws.listings, 1) + assert.Equal(t, 1, f.aws.live()) + assert.Equal(t, "ril-1", f.db.row.ListingID) + assert.Equal(t, config.ListingStateActive, f.db.row.ListingState) +} + +// Reconcile finding a listing that is already dead records it and asks for a +// retry; the next attempt starts with a fresh token. +func TestMarketplaceListHTTP_ReconcileFindsCanceledListing(t *testing.T) { + f := newMarketplaceFixture(t) + f.aws.loseResponses = 1 + assert.Equal(t, 502, f.list(t)) + _, err := f.aws.cancel(context.Background(), "ril-1") + require.NoError(t, err) + + assert.Equal(t, 502, f.list(t), "the dead listing is reported, not replayed as success") + assert.Equal(t, config.ListingStateCancelled, f.db.row.ListingState) + assert.Equal(t, "ril-1", f.db.row.ListingID) + assert.Empty(t, f.db.token) + + assert.Equal(t, 200, f.list(t)) + assert.Len(t, f.aws.listings, 2) + assert.Equal(t, "ril-2", f.db.row.ListingID) +} + +// A definitive AWS rejection created nothing, so its token is dropped and the +// operator's next request (possibly with another schedule) starts fresh. +func TestMarketplaceListHTTP_ClientFaultDropsTheAttempt(t *testing.T) { + f := newMarketplaceFixture(t) + f.h.marketplaceEC2Factory = func(aws.Config) marketplaceEC2Client { + return &stubMarketplaceEC2{createFn: func(context.Context, ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { + return ec2svc.MarketplaceListingResult{}, &marketplaceTestAPIError{code: "InvalidParameterValue", message: "bad price", fault: smithy.FaultClient} + }} + } + assert.Equal(t, 400, f.list(t)) + assert.Empty(t, f.db.token) + assert.Empty(t, f.db.row.ListingState) +} + +// The row was read before another request claimed, listed and canceled it, so +// the stale copy knows nothing of the canceled listing. The token must not be +// derived from that copy: with a persisted random token the stale request gets +// its own listing instead of AWS replaying the canceled one. +func TestMarketplaceListHTTP_StaleRowAfterCancelGetsNewListing(t *testing.T) { + f := newMarketplaceFixture(t) + stale := f.db.row + assert.Equal(t, 200, f.list(t)) + _, err := f.aws.cancel(context.Background(), "ril-1") + require.NoError(t, err) + f.db.recordCanceled("ril-1") + + f.db.staleOnce = &stale + assert.Equal(t, 200, f.list(t), "the stale request must create its own listing") + assert.Len(t, f.aws.listings, 2) + assert.Equal(t, "ril-2", f.db.row.ListingID) + assert.Equal(t, 1, f.aws.live()) +} + +// After a persist failure and a successful compensating cancel the canceled +// listing is recorded and the token dropped, so the retry creates a new one. +func TestMarketplaceListHTTP_RetryAfterRolledBackPersistFailureGetsNewListing(t *testing.T) { + f := newMarketplaceFixture(t) + f.db.failRecords = 1 + + assert.Equal(t, 500, f.list(t)) + assert.Equal(t, config.ListingStateCancelled, f.db.row.ListingState) + assert.Empty(t, f.db.token) + + assert.Equal(t, 200, f.list(t)) + assert.Equal(t, "ril-2", f.db.row.ListingID) + assert.Equal(t, 1, f.aws.live()) +} + +// A create response in a dead state is an error and the dead listing is recorded. +func TestMarketplaceListHTTP_ReturnedCanceledStateIsAnError(t *testing.T) { + f := newMarketplaceFixture(t) + f.h.marketplaceEC2Factory = func(aws.Config) marketplaceEC2Client { + return &stubMarketplaceEC2{createFn: func(context.Context, ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { + return ec2svc.MarketplaceListingResult{ListingID: "ril-dead", State: config.ListingStateCancelled}, nil + }} + } + assert.Equal(t, 502, f.list(t)) + assert.Equal(t, "ril-dead", f.db.row.ListingID) + assert.Empty(t, f.db.token) +} + +func TestMarketplaceListHTTP_ReconcileLookupFailureKeepsTheAttempt(t *testing.T) { + f := newMarketplaceFixture(t) + f.aws.loseResponses = 1 + assert.Equal(t, 502, f.list(t)) + token := f.db.token + f.h.marketplaceFinderFactory = func(aws.Config) marketplaceListingFinder { return failingFinder{} } + + assert.Equal(t, 502, f.list(t)) + assert.Equal(t, token, f.db.token, "an unresolved attempt keeps its token") + assert.Len(t, f.aws.creates, 1, "no create while the earlier attempt is unresolved") + assert.NotEqual(t, config.ListingStatePending, f.db.row.ListingState) +} + +type failingFinder struct{} + +func (failingFinder) FindMarketplaceListingByToken(context.Context, string, string) (ec2svc.MarketplaceListingResult, bool, error) { + return ec2svc.MarketplaceListingResult{}, false, errors.New("RequestLimitExceeded") +} + +func TestMarketplaceListHTTP_CorruptStoredScheduleFailsLoud(t *testing.T) { + f := newMarketplaceFixture(t) + f.db.token, f.db.schedule = "tok", []byte("[]") + f.db.row.ListingState = config.ListingStatePending + + assert.Equal(t, 500, f.list(t)) + assert.Empty(t, f.aws.creates) + assert.Equal(t, "tok", f.db.token) +} + +// The schedule is stored as JSON that decodes back to the same tiers. +func TestMarketplaceListHTTP_PersistsTheSchedule(t *testing.T) { + f := newMarketplaceFixture(t) + f.aws.loseResponses = 1 + assert.Equal(t, 502, f.list(t)) + + var stored []MarketplacePriceTier + require.NoError(t, json.Unmarshal(f.db.schedule, &stored)) + assert.Equal(t, toAWSPriceSchedule(stored), f.aws.creates[0].PriceSchedule) + assert.Equal(t, f.aws.creates[0].ClientToken, f.db.token) +} + +type fakeDescribeListings struct { + out []sdkec2types.ReservedInstancesListing + err error + in *sdkec2.DescribeReservedInstancesListingsInput +} + +func (f *fakeDescribeListings) DescribeReservedInstancesListings(_ context.Context, in *sdkec2.DescribeReservedInstancesListingsInput, _ ...func(*sdkec2.Options)) (*sdkec2.DescribeReservedInstancesListingsOutput, error) { + f.in = in + return &sdkec2.DescribeReservedInstancesListingsOutput{ReservedInstancesListings: f.out}, f.err +} + +func TestAWSMarketplaceListingFinder(t *testing.T) { + api := &fakeDescribeListings{out: []sdkec2types.ReservedInstancesListing{ + {ClientToken: aws.String("other"), ReservedInstancesListingId: aws.String("ril-other"), Status: sdkec2types.ListingStatusActive}, + {ClientToken: aws.String("tok"), ReservedInstancesListingId: aws.String("ril-mine"), Status: sdkec2types.ListingStatusCancelled}, + }} + finder := awsMarketplaceListingFinder{api: api} + + got, found, err := finder.FindMarketplaceListingByToken(context.Background(), "ri-1", "tok") + require.NoError(t, err) + assert.True(t, found) + assert.Equal(t, ec2svc.MarketplaceListingResult{ListingID: "ril-mine", State: string(sdkec2types.ListingStatusCancelled)}, got) + assert.Equal(t, "ri-1", aws.ToString(api.in.ReservedInstancesId)) + + _, found, err = finder.FindMarketplaceListingByToken(context.Background(), "ri-1", "missing") + require.NoError(t, err) + assert.False(t, found) + + api.err = errors.New("boom") + _, _, err = finder.FindMarketplaceListingByToken(context.Background(), "ri-1", "tok") + require.Error(t, err) +} diff --git a/internal/api/handler_marketplace_credentials_test.go b/internal/api/handler_marketplace_credentials_test.go index e2dd14ef6..ce5faa811 100644 --- a/internal/api/handler_marketplace_credentials_test.go +++ b/internal/api/handler_marketplace_credentials_test.go @@ -53,8 +53,9 @@ func marketplaceAccountHandler(t *testing.T, account *config.CloudAccount, row * adminSession(authSvc) cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(row, nil) cfgStore.On("GetCloudAccount", mock.Anything, "acct-1").Return(account, nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID).Return(true, nil).Maybe() + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything).Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil).Maybe() cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything).Return(nil).Maybe() + cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything).Return(nil).Maybe() var built aws.Config h := &Handler{ @@ -220,7 +221,7 @@ func TestMarketplaceList_CompensationAssumeRoleDeniedReturnsFixedMessage(t *test row := standardRow() cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(row, nil) cfgStore.On("GetCloudAccount", mock.Anything, "acct-1").Return(roleAccount(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID).Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything).Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything).Return(errors.New("db down")) h := &Handler{config: cfgStore, credStore: &accountKeyStore{}, auth: authSvc} h.awsCfgOnce.Do(func() { diff --git a/internal/api/handler_marketplace_test.go b/internal/api/handler_marketplace_test.go index 15878b726..e85f20eaf 100644 --- a/internal/api/handler_marketplace_test.go +++ b/internal/api/handler_marketplace_test.go @@ -3,7 +3,6 @@ package api import ( "context" "errors" - "fmt" "testing" "time" @@ -170,8 +169,8 @@ func TestMarketplaceList_SellOwnAllowed(t *testing.T) { cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). Return(standardRow(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive). Return(nil) @@ -242,10 +241,10 @@ func TestMarketplaceList_AWSClientFaultMapsTo400(t *testing.T) { adminSession(authSvc) cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). Return(standardRow(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) // AWS create fails, so the claim must be released back to the unlisted state. - cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "", ""). + cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, "", false). Return(nil) ec2 := &stubMarketplaceEC2{ @@ -272,10 +271,10 @@ func TestMarketplaceList_AWSServerFaultMapsTo502(t *testing.T) { adminSession(authSvc) cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). Return(standardRow(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) // AWS create fails, so the claim must be released back to the unlisted state. - cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "", ""). + cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, "", true). Return(nil) ec2 := &stubMarketplaceEC2{ @@ -301,10 +300,10 @@ func TestMarketplaceList_UnknownErrorMapsTo502(t *testing.T) { adminSession(authSvc) cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). Return(standardRow(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) // AWS create fails, so the claim must be released back to the unlisted state. - cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "", ""). + cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, "", true). Return(nil) ec2 := &stubMarketplaceEC2{ @@ -328,8 +327,8 @@ func TestMarketplaceList_DBFailureCompensatingRollback(t *testing.T) { adminSession(authSvc) cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). Return(standardRow(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) // DB persist fails after the listing was created. cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive). Return(errors.New("db down")) @@ -365,8 +364,8 @@ func TestMarketplaceList_DBFailureAndCancelFailureKeepsListing(t *testing.T) { adminSession(authSvc) cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). Return(standardRow(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive). Return(errors.New("db down")).Once() cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive). @@ -389,7 +388,7 @@ func TestMarketplaceList_DBFailureAndCancelFailureKeepsListing(t *testing.T) { assert.Contains(t, ce.message, "throttled") assert.Contains(t, ce.message, tc.wantMsg) assert.Equal(t, 1, ec2.cancelCallCount) - cfgStore.AssertNotCalled(t, "UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "", "") + cfgStore.AssertNotCalled(t, "ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything) cfgStore.AssertExpectations(t) }) } @@ -412,8 +411,8 @@ func TestMarketplaceList_CancelFailureRecordsNonActiveStateAsPending(t *testing. adminSession(authSvc) cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). Return(standardRow(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-x", tc.awsState). Return(errors.New("db down")).Once() cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-x", tc.wantState). @@ -448,8 +447,8 @@ func TestMarketplaceList_CancelFailureAfterRequestContextCanceledStillRecords(t adminSession(authSvc) cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). Return(standardRow(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) // The request context dies during the first write, as on a deadline. cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive). Run(func(mock.Arguments) { cancel() }).Return(errors.New("db down")).Once() @@ -558,8 +557,8 @@ func TestMarketplaceList_ConcurrentClaimLoses409(t *testing.T) { cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). Return(standardRow(), nil) // A concurrent request already holds the slot: the atomic claim reports lost. - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(false, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(nil, nil) ec2 := &stubMarketplaceEC2{} h := newMarketplaceHandler(cfgStore, authSvc, ec2) @@ -584,8 +583,8 @@ func TestMarketplaceList_ClaimErrorMapsToInternal(t *testing.T) { cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). Return(standardRow(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(false, errors.New("db down")) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(nil, errors.New("db down")) ec2 := &stubMarketplaceEC2{} h := newMarketplaceHandler(cfgStore, authSvc, ec2) @@ -701,8 +700,8 @@ func TestMarketplaceList_EmptyOfferingClassFetchedStandard(t *testing.T) { // The handler must stamp the fetched class back to the DB. cfgStore.On("StampOfferingClass", mock.Anything, validMarketplacePurchaseID, "standard"). Return(nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive). Return(nil) @@ -855,7 +854,7 @@ func TestMarketplaceListHTTP_UnlistableTermsReturn422(t *testing.T) { assert.Equal(t, 422, resp.StatusCode, "body: %s", resp.Body) assert.Contains(t, resp.Body, tc.wantBody) assert.Zero(t, ec2.createCallCount, "no AWS call for an unlistable RI") - cfgStore.AssertNotCalled(t, "ClaimMarketplaceListingSlot", mock.Anything, mock.Anything) + cfgStore.AssertNotCalled(t, "ClaimMarketplaceListingSlot", mock.Anything, mock.Anything, mock.Anything, mock.Anything) }) } } @@ -878,7 +877,7 @@ func TestMarketplaceListHTTP_ValidRIStillLists(t *testing.T) { row := standardRow() row.Timestamp = tc.timestamp cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(row, nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID).Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything).Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive).Return(nil) ec2 := &stubMarketplaceEC2{} @@ -917,8 +916,8 @@ func TestMarketplaceList_TermYearsConvertedToMonths(t *testing.T) { cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). Return(row, nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(true, nil) + cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). + Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive). Return(nil) @@ -1024,128 +1023,3 @@ func TestMarketplaceList_ZeroTimestampReturnsError(t *testing.T) { assert.Zero(t, ec2.createCallCount, "no listing may be created for an unpriceable RI") cfgStore.AssertExpectations(t) } - -// A timeout after AWS accepted CreateReservedInstancesListing releases the -// claim, so the operator retries. The retry must reuse the ClientToken so AWS -// returns the existing listing instead of creating a second live one. -func TestMarketplaceList_AmbiguousAWSErrorRetryDoesNotDuplicateListing(t *testing.T) { - cfgStore := &MockConfigStore{} - authSvc := &MockAuthService{} - adminSession(authSvc) - cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID). - Return(standardRow(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID). - Return(true, nil) - cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). - Return(nil) - - // Fake AWS: dedups on ClientToken like the real API, and the first call - // loses its response after creating the listing. - listingsByToken := map[string]string{} - calls := 0 - ec2 := &stubMarketplaceEC2{createFn: func(_ context.Context, req ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { - calls++ - id, ok := listingsByToken[req.ClientToken] - if !ok { - id = "ril-" + req.ClientToken - listingsByToken[req.ClientToken] = id - } - if calls == 1 { - return ec2svc.MarketplaceListingResult{}, errors.New("RequestTimeout: response lost") - } - return ec2svc.MarketplaceListingResult{ListingID: id, State: config.ListingStateActive}, nil - }} - h := newMarketplaceHandler(cfgStore, authSvc, ec2) - - _, err := h.marketplaceList(context.Background(), marketplaceReq(), validMarketplacePurchaseID) - require.Error(t, err) - _, err = h.marketplaceList(context.Background(), marketplaceReq(), validMarketplacePurchaseID) - require.NoError(t, err) - - assert.Len(t, listingsByToken, 1, "the retry must reuse the ClientToken so AWS does not create a second listing") -} - -// A deliberate re-list after a listing was canceled must not replay the dead -// listing: the recorded listing id changes the token. -func TestMarketplaceClientToken(t *testing.T) { - sched := []ec2svc.MarketplacePriceTier{{Term: 12, Price: 100}, {Term: 6, Price: 50}} - base := marketplaceClientToken("p1", "", 3, sched) - - assert.Equal(t, base, marketplaceClientToken("p1", "", 3, sched), "same attempt, same token") - assert.NotEqual(t, base, marketplaceClientToken("p1", "ril-old", 3, sched), "re-list after cancel") - assert.NotEqual(t, base, marketplaceClientToken("p2", "", 3, sched), "other RI") - assert.NotEqual(t, base, marketplaceClientToken("p1", "", 2, sched), "other count") - assert.NotEqual(t, base, marketplaceClientToken("p1", "", 3, sched[:1]), "other schedule") -} - -// replayingAWS dedups on ClientToken and replays a listing that was canceled. -type replayingAWS struct { - byToken map[string]string - canceled map[string]bool - n int -} - -func (a *replayingAWS) create(_ context.Context, req ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { - id, ok := a.byToken[req.ClientToken] - if !ok { - a.n++ - id = fmt.Sprintf("ril-%d", a.n) - a.byToken[req.ClientToken] = id - } - state := config.ListingStateActive - if a.canceled[id] { - state = config.ListingStateCancelled - } - return ec2svc.MarketplaceListingResult{ListingID: id, State: state}, nil -} - -func (a *replayingAWS) cancel(_ context.Context, id string) (ec2svc.MarketplaceListingResult, error) { - a.canceled[id] = true - return ec2svc.MarketplaceListingResult{ListingID: id, State: config.ListingStateCancelled}, nil -} - -func TestMarketplaceList_RetryAfterRolledBackPersistFailureGetsNewListing(t *testing.T) { - cfgStore := &MockConfigStore{} - authSvc := &MockAuthService{} - adminSession(authSvc) - // The store reflects what the handler records, like the real table. - cur := standardRow() - cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(cur, nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID).Return(true, nil) - record := func(args mock.Arguments) { cur.ListingID, cur.ListingState = args.String(2), args.String(3) } - cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-1", config.ListingStateActive). - Return(errors.New("db down")).Once() - cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). - Run(record).Return(nil) - - aws := &replayingAWS{byToken: map[string]string{}, canceled: map[string]bool{}} - ec2 := &stubMarketplaceEC2{createFn: aws.create, cancelFn: aws.cancel} - h := newMarketplaceHandler(cfgStore, authSvc, ec2) - - _, err := h.marketplaceList(context.Background(), marketplaceReq(), validMarketplacePurchaseID) - require.Error(t, err) - assert.Contains(t, err.Error(), "rolled back") - - got, err := h.marketplaceList(context.Background(), marketplaceReq(), validMarketplacePurchaseID) - require.NoError(t, err) - resp := got.(*MarketplaceListResponse) - assert.Equal(t, config.ListingStateActive, resp.ListingState) - assert.Equal(t, "ril-2", resp.ListingID, "the retry must create a new listing, not replay the canceled one") -} - -func TestMarketplaceList_ReturnedCanceledStateIsAnError(t *testing.T) { - cfgStore := &MockConfigStore{} - authSvc := &MockAuthService{} - adminSession(authSvc) - cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(standardRow(), nil) - cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID).Return(true, nil) - cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-dead", config.ListingStateCancelled).Return(nil) - ec2 := &stubMarketplaceEC2{createFn: func(context.Context, ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { - return ec2svc.MarketplaceListingResult{ListingID: "ril-dead", State: config.ListingStateCancelled}, nil - }} - h := newMarketplaceHandler(cfgStore, authSvc, ec2) - - _, err := h.marketplaceList(context.Background(), marketplaceReq(), validMarketplacePurchaseID) - require.Error(t, err) - cfgStore.AssertExpectations(t) -} diff --git a/internal/api/handler_purchase_action_test.go b/internal/api/handler_purchase_action_test.go index aadb10f29..7b54e4678 100644 --- a/internal/api/handler_purchase_action_test.go +++ b/internal/api/handler_purchase_action_test.go @@ -58,7 +58,7 @@ func TestMarketplaceScopeLookupErrors(t *testing.T) { require.ErrorIs(t, err, lookupErr) require.Zero(t, provider.createCallCount) require.Zero(t, provider.cancelCallCount) - store.AssertNotCalled(t, "ClaimMarketplaceListingSlot", mock.Anything, mock.Anything) + store.AssertNotCalled(t, "ClaimMarketplaceListingSlot", mock.Anything, mock.Anything, mock.Anything, mock.Anything) store.AssertNotCalled(t, "UpdatePurchaseHistoryListing", mock.Anything, mock.Anything, mock.Anything, mock.Anything) service.AssertExpectations(t) } diff --git a/internal/api/nullable_purchase_cost_test.go b/internal/api/nullable_purchase_cost_test.go index d398c0cc7..611190a98 100644 --- a/internal/api/nullable_purchase_cost_test.go +++ b/internal/api/nullable_purchase_cost_test.go @@ -117,7 +117,7 @@ func TestMarketplaceUnknownUpfront(t *testing.T) { row.UpfrontCost = tc.cost store.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(row, nil) if tc.allowed { - store.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID).Return(true, nil) + store.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything).Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) store.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive).Return(nil) } req := marketplaceReq() @@ -129,7 +129,7 @@ func TestMarketplaceUnknownUpfront(t *testing.T) { } else { require.ErrorContains(t, err, "upfront cost is unknown") assert.Zero(t, ec2.createCallCount) - store.AssertNotCalled(t, "ClaimMarketplaceListingSlot", mock.Anything, mock.Anything) + store.AssertNotCalled(t, "ClaimMarketplaceListingSlot", mock.Anything, mock.Anything, mock.Anything, mock.Anything) } store.AssertExpectations(t) }) diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 4e91041f8..14f62aa93 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -304,7 +304,9 @@ type StoreInterface interface { // UpdatePurchaseHistoryListing stamps the AWS marketplace listing_id and // listing_state onto a purchase_history row. Called after // CreateReservedInstancesListing succeeds (listing_state="active") and - // on subsequent poll/cancel transitions (issue #292). + // on subsequent poll/cancel transitions (issue #292). Recording a listing + // resolves the attempt, so it also clears the persisted client token and + // price schedule (issue #525). UpdatePurchaseHistoryListing(ctx context.Context, purchaseID, listingID, listingState string) error // StampOfferingClass writes the offering_class value to a purchase_history @@ -319,13 +321,22 @@ type StoreInterface interface { // slot for a purchase_history row so two concurrent marketplace-list // requests cannot both proceed to create a duplicate AWS listing (issue // #292). It transitions listing_state to ListingStatePending only when the - // row is not already listed or mid-listing, and reports whether this call - // won the claim: (true, nil) means the caller reserved the slot and must - // then persist the real listing on success or release the slot back to its - // prior state on failure; (false, nil) means another request already holds - // an active or pending listing (the caller maps this to a 409). Modeled on - // FlipPurchaseRevocationInFlight. - ClaimMarketplaceListingSlot(ctx context.Context, purchaseID string) (bool, error) + // row is not listed, and in the same UPDATE persists clientToken and + // priceSchedule (JSON) unless an unresolved attempt already stored a token, + // in which case that token and schedule are kept so the retry is idempotent + // on AWS (issue #525). A pending row that holds a token is claimable again: + // the same token makes a concurrent or crashed attempt safe to resume. + // It returns the claimed row's values, or (nil, nil) when the row is active, + // pending without a token, or absent (the caller maps that to a 409). The + // caller must persist the real listing on success or call + // ReleaseMarketplaceListingClaim on failure. + ClaimMarketplaceListingSlot(ctx context.Context, purchaseID, clientToken string, priceSchedule []byte) (*MarketplaceListingClaim, error) + + // ReleaseMarketplaceListingClaim moves a pending row back to priorState + // (empty for none) after a failed attempt. keepAttempt keeps the persisted + // token and schedule so the retry resends them (ambiguous AWS outcome); + // otherwise they are cleared because AWS created nothing reusable. + ReleaseMarketplaceListingClaim(ctx context.Context, purchaseID, priorState string, keepAttempt bool) error // ClaimRIExchangeIdempotencyKey atomically claims key for an RI exchange // submit, so a client that retries a timed-out execute request cannot diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 63c284904..1554c0119 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -2055,7 +2055,8 @@ func (s *PostgresStore) SavePurchaseHistory(ctx context.Context, record *Purchas func (s *PostgresStore) UpdatePurchaseHistoryListing(ctx context.Context, purchaseID, listingID, listingState string) error { query := ` UPDATE purchase_history - SET listing_id = $1, listing_state = $2 + SET listing_id = $1, listing_state = $2, + listing_client_token = NULL, listing_price_schedule = NULL WHERE purchase_id = $3 ` tag, err := s.db.Exec(ctx, query, listingID, listingState, purchaseID) @@ -2085,23 +2086,57 @@ func (s *PostgresStore) StampOfferingClass(ctx context.Context, purchaseID, offe // ClaimMarketplaceListingSlot atomically reserves the marketplace-listing slot // for a purchase_history row so two concurrent marketplace-list requests cannot // both proceed to create a duplicate AWS listing (issue #292). The single -// conditional UPDATE transitions listing_state to ListingStatePending only when -// the row is not already active or pending, so exactly one racing request wins. -// It returns (true, nil) when this call reserved the slot and (false, nil) when -// the row is already active/pending (or absent); the caller maps false to a 409. -func (s *PostgresStore) ClaimMarketplaceListingSlot(ctx context.Context, purchaseID string) (bool, error) { +// conditional UPDATE transitions listing_state to ListingStatePending unless the +// row is active, or pending without a persisted token. It stores clientToken and +// priceSchedule in the same statement unless an earlier attempt already stored a +// token, and returns the row's values as of the claim (issue #525): the handler +// must not trust a row read before the claim. It returns (nil, nil) when the row +// could not be claimed (active, pending without a token, or absent). +func (s *PostgresStore) ClaimMarketplaceListingSlot(ctx context.Context, purchaseID, clientToken string, priceSchedule []byte) (*MarketplaceListingClaim, error) { query := ` - UPDATE purchase_history - SET listing_state = $1 - WHERE purchase_id = $2 - AND lower(COALESCE(listing_state, '')) NOT IN ($3, $4) + UPDATE purchase_history AS p + SET listing_state = $1, + listing_client_token = COALESCE(p.listing_client_token, $3::text), + listing_price_schedule = CASE WHEN p.listing_client_token IS NULL + THEN $4::jsonb ELSE p.listing_price_schedule END + FROM (SELECT purchase_id, listing_state AS prior_state, + listing_client_token IS NOT NULL AS had_token + FROM purchase_history WHERE purchase_id = $2) AS prev + WHERE p.purchase_id = prev.purchase_id + AND lower(COALESCE(p.listing_state, '')) <> $5 + AND (lower(COALESCE(p.listing_state, '')) <> $1 OR p.listing_client_token IS NOT NULL) + RETURNING COALESCE(p.listing_id, ''), COALESCE(prev.prior_state, ''), + p.listing_client_token, p.listing_price_schedule::text, prev.had_token ` - tag, err := s.db.Exec(ctx, query, - ListingStatePending, purchaseID, ListingStateActive, ListingStatePending) + var claim MarketplaceListingClaim + var schedule string + err := s.db.QueryRow(ctx, query, ListingStatePending, purchaseID, clientToken, priceSchedule, ListingStateActive). + Scan(&claim.ListingID, &claim.PriorState, &claim.ClientToken, &schedule, &claim.Resumed) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } if err != nil { - return false, fmt.Errorf("failed to claim marketplace listing slot for purchase %s: %w", purchaseID, err) + return nil, fmt.Errorf("failed to claim marketplace listing slot for purchase %s: %w", purchaseID, err) } - return tag.RowsAffected() == 1, nil + claim.PriceSchedule = []byte(schedule) + return &claim, nil +} + +// ReleaseMarketplaceListingClaim moves a pending row back to priorState and, +// unless keepAttempt, clears the persisted token and schedule. It only touches +// a row that is still pending so it never overwrites a recorded listing. +func (s *PostgresStore) ReleaseMarketplaceListingClaim(ctx context.Context, purchaseID, priorState string, keepAttempt bool) error { + query := ` + UPDATE purchase_history + SET listing_state = NULLIF($2, ''), + listing_client_token = CASE WHEN $3 THEN listing_client_token END, + listing_price_schedule = CASE WHEN $3 THEN listing_price_schedule END + WHERE purchase_id = $1 AND lower(COALESCE(listing_state, '')) = $4 + ` + if _, err := s.db.Exec(ctx, query, purchaseID, priorState, keepAttempt, ListingStatePending); err != nil { + return fmt.Errorf("failed to release marketplace listing claim for purchase %s: %w", purchaseID, err) + } + return nil } // GetPurchaseHistory retrieves purchase history for an account. diff --git a/internal/config/store_postgres_marketplace_claim_integration_test.go b/internal/config/store_postgres_marketplace_claim_integration_test.go new file mode 100644 index 000000000..9e97a4e9e --- /dev/null +++ b/internal/config/store_postgres_marketplace_claim_integration_test.go @@ -0,0 +1,113 @@ +//go:build integration +// +build integration + +package config + +// Real-DB tests for the marketplace listing claim (issue #525). The claim, +// release and record statements decide whether a retry may resume an attempt, +// so their predicates are checked against Postgres rather than a mock. + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const claimTestSchedule = `[{"term_months":36,"price":100}]` + +func TestMarketplaceListingClaim_PostgresSemantics(t *testing.T) { + ctx := context.Background() + container := testhelpers.RequirePostgresContainer(ctx, t) + t.Cleanup(func() { assert.NoError(t, container.Cleanup(context.Background())) }) + require.NoError(t, migrations.RunMigrations(ctx, container.DB.Pool(), getTestMigrationsPath(), "", "")) + + checkMarketplaceListingClaim(ctx, t, NewPostgresStore(container.DB)) +} + +func checkMarketplaceListingClaim(ctx context.Context, t *testing.T, store *PostgresStore) { + t.Helper() + const id = "ri-claim-525" + require.NoError(t, store.SavePurchaseHistory(ctx, &PurchaseHistoryRecord{ + AccountID: "acct", PurchaseID: id, Timestamp: time.Now(), Provider: "aws", Service: "ec2", + Region: "us-east-1", ResourceType: "t3.micro", Count: 1, Term: 3, Payment: "All Upfront", + })) + row := func() (listingID, state string) { + got, err := store.GetPurchaseHistoryByPurchaseID(ctx, id) + require.NoError(t, err) + return got.ListingID, got.ListingState + } + + // A fresh row is claimed and the token and schedule are stored with it. + claim, err := store.ClaimMarketplaceListingSlot(ctx, id, "tok-1", []byte(claimTestSchedule)) + require.NoError(t, err) + require.NotNil(t, claim) + assert.False(t, claim.Resumed) + assert.Equal(t, "tok-1", claim.ClientToken) + assert.JSONEq(t, claimTestSchedule, string(claim.PriceSchedule)) + assert.Empty(t, claim.PriorState) + _, state := row() + assert.Equal(t, ListingStatePending, state) + + // A pending row holding a token can be claimed again: the stored token and + // schedule win over the new ones, and the claim says it resumed. + claim, err = store.ClaimMarketplaceListingSlot(ctx, id, "tok-2", []byte(`[{"term_months":35,"price":1}]`)) + require.NoError(t, err) + require.NotNil(t, claim) + assert.True(t, claim.Resumed) + assert.Equal(t, "tok-1", claim.ClientToken) + assert.JSONEq(t, claimTestSchedule, string(claim.PriceSchedule)) + assert.Equal(t, ListingStatePending, claim.PriorState) + + // Release keeping the attempt: the state returns, the token stays. + require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, "", true)) + _, state = row() + assert.Empty(t, state) + claim, err = store.ClaimMarketplaceListingSlot(ctx, id, "tok-3", []byte(claimTestSchedule)) + require.NoError(t, err) + require.NotNil(t, claim) + assert.True(t, claim.Resumed) + assert.Equal(t, "tok-1", claim.ClientToken) + + // Release dropping the attempt: the next claim starts with its own token. + require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, "", false)) + claim, err = store.ClaimMarketplaceListingSlot(ctx, id, "tok-4", []byte(claimTestSchedule)) + require.NoError(t, err) + require.NotNil(t, claim) + assert.False(t, claim.Resumed) + assert.Equal(t, "tok-4", claim.ClientToken) + + // Recording a listing resolves the attempt and clears the token. + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-1", ListingStateActive)) + claim, err = store.ClaimMarketplaceListingSlot(ctx, id, "tok-5", []byte(claimTestSchedule)) + require.NoError(t, err) + assert.Nil(t, claim, "an active listing is never claimable") + + // After a cancel the claim sees the recorded listing and a fresh attempt; + // release only touches a pending row, never a recorded listing. + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-1", ListingStateCancelled)) + claim, err = store.ClaimMarketplaceListingSlot(ctx, id, "tok-6", []byte(claimTestSchedule)) + require.NoError(t, err) + require.NotNil(t, claim) + assert.False(t, claim.Resumed) + assert.Equal(t, "ril-1", claim.ListingID) + assert.Equal(t, ListingStateCancelled, claim.PriorState) + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-2", ListingStateActive)) + require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, ListingStateCancelled, false)) + listingID, state := row() + assert.Equal(t, "ril-2", listingID) + assert.Equal(t, ListingStateActive, state, "release must not overwrite a recorded listing") + + // Pending with no token (a legacy stuck row) is not claimable; an absent row is not either. + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-2", ListingStatePending)) + claim, err = store.ClaimMarketplaceListingSlot(ctx, id, "tok-7", []byte(claimTestSchedule)) + require.NoError(t, err) + assert.Nil(t, claim) + claim, err = store.ClaimMarketplaceListingSlot(ctx, "no-such-ri", "tok-8", []byte(claimTestSchedule)) + require.NoError(t, err) + assert.Nil(t, claim) +} diff --git a/internal/config/types.go b/internal/config/types.go index e1e2161a9..29c3d258e 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -930,6 +930,24 @@ type PurchaseHistoryRecord struct { ListingState string `json:"listing_state,omitempty" dynamodbav:"listing_state,omitempty"` } +// MarketplaceListingClaim is what ClaimMarketplaceListingSlot reads back from +// the row it just claimed, so the handler works from the row as it is at claim +// time instead of a copy read before the claim (issue #525). +type MarketplaceListingClaim struct { + // ListingID is the listing recorded on the row when it was claimed. + ListingID string + // PriorState is the listing_state before the claim; empty when none. + PriorState string + // ClientToken is the token of this attempt: the one passed to the claim, or + // the persisted one when an unresolved attempt already held a token. + ClientToken string + // PriceSchedule is the JSON price schedule stored with ClientToken. + PriceSchedule []byte + // Resumed reports that the token and schedule were already persisted by an + // earlier attempt that never recorded a listing. + Resumed bool +} + // AWS EC2 ReservedInstancesListing status values, mirroring the // ec2types.ListingStatus enum. Stored verbatim in purchase_history.listing_state // so the strings must match AWS exactly. ListingStatePending is additionally diff --git a/internal/database/postgres/migrations/000105_purchase_history_listing_attempt.down.sql b/internal/database/postgres/migrations/000105_purchase_history_listing_attempt.down.sql new file mode 100644 index 000000000..44f72d0e8 --- /dev/null +++ b/internal/database/postgres/migrations/000105_purchase_history_listing_attempt.down.sql @@ -0,0 +1,4 @@ +-- Revert migration 000105 +ALTER TABLE purchase_history + DROP COLUMN IF EXISTS listing_client_token, + DROP COLUMN IF EXISTS listing_price_schedule; diff --git a/internal/database/postgres/migrations/000105_purchase_history_listing_attempt.up.sql b/internal/database/postgres/migrations/000105_purchase_history_listing_attempt.up.sql new file mode 100644 index 000000000..d4378bd8f --- /dev/null +++ b/internal/database/postgres/migrations/000105_purchase_history_listing_attempt.up.sql @@ -0,0 +1,17 @@ +-- Migration 000105: persist the in-flight marketplace listing attempt (issue #525). +-- +-- listing_client_token: the CreateReservedInstancesListing ClientToken of an +-- attempt that has not been resolved. It is written in the same UPDATE that +-- claims the listing slot, before the AWS call, so a retry after an ambiguous +-- AWS error or a crash reuses it and AWS returns the existing listing instead +-- of creating a duplicate. +-- listing_price_schedule: the price schedule that token was sent with. The +-- default schedule depends on the clock, and AWS rejects a reused token with +-- different parameters, so a retry must resend the stored schedule. +-- +-- Both columns are NULL when no attempt is unresolved (legacy rows included) +-- and are cleared when a listing is recorded or the attempt is abandoned. + +ALTER TABLE purchase_history + ADD COLUMN IF NOT EXISTS listing_client_token TEXT, + ADD COLUMN IF NOT EXISTS listing_price_schedule JSONB; diff --git a/internal/database/postgres/migrations/000105_purchase_history_listing_attempt_test.go b/internal/database/postgres/migrations/000105_purchase_history_listing_attempt_test.go new file mode 100644 index 000000000..baee477c1 --- /dev/null +++ b/internal/database/postgres/migrations/000105_purchase_history_listing_attempt_test.go @@ -0,0 +1,53 @@ +//go:build integration + +package migrations_test + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Migration 000105 adds the nullable listing attempt columns to +// purchase_history: existing rows have no unresolved attempt, and the down +// migration removes both columns. +func TestMigration_PurchaseHistoryListingAttempt(t *testing.T) { + ctx := context.Background() + migrationsPath := getMigrationsPath() + + container, err := testhelpers.SetupPostgresContainer(ctx, t) + require.NoError(t, err) + defer container.Cleanup(ctx) + pool := container.DB.Pool() + + const insertSQL = ` + INSERT INTO purchase_history ( + account_id, purchase_id, timestamp, provider, service, region, + resource_type, term, payment + ) VALUES ('acct', 'ri-105', $1, 'aws', 'ec2', 'us-east-1', 't3.micro', 12, 'All Upfront')` + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 105))) + _, err = pool.Exec(ctx, insertSQL, time.Now()) + require.NoError(t, err) + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 105)) + var token, schedule *string + require.NoError(t, pool.QueryRow(ctx, + `SELECT listing_client_token, listing_price_schedule::text FROM purchase_history WHERE purchase_id = 'ri-105'`). + Scan(&token, &schedule)) + assert.Nil(t, token, "an existing row has no unresolved attempt") + assert.Nil(t, schedule) + + _, err = pool.Exec(ctx, + `UPDATE purchase_history SET listing_client_token = 'tok', listing_price_schedule = '[{"term_months":12,"price":1}]' WHERE purchase_id = 'ri-105'`) + require.NoError(t, err, "both columns must be writable") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 105))) + err = pool.QueryRow(ctx, `SELECT listing_client_token FROM purchase_history WHERE purchase_id = 'ri-105'`).Scan(&token) + require.Error(t, err, "the down migration must drop the column") +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index cc6d388a5..6a2923f6a 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -619,10 +619,18 @@ func (m *MockConfigStore) StampOfferingClass(ctx context.Context, purchaseID, of } // ClaimMarketplaceListingSlot mocks the atomic listing-slot claim (issue #292). -func (m *MockConfigStore) ClaimMarketplaceListingSlot(ctx context.Context, purchaseID string) (bool, error) { - m.record("ClaimMarketplaceListingSlot", ctx, purchaseID) - args := m.Called(ctx, purchaseID) - return args.Bool(0), args.Error(1) +func (m *MockConfigStore) ClaimMarketplaceListingSlot(ctx context.Context, purchaseID, clientToken string, priceSchedule []byte) (*config.MarketplaceListingClaim, error) { + m.record("ClaimMarketplaceListingSlot", ctx, purchaseID, clientToken, priceSchedule) + args := m.Called(ctx, purchaseID, clientToken, priceSchedule) + claim, _ := args.Get(0).(*config.MarketplaceListingClaim) + return claim, args.Error(1) +} + +// ReleaseMarketplaceListingClaim mocks releasing a pending listing claim (issue #525). +func (m *MockConfigStore) ReleaseMarketplaceListingClaim(ctx context.Context, purchaseID, priorState string, keepAttempt bool) error { + m.record("ReleaseMarketplaceListingClaim", ctx, purchaseID, priorState, keepAttempt) + args := m.Called(ctx, purchaseID, priorState, keepAttempt) + return args.Error(0) } // ClaimRIExchangeIdempotencyKey mocks the atomic RI exchange submit claim diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 0afc4e080..09873fa7b 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -406,8 +406,12 @@ func (m *mockConfigStoreForHealth) UpdateGlobalConfigAtomic(_ context.Context, a func (m *mockConfigStoreForHealth) UpdatePurchaseHistoryListing(_ context.Context, _, _, _ string) error { return nil } -func (m *mockConfigStoreForHealth) ClaimMarketplaceListingSlot(_ context.Context, _ string) (bool, error) { - return true, nil +func (m *mockConfigStoreForHealth) ClaimMarketplaceListingSlot(_ context.Context, _, _ string, _ []byte) (*config.MarketplaceListingClaim, error) { + return &config.MarketplaceListingClaim{}, nil +} + +func (m *mockConfigStoreForHealth) ReleaseMarketplaceListingClaim(_ context.Context, _, _ string, _ bool) error { + return nil } func (m *mockConfigStoreForHealth) ClaimRIExchangeIdempotencyKey(_ context.Context, _ string, _ time.Duration) (bool, error) { From c67bbb5462d57c792ed5bccdc505d353cba18967 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 6 Oct 2026 21:38:30 +0200 Subject: [PATCH 2/2] fix(marketplace): make the claim race-safe and fence the release A claim that raced another one derived Resumed and the prior state from a stale snapshot, so it skipped the reconcile and sent its own schedule with the first request's token. Lock the row first and derive Resumed from the returned token. Fence the release on the persisted token so a request only drops the attempt it owns. Also fix the lint findings (errcheck in the mock, rangeValCopy, gosec G117 on the token field). Refs #525 --- internal/analytics/collector_test.go | 2 +- internal/api/handler_marketplace_attempt.go | 5 +- .../api/handler_marketplace_attempt_test.go | 4 +- .../handler_marketplace_credentials_test.go | 2 +- internal/api/handler_marketplace_test.go | 8 +- internal/config/interfaces.go | 10 ++- internal/config/store_postgres.go | 23 ++++-- ...gres_marketplace_claim_integration_test.go | 80 ++++++++++++++++++- internal/config/types.go | 9 ++- internal/mocks/stores.go | 15 +++- internal/server/test_helpers_test.go | 2 +- 11 files changed, 127 insertions(+), 33 deletions(-) diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index da30c67ca..58685535d 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -479,7 +479,7 @@ func (m *mockConfigStore) ClaimMarketplaceListingSlot(_ context.Context, _, _ st return &config.MarketplaceListingClaim{}, nil } -func (m *mockConfigStore) ReleaseMarketplaceListingClaim(_ context.Context, _, _ string, _ bool) error { +func (m *mockConfigStore) ReleaseMarketplaceListingClaim(_ context.Context, _, _, _ string, _ bool) error { return nil } diff --git a/internal/api/handler_marketplace_attempt.go b/internal/api/handler_marketplace_attempt.go index 352a59030..822520648 100644 --- a/internal/api/handler_marketplace_attempt.go +++ b/internal/api/handler_marketplace_attempt.go @@ -43,7 +43,8 @@ func (f awsMarketplaceListingFinder) FindMarketplaceListingByToken(ctx context.C if err != nil { return ec2svc.MarketplaceListingResult{}, false, fmt.Errorf("DescribeReservedInstancesListings failed: %w", err) } - for _, l := range out.ReservedInstancesListings { + for i := range out.ReservedInstancesListings { + l := &out.ReservedInstancesListings[i] if aws.ToString(l.ClientToken) == clientToken { return ec2svc.MarketplaceListingResult{ListingID: aws.ToString(l.ReservedInstancesListingId), State: string(l.Status)}, true, nil } @@ -243,7 +244,7 @@ func (h *Handler) releaseMarketplaceClaim(ctx context.Context, purchaseID string if strings.EqualFold(prior, config.ListingStatePending) { prior = "" } - if err := h.config.ReleaseMarketplaceListingClaim(ctx, purchaseID, prior, keepAttempt); err != nil { + if err := h.config.ReleaseMarketplaceListingClaim(ctx, purchaseID, claim.ClientToken, prior, keepAttempt); err != nil { logging.Errorf("marketplace: failed to release listing claim for purchase %s (row may be stuck in %q): %v", purchaseID, config.ListingStatePending, err) } } diff --git a/internal/api/handler_marketplace_attempt_test.go b/internal/api/handler_marketplace_attempt_test.go index d3a124091..385d1c544 100644 --- a/internal/api/handler_marketplace_attempt_test.go +++ b/internal/api/handler_marketplace_attempt_test.go @@ -62,8 +62,8 @@ func (f *fakeListingDB) ClaimMarketplaceListingSlot(_ context.Context, _, client return &config.MarketplaceListingClaim{ListingID: f.row.ListingID, PriorState: prior, ClientToken: f.token, PriceSchedule: f.schedule, Resumed: resumed}, nil } -func (f *fakeListingDB) ReleaseMarketplaceListingClaim(_ context.Context, _, priorState string, keepAttempt bool) error { - if strings.EqualFold(f.row.ListingState, config.ListingStatePending) { +func (f *fakeListingDB) ReleaseMarketplaceListingClaim(_ context.Context, _, clientToken, priorState string, keepAttempt bool) error { + if strings.EqualFold(f.row.ListingState, config.ListingStatePending) && f.token == clientToken { f.row.ListingState = priorState if !keepAttempt { f.token, f.schedule = "", nil diff --git a/internal/api/handler_marketplace_credentials_test.go b/internal/api/handler_marketplace_credentials_test.go index ce5faa811..a5a70db43 100644 --- a/internal/api/handler_marketplace_credentials_test.go +++ b/internal/api/handler_marketplace_credentials_test.go @@ -55,7 +55,7 @@ func marketplaceAccountHandler(t *testing.T, account *config.CloudAccount, row * cfgStore.On("GetCloudAccount", mock.Anything, "acct-1").Return(account, nil) cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything).Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil).Maybe() cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything).Return(nil).Maybe() - cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything).Return(nil).Maybe() + cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() var built aws.Config h := &Handler{ diff --git a/internal/api/handler_marketplace_test.go b/internal/api/handler_marketplace_test.go index e85f20eaf..6f877f572 100644 --- a/internal/api/handler_marketplace_test.go +++ b/internal/api/handler_marketplace_test.go @@ -244,7 +244,7 @@ func TestMarketplaceList_AWSClientFaultMapsTo400(t *testing.T) { cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) // AWS create fails, so the claim must be released back to the unlisted state. - cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, "", false). + cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, "tok", "", false). Return(nil) ec2 := &stubMarketplaceEC2{ @@ -274,7 +274,7 @@ func TestMarketplaceList_AWSServerFaultMapsTo502(t *testing.T) { cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) // AWS create fails, so the claim must be released back to the unlisted state. - cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, "", true). + cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, "tok", "", true). Return(nil) ec2 := &stubMarketplaceEC2{ @@ -303,7 +303,7 @@ func TestMarketplaceList_UnknownErrorMapsTo502(t *testing.T) { cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything). Return(&config.MarketplaceListingClaim{ClientToken: "tok"}, nil) // AWS create fails, so the claim must be released back to the unlisted state. - cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, "", true). + cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, "tok", "", true). Return(nil) ec2 := &stubMarketplaceEC2{ @@ -388,7 +388,7 @@ func TestMarketplaceList_DBFailureAndCancelFailureKeepsListing(t *testing.T) { assert.Contains(t, ce.message, "throttled") assert.Contains(t, ce.message, tc.wantMsg) assert.Equal(t, 1, ec2.cancelCallCount) - cfgStore.AssertNotCalled(t, "ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything) + cfgStore.AssertNotCalled(t, "ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, mock.Anything, mock.Anything, mock.Anything) cfgStore.AssertExpectations(t) }) } diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 14f62aa93..ed920eb6c 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -333,10 +333,12 @@ type StoreInterface interface { ClaimMarketplaceListingSlot(ctx context.Context, purchaseID, clientToken string, priceSchedule []byte) (*MarketplaceListingClaim, error) // ReleaseMarketplaceListingClaim moves a pending row back to priorState - // (empty for none) after a failed attempt. keepAttempt keeps the persisted - // token and schedule so the retry resends them (ambiguous AWS outcome); - // otherwise they are cleared because AWS created nothing reusable. - ReleaseMarketplaceListingClaim(ctx context.Context, purchaseID, priorState string, keepAttempt bool) error + // (empty for none) after a failed attempt. It only acts on a row whose + // persisted token is clientToken, so a request never releases or drops the + // attempt of another request. keepAttempt keeps the persisted token and + // schedule so the retry resends them (ambiguous AWS outcome); otherwise + // they are cleared because AWS created nothing reusable. + ReleaseMarketplaceListingClaim(ctx context.Context, purchaseID, clientToken, priorState string, keepAttempt bool) error // ClaimRIExchangeIdempotencyKey atomically claims key for an RI exchange // submit, so a client that retries a timed-out execute request cannot diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 1554c0119..bd94c2475 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -2093,25 +2093,29 @@ func (s *PostgresStore) StampOfferingClass(ctx context.Context, purchaseID, offe // must not trust a row read before the claim. It returns (nil, nil) when the row // could not be claimed (active, pending without a token, or absent). func (s *PostgresStore) ClaimMarketplaceListingSlot(ctx context.Context, purchaseID, clientToken string, priceSchedule []byte) (*MarketplaceListingClaim, error) { + // prev locks the row first, so a claim that races another one waits for it + // and then sees its committed token and state instead of a stale snapshot. query := ` + WITH prev AS ( + SELECT purchase_id, listing_state AS prior_state + FROM purchase_history WHERE purchase_id = $2 FOR UPDATE + ) UPDATE purchase_history AS p SET listing_state = $1, listing_client_token = COALESCE(p.listing_client_token, $3::text), listing_price_schedule = CASE WHEN p.listing_client_token IS NULL THEN $4::jsonb ELSE p.listing_price_schedule END - FROM (SELECT purchase_id, listing_state AS prior_state, - listing_client_token IS NOT NULL AS had_token - FROM purchase_history WHERE purchase_id = $2) AS prev + FROM prev WHERE p.purchase_id = prev.purchase_id AND lower(COALESCE(p.listing_state, '')) <> $5 AND (lower(COALESCE(p.listing_state, '')) <> $1 OR p.listing_client_token IS NOT NULL) RETURNING COALESCE(p.listing_id, ''), COALESCE(prev.prior_state, ''), - p.listing_client_token, p.listing_price_schedule::text, prev.had_token + p.listing_client_token, p.listing_price_schedule::text ` var claim MarketplaceListingClaim var schedule string err := s.db.QueryRow(ctx, query, ListingStatePending, purchaseID, clientToken, priceSchedule, ListingStateActive). - Scan(&claim.ListingID, &claim.PriorState, &claim.ClientToken, &schedule, &claim.Resumed) + Scan(&claim.ListingID, &claim.PriorState, &claim.ClientToken, &schedule) if errors.Is(err, pgx.ErrNoRows) { return nil, nil } @@ -2119,21 +2123,24 @@ func (s *PostgresStore) ClaimMarketplaceListingSlot(ctx context.Context, purchas return nil, fmt.Errorf("failed to claim marketplace listing slot for purchase %s: %w", purchaseID, err) } claim.PriceSchedule = []byte(schedule) + claim.Resumed = claim.ClientToken != clientToken return &claim, nil } // ReleaseMarketplaceListingClaim moves a pending row back to priorState and, // unless keepAttempt, clears the persisted token and schedule. It only touches -// a row that is still pending so it never overwrites a recorded listing. -func (s *PostgresStore) ReleaseMarketplaceListingClaim(ctx context.Context, purchaseID, priorState string, keepAttempt bool) error { +// a pending row whose token is clientToken, so it never overwrites a recorded +// listing or the attempt of another request. +func (s *PostgresStore) ReleaseMarketplaceListingClaim(ctx context.Context, purchaseID, clientToken, priorState string, keepAttempt bool) error { query := ` UPDATE purchase_history SET listing_state = NULLIF($2, ''), listing_client_token = CASE WHEN $3 THEN listing_client_token END, listing_price_schedule = CASE WHEN $3 THEN listing_price_schedule END WHERE purchase_id = $1 AND lower(COALESCE(listing_state, '')) = $4 + AND listing_client_token = $5 ` - if _, err := s.db.Exec(ctx, query, purchaseID, priorState, keepAttempt, ListingStatePending); err != nil { + if _, err := s.db.Exec(ctx, query, purchaseID, priorState, keepAttempt, ListingStatePending, clientToken); err != nil { return fmt.Errorf("failed to release marketplace listing claim for purchase %s: %w", purchaseID, err) } return nil diff --git a/internal/config/store_postgres_marketplace_claim_integration_test.go b/internal/config/store_postgres_marketplace_claim_integration_test.go index 9e97a4e9e..6c07c791b 100644 --- a/internal/config/store_postgres_marketplace_claim_integration_test.go +++ b/internal/config/store_postgres_marketplace_claim_integration_test.go @@ -14,6 +14,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/jackc/pgx/v5/pgxpool" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -26,7 +27,78 @@ func TestMarketplaceListingClaim_PostgresSemantics(t *testing.T) { t.Cleanup(func() { assert.NoError(t, container.Cleanup(context.Background())) }) require.NoError(t, migrations.RunMigrations(ctx, container.DB.Pool(), getTestMigrationsPath(), "", "")) - checkMarketplaceListingClaim(ctx, t, NewPostgresStore(container.DB)) + store := NewPostgresStore(container.DB) + checkMarketplaceListingClaim(ctx, t, store) + checkConcurrentMarketplaceClaim(ctx, t, store, container.DB.Pool()) + checkMarketplaceReleaseIsFenced(ctx, t, store) +} + +func seedClaimRow(ctx context.Context, t *testing.T, store *PostgresStore, id string) { + t.Helper() + require.NoError(t, store.SavePurchaseHistory(ctx, &PurchaseHistoryRecord{ + AccountID: "acct", PurchaseID: id, Timestamp: time.Now(), Provider: "aws", Service: "ec2", + Region: "us-east-1", ResourceType: "t3.micro", Count: 1, Term: 3, Payment: "All Upfront", + })) +} + +// A claim that races another one waits for its row lock and must come back as +// a resumed claim carrying the first claim's token and schedule. The first +// claim is held in an open transaction, as it is while its statement commits. +func checkConcurrentMarketplaceClaim(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + t.Helper() + const id = "ri-race-525" + seedClaimRow(ctx, t, store, id) + + tx, err := pool.Begin(ctx) + require.NoError(t, err) + _, err = tx.Exec(ctx, `UPDATE purchase_history SET listing_state = 'pending', listing_client_token = 'tok-A', + listing_price_schedule = '[{"term_months":1,"price":1}]' WHERE purchase_id = $1`, id) + require.NoError(t, err) + + type result struct { + claim *MarketplaceListingClaim + err error + } + done := make(chan result, 1) + go func() { + claim, claimErr := store.ClaimMarketplaceListingSlot(ctx, id, "tok-B", []byte(`[{"term_months":2,"price":2}]`)) + done <- result{claim, claimErr} + }() + + require.Eventually(t, func() bool { + var waiting int + require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM pg_stat_activity WHERE wait_event_type = 'Lock' AND query LIKE '%listing_client_token%'`).Scan(&waiting)) + return waiting > 0 + }, 10*time.Second, 20*time.Millisecond, "the second claim must be blocked on the first claim's row lock") + require.NoError(t, tx.Commit(ctx)) + + got := <-done + require.NoError(t, got.err) + require.NotNil(t, got.claim) + assert.True(t, got.claim.Resumed, "the second claim is a resumed one") + assert.Equal(t, "tok-A", got.claim.ClientToken) + assert.JSONEq(t, `[{"term_months":1,"price":1}]`, string(got.claim.PriceSchedule)) + assert.Equal(t, ListingStatePending, got.claim.PriorState) +} + +// A request only releases or drops the attempt it owns. +func checkMarketplaceReleaseIsFenced(ctx context.Context, t *testing.T, store *PostgresStore) { + t.Helper() + const id = "ri-fence-525" + seedClaimRow(ctx, t, store, id) + claim, err := store.ClaimMarketplaceListingSlot(ctx, id, "tok-owner", []byte(claimTestSchedule)) + require.NoError(t, err) + require.NotNil(t, claim) + + require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, "tok-stale", "", false)) + require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, "tok-stale", "", true)) + got, err := store.GetPurchaseHistoryByPurchaseID(ctx, id) + require.NoError(t, err) + assert.Equal(t, ListingStatePending, got.ListingState, "a stale release must not touch the row") + resumed, err := store.ClaimMarketplaceListingSlot(ctx, id, "tok-next", []byte(claimTestSchedule)) + require.NoError(t, err) + require.NotNil(t, resumed) + assert.Equal(t, "tok-owner", resumed.ClientToken, "a stale drop must not erase the owner's token") } func checkMarketplaceListingClaim(ctx context.Context, t *testing.T, store *PostgresStore) { @@ -64,7 +136,7 @@ func checkMarketplaceListingClaim(ctx context.Context, t *testing.T, store *Post assert.Equal(t, ListingStatePending, claim.PriorState) // Release keeping the attempt: the state returns, the token stays. - require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, "", true)) + require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, "tok-1", "", true)) _, state = row() assert.Empty(t, state) claim, err = store.ClaimMarketplaceListingSlot(ctx, id, "tok-3", []byte(claimTestSchedule)) @@ -74,7 +146,7 @@ func checkMarketplaceListingClaim(ctx context.Context, t *testing.T, store *Post assert.Equal(t, "tok-1", claim.ClientToken) // Release dropping the attempt: the next claim starts with its own token. - require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, "", false)) + require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, "tok-1", "", false)) claim, err = store.ClaimMarketplaceListingSlot(ctx, id, "tok-4", []byte(claimTestSchedule)) require.NoError(t, err) require.NotNil(t, claim) @@ -97,7 +169,7 @@ func checkMarketplaceListingClaim(ctx context.Context, t *testing.T, store *Post assert.Equal(t, "ril-1", claim.ListingID) assert.Equal(t, ListingStateCancelled, claim.PriorState) require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-2", ListingStateActive)) - require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, ListingStateCancelled, false)) + require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, "tok-6", ListingStateCancelled, false)) listingID, state := row() assert.Equal(t, "ril-2", listingID) assert.Equal(t, ListingStateActive, state, "release must not overwrite a recorded listing") diff --git a/internal/config/types.go b/internal/config/types.go index 29c3d258e..a148d61f9 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -940,11 +940,16 @@ type MarketplaceListingClaim struct { PriorState string // ClientToken is the token of this attempt: the one passed to the claim, or // the persisted one when an unresolved attempt already held a token. - ClientToken string + ClientToken string //nolint:gosec // G117: an idempotency token for AWS, not a credential // PriceSchedule is the JSON price schedule stored with ClientToken. PriceSchedule []byte // Resumed reports that the token and schedule were already persisted by an - // earlier attempt that never recorded a listing. + // earlier attempt that never recorded a listing. It is derived from the + // returned row (its token differs from the one this claim tried to write), + // so it stays correct when two claims race. PriorState is then "pending", + // the state the earlier attempt left; releasing restores an empty state + // and the row keeps its recorded listing id, which every list, cancel and + // claim check treats the same as the original dead state. Resumed bool } diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 6a2923f6a..a51b56151 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -622,14 +622,21 @@ func (m *MockConfigStore) StampOfferingClass(ctx context.Context, purchaseID, of func (m *MockConfigStore) ClaimMarketplaceListingSlot(ctx context.Context, purchaseID, clientToken string, priceSchedule []byte) (*config.MarketplaceListingClaim, error) { m.record("ClaimMarketplaceListingSlot", ctx, purchaseID, clientToken, priceSchedule) args := m.Called(ctx, purchaseID, clientToken, priceSchedule) - claim, _ := args.Get(0).(*config.MarketplaceListingClaim) + var claim *config.MarketplaceListingClaim + if v := args.Get(0); v != nil { + c, ok := v.(*config.MarketplaceListingClaim) + if !ok { + panic(fmt.Sprintf("mock: expected *config.MarketplaceListingClaim, got %T", v)) + } + claim = c + } return claim, args.Error(1) } // ReleaseMarketplaceListingClaim mocks releasing a pending listing claim (issue #525). -func (m *MockConfigStore) ReleaseMarketplaceListingClaim(ctx context.Context, purchaseID, priorState string, keepAttempt bool) error { - m.record("ReleaseMarketplaceListingClaim", ctx, purchaseID, priorState, keepAttempt) - args := m.Called(ctx, purchaseID, priorState, keepAttempt) +func (m *MockConfigStore) ReleaseMarketplaceListingClaim(ctx context.Context, purchaseID, clientToken, priorState string, keepAttempt bool) error { + m.record("ReleaseMarketplaceListingClaim", ctx, purchaseID, clientToken, priorState, keepAttempt) + args := m.Called(ctx, purchaseID, clientToken, priorState, keepAttempt) return args.Error(0) } diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 09873fa7b..faa1defc1 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -410,7 +410,7 @@ func (m *mockConfigStoreForHealth) ClaimMarketplaceListingSlot(_ context.Context return &config.MarketplaceListingClaim{}, nil } -func (m *mockConfigStoreForHealth) ReleaseMarketplaceListingClaim(_ context.Context, _, _ string, _ bool) error { +func (m *mockConfigStoreForHealth) ReleaseMarketplaceListingClaim(_ context.Context, _, _, _ string, _ bool) error { return nil }