diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 6870feb9e..58685535d 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..822520648 --- /dev/null +++ b/internal/api/handler_marketplace_attempt.go @@ -0,0 +1,250 @@ +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 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 + } + } + 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, 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 new file mode 100644 index 000000000..385d1c544 --- /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, _, 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 + } + } + 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..a5a70db43 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, 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..6f877f572 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, "tok", "", 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, "tok", "", 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, "tok", "", 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, 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..ed920eb6c 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,24 @@ 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. 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 63c284904..bd94c2475 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,64 @@ 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) { + // 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 := ` - UPDATE purchase_history - SET listing_state = $1 - WHERE purchase_id = $2 - AND lower(COALESCE(listing_state, '')) NOT IN ($3, $4) + 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 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 ` - 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) + 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) + 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 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, clientToken); 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..6c07c791b --- /dev/null +++ b/internal/config/store_postgres_marketplace_claim_integration_test.go @@ -0,0 +1,185 @@ +//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/jackc/pgx/v5/pgxpool" + "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(), "", "")) + + 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) { + 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, "tok-1", "", 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, "tok-1", "", 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, "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") + + // 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..a148d61f9 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -930,6 +930,29 @@ 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 //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. 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 +} + // 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..a51b56151 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -619,10 +619,25 @@ 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) + 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, 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) } // 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..faa1defc1 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) {