From 013dd5d521624f23df876871f5f84d3f906cdac8 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 23 Oct 2026 07:58:38 +0000 Subject: [PATCH 1/2] fix(marketplace): recover pending listings through cancel and never record an empty or revived state The cancel endpoint only accepted active rows, so a row left pending by a list request that crashed between the AWS create and the DB write, a listing recorded in AWS's pending state, or a pre-000105 claim stayed stuck. Cancel now recovers pending rows: it finds the attempt's listing on AWS by the persisted ClientToken and cancels it, cancels a recorded pending listing, frees a pending attempt AWS has nothing for (keeping its token so the next list resumes it), and clears a legacy claim. Its writes are fenced on the row it read (ResolveMarketplaceListingAttempt). UpdatePurchaseHistoryListing no longer marks a listing recorded as canceled or closed live again (ErrMarketplaceListingEnded), so a list request that records its listing after a concurrent cancel gets a 409 instead of reviving it. A cancel retried after its DB write failed reads the listing back and records the ended state instead of failing on AWS's error; the cancel's write runs on a detached context. An empty AWS status is never recorded: the created listing is read back, else recorded as pending. The rollback release runs on the detached compensation context. No migration. Closes #525 Co-Authored-By: Claude Opus 5.5 --- internal/analytics/collector_test.go | 8 + internal/api/handler_marketplace.go | 59 +-- internal/api/handler_marketplace_attempt.go | 44 +- .../api/handler_marketplace_attempt_test.go | 9 +- internal/api/handler_marketplace_cancel.go | 211 +++++++++ ...rketplace_compensation_integration_test.go | 5 + ...er_marketplace_recover_integration_test.go | 203 +++++++++ .../api/handler_marketplace_recover_test.go | 421 ++++++++++++++++++ internal/api/handler_marketplace_test.go | 16 +- internal/config/interfaces.go | 16 +- internal/config/store_postgres.go | 10 +- .../store_postgres_marketplace_attempt.go | 97 ++++ ...es_marketplace_attempt_integration_test.go | 237 ++++++++++ internal/mocks/stores.go | 22 + internal/server/test_helpers_test.go | 6 + 15 files changed, 1301 insertions(+), 63 deletions(-) create mode 100644 internal/api/handler_marketplace_cancel.go create mode 100644 internal/api/handler_marketplace_recover_integration_test.go create mode 100644 internal/api/handler_marketplace_recover_test.go create mode 100644 internal/config/store_postgres_marketplace_attempt.go create mode 100644 internal/config/store_postgres_marketplace_attempt_integration_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 76ec0b712..1cee0fe40 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -528,6 +528,14 @@ func (m *mockConfigStore) UpdatePurchaseHistoryListing(_ context.Context, _, _, return nil } +func (m *mockConfigStore) GetMarketplaceListingAttempt(_ context.Context, _ string) (*config.MarketplaceListingAttempt, error) { + return nil, nil +} + +func (m *mockConfigStore) ResolveMarketplaceListingAttempt(_ context.Context, _ string, _ config.MarketplaceListingAttempt, _, _ string) (bool, error) { + return false, nil +} + func (m *mockConfigStore) ClaimMarketplaceListingSlot(_ context.Context, _, _ string, _ []byte) (*config.MarketplaceListingClaim, error) { return &config.MarketplaceListingClaim{}, nil } diff --git a/internal/api/handler_marketplace.go b/internal/api/handler_marketplace.go index fc01bdb95..cded31ab4 100644 --- a/internal/api/handler_marketplace.go +++ b/internal/api/handler_marketplace.go @@ -5,7 +5,8 @@ package api // // Endpoints: // POST /api/purchases/{id}/marketplace-list create a listing -// POST /api/purchases/{id}/marketplace-cancel cancel an active listing +// POST /api/purchases/{id}/marketplace-cancel cancel an active listing or +// recover a pending one // // Both endpoints require the caller to hold sell-any:purchases (admin or a // custom operator group) or sell-own:purchases for rows they purchased @@ -315,6 +316,9 @@ func (h *Handler) keepUncanceledListing(ctx context.Context, purchaseID string, } msg := fmt.Sprintf("marketplace listing %s was created but could not be saved, and canceling it failed (%s); the listing may still be active on AWS", listing.ListingID, cancelDetail) if err := h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, listing.ListingID, state); err != nil { + if isListingEndedConflict(err) { + return listingCanceledConcurrently(listing.ListingID) + } logging.Errorf("marketplace: failed to record uncanceled listing %s for purchase %s (row stays %q): %v", listing.ListingID, purchaseID, config.ListingStatePending, err) return NewClientError(502, msg+"; the RI stays locked as pending until an operator reconciles the listing") } @@ -347,59 +351,6 @@ func (h *Handler) populateOfferingClass(ctx context.Context, purchaseID string, return class, nil } -// 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) - if err != nil { - return nil, err - } - - err = validateUUID(purchaseID) - if err != nil { - return nil, err - } - - row, err := h.config.GetPurchaseHistoryByPurchaseID(ctx, purchaseID) - if err != nil { - return nil, fmt.Errorf("failed to look up purchase: %w", err) - } - if row == nil { - return nil, NewClientError(404, "purchase not found") - } - - err = h.authorizeSessionSell(ctx, session, action, row) - if err != nil { - return nil, err - } - - if !strings.EqualFold(row.ListingState, config.ListingStateActive) { - return nil, NewClientError(409, "no active listing found for this RI; current state: "+row.ListingState) - } - - cfg, err := h.loadMarketplaceAWSConfig(ctx, row) - if err != nil { - return nil, err - } - - ec2Client := h.buildMarketplaceEC2Client(cfg) - - result, err := ec2Client.CancelMarketplaceListing(ctx, row.ListingID) - if err != nil { - logging.Warnf("marketplace: CancelReservedInstancesListing for listing %s failed: %v", row.ListingID, err) - return nil, mapAWSMarketplaceError("AWS cancel listing failed", err) - } - - // The listing is already canceled in AWS; there is no compensating rollback - // available if the DB write fails. Return an internal error so the caller - // knows the state is out of sync and can retry or contact an administrator. - if dbErr := h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, result.ListingID, result.State); dbErr != nil { - logging.Errorf("marketplace: listing canceled in AWS (%s) but DB update failed: %v; state is out of sync", result.ListingID, dbErr) - return nil, fmt.Errorf("listing canceled in AWS but could not be persisted: %w", dbErr) - } - - return map[string]string{"listing_id": result.ListingID, "listing_state": result.State}, nil -} - // sell-own uses account access rather than creator identity for purchase-history rows. func (h *Handler) authorizeSessionSell(ctx context.Context, session *Session, action string, row *config.PurchaseHistoryRecord) error { scope, err := h.getAccountScope(ctx, session) diff --git a/internal/api/handler_marketplace_attempt.go b/internal/api/handler_marketplace_attempt.go index 822520648..453acfe4f 100644 --- a/internal/api/handler_marketplace_attempt.go +++ b/internal/api/handler_marketplace_attempt.go @@ -153,17 +153,46 @@ func (h *Handler) reconcileAttempt(ctx context.Context, purchaseID string, claim if !found { return ec2svc.MarketplaceListingResult{}, false, nil } + if listing.State == "" { + listing.State = config.ListingStatePending + } 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 { + if isListingEndedConflict(err) { + return ec2svc.MarketplaceListingResult{}, true, listingCanceledConcurrently(listing.ListingID) + } 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 } +// listingCanceledConcurrently is the error for a listing the cancel endpoint +// canceled and recorded while this list request was creating or recording it. +func listingCanceledConcurrently(listingID string) error { + return NewClientError(409, fmt.Sprintf("marketplace listing %s was canceled by another request before this one could record it; list again to create a new one", listingID)) +} + +// createdListingState returns the state to record for a listing AWS just +// created. AWS has been seen to omit the status; an empty state would read as +// "not listed" and let the next request create a second listing, so the +// listing is read back, and recorded as pending (live, cancelable) if its +// state is still unknown. +func createdListingState(ctx context.Context, ec2Client marketplaceEC2Client, created ec2svc.MarketplaceListingResult) string { + if created.State != "" { + return created.State + } + described, err := ec2Client.DescribeMarketplaceListing(ctx, created.ListingID) + if err == nil && described.State != "" { + return described.State + } + logging.Warnf("marketplace: AWS reported no status for new listing %s (describe error: %v); recording it as %s", created.ListingID, err, config.ListingStatePending) + return config.ListingStatePending +} + 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, @@ -181,6 +210,7 @@ func (h *Handler) createAndPersistListing(ctx context.Context, purchaseID string return ec2svc.MarketplaceListingResult{}, mapped } + result.State = createdListingState(ctx, ec2Client, result) if isDeadListingState(result.State) { return ec2svc.MarketplaceListingResult{}, h.recordDeadListing(ctx, purchaseID, claim, result) } @@ -190,6 +220,11 @@ func (h *Handler) createAndPersistListing(ctx context.Context, purchaseID string // 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 { + if isListingEndedConflict(dbErr) { + // The cancel endpoint found this attempt's listing through its + // token, canceled it on AWS and recorded that first. + return ec2svc.MarketplaceListingResult{}, listingCanceledConcurrently(result.ListingID) + } 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() @@ -197,7 +232,7 @@ func (h *Handler) createAndPersistListing(ctx context.Context, purchaseID string 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) + h.recordRolledBackListing(compCtx, purchaseID, claim, result.ListingID) return ec2svc.MarketplaceListingResult{}, fmt.Errorf("listing created but could not be persisted; listing has been rolled back: %w", dbErr) } @@ -226,10 +261,13 @@ func (h *Handler) recordDeadListing(ctx context.Context, purchaseID string, clai // 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) { +// Both writes run on the detached compensation context: the listing is +// already canceled on AWS, so a caller that went away must not leave the row +// pending (issue #525). +func (h *Handler) recordRolledBackListing(compCtx 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) + h.releaseMarketplaceClaim(compCtx, purchaseID, claim, false) } } diff --git a/internal/api/handler_marketplace_attempt_test.go b/internal/api/handler_marketplace_attempt_test.go index 385d1c544..2be9bb1a1 100644 --- a/internal/api/handler_marketplace_attempt_test.go +++ b/internal/api/handler_marketplace_attempt_test.go @@ -77,6 +77,9 @@ func (f *fakeListingDB) UpdatePurchaseHistoryListing(_ context.Context, _, listi f.failRecords-- return errors.New("db down") } + if f.row.ListingID == listingID && isDeadListingState(f.row.ListingState) && !isDeadListingState(state) { + return config.ErrMarketplaceListingEnded + } f.row.ListingID, f.row.ListingState = listingID, state f.token, f.schedule = "", nil return nil @@ -137,6 +140,10 @@ func (a *fakeMarketplaceAWS) cancel(_ context.Context, id string) (ec2svc.Market } for _, l := range a.listings { if l.id == id { + if isDeadListingState(l.state) { + // Like AWS, a listing that already ended cannot be canceled. + return ec2svc.MarketplaceListingResult{}, &marketplaceTestAPIError{code: "IncorrectState", message: "listing " + id + " is " + l.state, fault: smithy.FaultClient} + } l.state = config.ListingStateCancelled } } @@ -183,7 +190,7 @@ func newMarketplaceFixture(t *testing.T) *marketplaceFixture { require.Equal(t, toAWSPriceSchedule(stored), req.PriceSchedule) return fake.create(ctx, req) } - stub := &stubMarketplaceEC2{createFn: create, cancelFn: fake.cancel} + stub := &stubMarketplaceEC2{createFn: create, cancelFn: fake.cancel, describeFn: fake.describe} h := newMarketplaceHandler(db.MockConfigStore, authSvc, stub) h.config = db h.marketplaceFinderFactory = func(aws.Config) marketplaceListingFinder { return fake } diff --git a/internal/api/handler_marketplace_cancel.go b/internal/api/handler_marketplace_cancel.go new file mode 100644 index 000000000..c000a3192 --- /dev/null +++ b/internal/api/handler_marketplace_cancel.go @@ -0,0 +1,211 @@ +package api + +// handler_marketplace_cancel.go implements POST +// /api/purchases/{id}/marketplace-cancel. Besides canceling an active listing +// it recovers a row that a marketplace-list attempt left pending (issue #525): +// a crash between the AWS create and the DB write, an AWS listing recorded in +// its own pending state, or a claim from before the attempt token was +// persisted. Every write after an AWS call is fenced on what this request +// read, so it never overwrites a listing or attempt it did not see, and runs +// on a detached context so a disconnecting caller cannot lose it. + +import ( + "context" + "errors" + "fmt" + "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/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/aws/aws-lambda-go/events" +) + +// marketplaceListingChangedMessage is the 409 for a fenced write that found the +// row changed after this request read it. +const marketplaceListingChangedMessage = "the marketplace listing of this RI changed while this request was handling it; reload and retry" + +// 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) + if err != nil { + return nil, err + } + err = validateUUID(purchaseID) + if err != nil { + return nil, err + } + row, err := h.config.GetPurchaseHistoryByPurchaseID(ctx, purchaseID) + if err != nil { + return nil, fmt.Errorf("failed to look up purchase: %w", err) + } + if row == nil { + return nil, NewClientError(404, "purchase not found") + } + err = h.authorizeSessionSell(ctx, session, action, row) + if err != nil { + return nil, err + } + + switch { + case strings.EqualFold(row.ListingState, config.ListingStatePending): + return h.recoverPendingListing(ctx, purchaseID, row) + case !strings.EqualFold(row.ListingState, config.ListingStateActive): + return nil, NewClientError(409, "no active or pending listing found for this RI; current state: "+row.ListingState) + } + ec2Client, err := h.marketplaceClientFor(ctx, row) + if err != nil { + return nil, err + } + return h.cancelListing(ctx, purchaseID, nil, row.ListingID, ec2Client) +} + +func (h *Handler) marketplaceClientFor(ctx context.Context, row *config.PurchaseHistoryRecord) (marketplaceEC2Client, error) { + cfg, err := h.loadMarketplaceAWSConfig(ctx, row) + if err != nil { + return nil, err + } + return h.buildMarketplaceEC2Client(cfg), nil +} + +// recoverPendingListing resolves a pending row. It reads the row's attempt +// token, which the purchase lookup does not return, and fences its writes on +// what it read. +func (h *Handler) recoverPendingListing(ctx context.Context, purchaseID string, row *config.PurchaseHistoryRecord) (any, error) { + seen, err := h.config.GetMarketplaceListingAttempt(ctx, purchaseID) + if err != nil { + return nil, fmt.Errorf("failed to look up the marketplace listing attempt: %w", err) + } + if seen == nil { + return nil, NewClientError(404, "purchase not found") + } + if !strings.EqualFold(seen.State, config.ListingStatePending) { + return nil, NewClientError(409, marketplaceListingChangedMessage) + } + if seen.ClientToken == "" && seen.ListingID == "" { + // A claim from before the attempt token was persisted (migration + // 000105) that never finished: nothing on AWS can be tied to it, so + // the stuck row is cleared. + logging.Warnf("marketplace: purchase %s: clearing a pending listing claim with no attempt token and no listing", purchaseID) + return h.resolveListingAttempt(ctx, purchaseID, *seen, "", "") + } + if seen.ClientToken == "" { + // A recorded listing in AWS's own pending state, or one recorded after + // its rollback cancel failed: cancel it like an active one. + ec2Client, clientErr := h.marketplaceClientFor(ctx, row) + if clientErr != nil { + return nil, clientErr + } + return h.cancelListing(ctx, purchaseID, seen, seen.ListingID, ec2Client) + } + return h.recoverPendingAttempt(ctx, purchaseID, row, *seen) +} + +// recoverPendingAttempt resolves a pending row that holds the token of a list +// attempt that never recorded its outcome: in flight, or crashed between the +// AWS create and the DB write. AWS is asked whether the attempt created a +// listing; a live one is canceled, an ended one recorded. +func (h *Handler) recoverPendingAttempt(ctx context.Context, purchaseID string, row *config.PurchaseHistoryRecord, seen config.MarketplaceListingAttempt) (any, error) { + cfg, err := h.loadMarketplaceAWSConfig(ctx, row) + if err != nil { + return nil, err + } + listing, found, err := h.buildMarketplaceFinder(cfg).FindMarketplaceListingByToken(ctx, purchaseID, seen.ClientToken) + if err != nil { + return nil, mapAWSMarketplaceError("could not look up the pending listing attempt on AWS", err) + } + if !found { + return h.releaseAbandonedAttempt(ctx, purchaseID, seen) + } + if isDeadListingState(listing.State) { + return h.resolveListingAttempt(ctx, purchaseID, seen, listing.ListingID, listing.State) + } + return h.cancelListing(ctx, purchaseID, &seen, listing.ListingID, h.buildMarketplaceEC2Client(cfg)) +} + +// releaseAbandonedAttempt frees a pending row whose attempt created nothing on +// AWS. The token is kept: a list request still in flight with it records its +// listing as usual, and the next list request resumes the same attempt, so +// AWS still deduplicates on the token and no second listing can appear. +func (h *Handler) releaseAbandonedAttempt(ctx context.Context, purchaseID string, seen config.MarketplaceListingAttempt) (any, error) { + if err := h.config.ReleaseMarketplaceListingClaim(ctx, purchaseID, seen.ClientToken, "", true); err != nil { + return nil, fmt.Errorf("failed to release the pending marketplace listing claim: %w", err) + } + logging.Warnf("marketplace: purchase %s: no AWS listing for the pending attempt; released the claim", purchaseID) + return map[string]string{ + "listing_id": "", + "listing_state": "", + "note": "AWS has no listing for the pending attempt, so nothing was canceled and the RI can be listed again", + }, nil +} + +// cancelListing cancels listingID on AWS and records the outcome. seen is the +// attempt a pending row was read with, nil for an active row. A cancel that +// fails because the listing already ended (a retry after the DB write of an +// earlier cancel failed) records the ended state instead. +func (h *Handler) cancelListing(ctx context.Context, purchaseID string, seen *config.MarketplaceListingAttempt, listingID string, ec2Client marketplaceEC2Client) (any, error) { + result, err := ec2Client.CancelMarketplaceListing(ctx, listingID) + if err != nil { + ended, ok := endedMarketplaceListing(ctx, ec2Client, listingID, err) + if !ok { + logging.Warnf("marketplace: CancelReservedInstancesListing for listing %s failed: %v", listingID, err) + return nil, mapAWSMarketplaceError("AWS cancel listing failed", err) + } + result = ended + } + state := result.State + if state == "" { + // The cancel call succeeded but AWS reported no status: never record + // an empty state, which would read as "not listed". + state = config.ListingStateCancelled + } + + // AWS has canceled the listing; there is no rollback, so record it on a + // detached context the caller disconnecting cannot abort. + recCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), marketplaceCompensationTimeout) + defer cancel() + if seen != nil { + return h.resolveListingAttempt(recCtx, purchaseID, *seen, listingID, state) + } + if dbErr := h.config.UpdatePurchaseHistoryListing(recCtx, purchaseID, listingID, state); dbErr != nil { + logging.Errorf("marketplace: listing canceled in AWS (%s) but DB update failed: %v; state is out of sync", listingID, dbErr) + return nil, fmt.Errorf("listing canceled in AWS but could not be persisted; retry the cancel to record it: %w", dbErr) + } + return map[string]string{"listing_id": listingID, "listing_state": state}, nil +} + +// endedMarketplaceListing reports whether listingID has already ended on AWS, +// read after its cancel failed. A role-assumption failure would fail the read +// the same way, so it is not retried. +func endedMarketplaceListing(ctx context.Context, ec2Client marketplaceEC2Client, listingID string, cancelErr error) (ec2svc.MarketplaceListingResult, bool) { + if isRoleAssumptionFailure(cancelErr) { + return ec2svc.MarketplaceListingResult{}, false + } + listing, err := ec2Client.DescribeMarketplaceListing(ctx, listingID) + if err != nil || !isDeadListingState(listing.State) { + return ec2svc.MarketplaceListingResult{}, false + } + logging.Warnf("marketplace: listing %s is already %s on AWS (cancel said: %v); recording it", listingID, listing.State, cancelErr) + return ec2svc.MarketplaceListingResult{ListingID: listingID, State: listing.State}, true +} + +// resolveListingAttempt records listingID in state on a pending row, only +// while the row is still as seen or already holds listingID. +func (h *Handler) resolveListingAttempt(ctx context.Context, purchaseID string, seen config.MarketplaceListingAttempt, listingID, state string) (any, error) { + applied, err := h.config.ResolveMarketplaceListingAttempt(ctx, purchaseID, seen, listingID, state) + if err != nil { + logging.Errorf("marketplace: failed to record listing %s (%s) for purchase %s: %v", listingID, state, purchaseID, err) + return nil, fmt.Errorf("failed to record the marketplace listing outcome; retry the cancel to record it: %w", err) + } + if !applied { + return nil, NewClientError(409, marketplaceListingChangedMessage) + } + return map[string]string{"listing_id": listingID, "listing_state": state}, nil +} + +// isListingEndedConflict reports a list-path write refused because another +// request already recorded the listing as canceled or closed. +func isListingEndedConflict(err error) bool { + return errors.Is(err, config.ErrMarketplaceListingEnded) +} diff --git a/internal/api/handler_marketplace_compensation_integration_test.go b/internal/api/handler_marketplace_compensation_integration_test.go index fdb1f1659..d384fe582 100644 --- a/internal/api/handler_marketplace_compensation_integration_test.go +++ b/internal/api/handler_marketplace_compensation_integration_test.go @@ -182,6 +182,11 @@ func TestMarketplaceListCancelFailureRecordsEmptyAWSStateAsPending(t *testing.T) cancelFn: func(_ context.Context, _ string) (ec2svc.MarketplaceListingResult, error) { return ec2svc.MarketplaceListingResult{}, errors.New("RequestLimitExceeded") }, + // The read-back of the status fails too, so the listing is recorded + // as pending rather than with an empty state (issue #525). + describeFn: func(_ context.Context, _ string) (ec2svc.MarketplaceListingResult, error) { + return ec2svc.MarketplaceListingResult{}, errors.New("RequestLimitExceeded") + }, } h := &Handler{config: &failFirstListingWriteStore{StoreInterface: store}, auth: newCompensationAuth(), marketplaceEC2Factory: func(_ aws.Config) marketplaceEC2Client { return ec2 }} diff --git a/internal/api/handler_marketplace_recover_integration_test.go b/internal/api/handler_marketplace_recover_integration_test.go new file mode 100644 index 000000000..3fd1cfa64 --- /dev/null +++ b/internal/api/handler_marketplace_recover_integration_test.go @@ -0,0 +1,203 @@ +//go:build integration + +package api + +// Real-Postgres tests for issue #525 with fake marketplace clients (no AWS +// call is made): a list request that crashes between the AWS create and the DB +// write is recovered through the cancel endpoint, and a cancel that lands +// while a list request is about to record its listing is not undone. + +import ( + "context" + "errors" + "sync" + "testing" + + ec2svc "github.com/LeanerCloud/cloud-commitments-go/providers/aws/services/ec2" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// lockedMarketplaceAWS serializes the fake AWS for the race test. The embedded +// finder is nil: only FindMarketplaceListingByToken, overridden below, is used, +// and embedding keeps the type a marketplaceListingFinder if that interface +// grows. +type lockedMarketplaceAWS struct { + marketplaceListingFinder //nolint:unused // nil; fills in finder methods these tests never call + mu sync.Mutex + fake fakeMarketplaceAWS +} + +func (a *lockedMarketplaceAWS) create(ctx context.Context, req ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { + a.mu.Lock() + defer a.mu.Unlock() + return a.fake.create(ctx, req) +} + +func (a *lockedMarketplaceAWS) cancel(ctx context.Context, id string) (ec2svc.MarketplaceListingResult, error) { + a.mu.Lock() + defer a.mu.Unlock() + return a.fake.cancel(ctx, id) +} + +func (a *lockedMarketplaceAWS) describe(ctx context.Context, id string) (ec2svc.MarketplaceListingResult, error) { + a.mu.Lock() + defer a.mu.Unlock() + return a.fake.describe(ctx, id) +} + +func (a *lockedMarketplaceAWS) FindMarketplaceListingByToken(ctx context.Context, ri, token string) (ec2svc.MarketplaceListingResult, bool, error) { + a.mu.Lock() + defer a.mu.Unlock() + return a.fake.FindMarketplaceListingByToken(ctx, ri, token) +} + +func (a *lockedMarketplaceAWS) live() int { + a.mu.Lock() + defer a.mu.Unlock() + return a.fake.live() +} + +func (a *lockedMarketplaceAWS) setFailCancel(fail bool) { + a.mu.Lock() + defer a.mu.Unlock() + a.fake.failCancel = fail +} + +// listingWriteHook wraps the real store and runs hook before every +// UpdatePurchaseHistoryListing; a non-nil error from hook replaces the write. +type listingWriteHook struct { + config.StoreInterface + mu sync.Mutex + hook func(listingID, state string) error +} + +func (s *listingWriteHook) setHook(hook func(listingID, state string) error) { + s.mu.Lock() + defer s.mu.Unlock() + s.hook = hook +} + +func (s *listingWriteHook) UpdatePurchaseHistoryListing(ctx context.Context, purchaseID, listingID, state string) error { + s.mu.Lock() + hook := s.hook + s.mu.Unlock() + if hook != nil { + if err := hook(listingID, state); err != nil { + return err + } + } + return s.StoreInterface.UpdatePurchaseHistoryListing(ctx, purchaseID, listingID, state) +} + +type recoverRig struct { + h *Handler + store *config.PostgresStore + hook *listingWriteHook + aws *lockedMarketplaceAWS + ec2 *stubMarketplaceEC2 + row *config.PurchaseHistoryRecord +} + +func newRecoverRig(t *testing.T) *recoverRig { + t.Helper() + ctx := t.Context() + pg := testhelpers.RequireMigratedDatabase(ctx, t, "../database/postgres/migrations") + store := config.NewPostgresStore(pg.DB) + rig := &recoverRig{store: store, hook: &listingWriteHook{StoreInterface: store}, aws: &lockedMarketplaceAWS{}} + rig.row = setupCompensationRow(t, store) + rig.ec2 = &stubMarketplaceEC2{createFn: rig.aws.create, cancelFn: rig.aws.cancel, describeFn: rig.aws.describe} + rig.h = &Handler{config: rig.hook, auth: newCompensationAuth(), + marketplaceEC2Factory: func(aws.Config) marketplaceEC2Client { return rig.ec2 }, + marketplaceFinderFactory: func(aws.Config) marketplaceListingFinder { return rig.aws }} + rig.h.awsCfgOnce.Do(func() { rig.h.awsCfg = hostAWSConfig() }) + return rig +} + +func (r *recoverRig) attempt(t *testing.T) config.MarketplaceListingAttempt { + t.Helper() + a, err := r.store.GetMarketplaceListingAttempt(t.Context(), r.row.PurchaseID) + require.NoError(t, err) + require.NotNil(t, a) + return *a +} + +// A list request creates the listing on AWS, then every write after it fails +// (the DB write, the rollback cancel, the record of the uncanceled listing), +// as when the process dies right there. The row is left pending with the +// attempt's token. The cancel endpoint recovers it: it finds the listing by +// the token, cancels it on AWS and records it, and the RI lists again. +func TestMarketplaceCancelRecoversCrashedListAttempt(t *testing.T) { + rig := newRecoverRig(t) + ctx := t.Context() + rig.hook.setHook(func(string, string) error { return errors.New("process died") }) + rig.aws.setFailCancel(true) + + _, err := rig.h.marketplaceList(ctx, marketplaceReq(), rig.row.PurchaseID) + require.Error(t, err) + crashed := rig.attempt(t) + require.Equal(t, config.ListingStatePending, crashed.State) + require.NotEmpty(t, crashed.ClientToken, "the token was persisted before the AWS call") + require.Equal(t, 1, rig.aws.live()) + + rig.hook.setHook(nil) + rig.aws.setFailCancel(false) + resp, err := rig.h.marketplaceCancel(ctx, marketplaceReq(), rig.row.PurchaseID) + require.NoError(t, err) + assert.Equal(t, map[string]string{"listing_id": "ril-1", "listing_state": config.ListingStateCancelled}, resp) + assert.Equal(t, 0, rig.aws.live()) + assert.Equal(t, config.MarketplaceListingAttempt{ListingID: "ril-1", State: config.ListingStateCancelled}, rig.attempt(t)) + + _, err = rig.h.marketplaceList(ctx, marketplaceReq(), rig.row.PurchaseID) + require.NoError(t, err) + assert.Equal(t, config.MarketplaceListingAttempt{ListingID: "ril-2", State: config.ListingStateActive}, rig.attempt(t)) + assert.Equal(t, 1, rig.aws.live()) +} + +// A cancel arrives while a list request has created its listing on AWS but +// not yet recorded it. The cancel finds the listing by the persisted token, +// cancels it and records it; the list request's record then reaches Postgres, +// which refuses to mark the canceled listing active again. The list request +// reports the conflict and does not run its rollback. +func TestMarketplaceCancelDuringListIsNotUndone(t *testing.T) { + rig := newRecoverRig(t) + ctx := t.Context() + reached := make(chan struct{}) + proceed := make(chan struct{}) + var once sync.Once + rig.hook.setHook(func(_, state string) error { + if state == config.ListingStateActive { + once.Do(func() { + close(reached) + <-proceed + }) + } + return nil + }) + + listDone := make(chan error, 1) + go func() { + _, listErr := rig.h.marketplaceList(ctx, marketplaceReq(), rig.row.PurchaseID) + listDone <- listErr + }() + testhelpers.Await(t, reached) + require.Equal(t, config.ListingStatePending, rig.attempt(t).State) + + resp, err := rig.h.marketplaceCancel(ctx, marketplaceReq(), rig.row.PurchaseID) + require.NoError(t, err) + assert.Equal(t, map[string]string{"listing_id": "ril-1", "listing_state": config.ListingStateCancelled}, resp) + close(proceed) + + listErr := testhelpers.Await(t, listDone) + ce, ok := IsClientError(listErr) + require.True(t, ok, "%v", listErr) + assert.Equal(t, 409, ce.code) + assert.Contains(t, ce.message, "canceled by another request") + assert.Equal(t, config.MarketplaceListingAttempt{ListingID: "ril-1", State: config.ListingStateCancelled}, rig.attempt(t), + "the canceled listing must not be recorded as active") + assert.Equal(t, 0, rig.aws.live()) + assert.Equal(t, 1, rig.ec2.cancelCallCount, "only the cancel endpoint's cancel; no rollback cancel") +} diff --git a/internal/api/handler_marketplace_recover_test.go b/internal/api/handler_marketplace_recover_test.go new file mode 100644 index 000000000..9e7f76e55 --- /dev/null +++ b/internal/api/handler_marketplace_recover_test.go @@ -0,0 +1,421 @@ +package api + +// Tests for the issue #525 failure modes the cancel endpoint and the list +// path's record step close: a pending row left by a crashed or concurrent list +// attempt, a cancel whose DB write failed, an empty AWS status, and the +// detached context of the rollback release. They drive HandleRequest against +// the in-memory row and fake AWS of handler_marketplace_attempt_test.go. + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + + ec2svc "github.com/LeanerCloud/cloud-commitments-go/providers/aws/services/ec2" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/aws/aws-lambda-go/events" + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +func (a *fakeMarketplaceAWS) describe(_ context.Context, id string) (ec2svc.MarketplaceListingResult, error) { + for _, l := range a.listings { + if l.id == id { + return ec2svc.MarketplaceListingResult{ListingID: l.id, State: l.state}, nil + } + } + return ec2svc.MarketplaceListingResult{}, errors.New("InvalidReservedInstancesListingId.NotFound") +} + +func (f *fakeListingDB) GetMarketplaceListingAttempt(_ context.Context, _ string) (*config.MarketplaceListingAttempt, error) { + return &config.MarketplaceListingAttempt{ListingID: f.row.ListingID, State: f.row.ListingState, ClientToken: f.token}, nil +} + +// ResolveMarketplaceListingAttempt follows the Postgres store: it writes only +// while the row is as seen or already holds listingID. +func (f *fakeListingDB) ResolveMarketplaceListingAttempt(_ context.Context, _ string, seen config.MarketplaceListingAttempt, listingID, state string) (bool, error) { + asSeen := strings.EqualFold(f.row.ListingState, seen.State) && f.row.ListingID == seen.ListingID && f.token == seen.ClientToken + if !asSeen && (listingID == "" || f.row.ListingID != listingID) { + return false, nil + } + f.row.ListingID, f.row.ListingState = listingID, state + f.token, f.schedule = "", nil + return true, nil +} + +func marketplaceCancelHTTPRequest() *events.LambdaFunctionURLRequest { + req := marketplaceListHTTPRequest() + req.RequestContext.HTTP.Path = "/api/purchases/" + validMarketplacePurchaseID + "/marketplace-cancel" + return req +} + +// cancel drives POST /marketplace-cancel through HandleRequest and returns the +// status and the decoded body. +func (f *marketplaceFixture) cancel(t *testing.T) (int, map[string]string) { + t.Helper() + resp, err := f.h.HandleRequest(context.Background(), marketplaceCancelHTTPRequest()) + require.NoError(t, err) + body := map[string]string{} + if resp.StatusCode == 200 { + require.NoError(t, json.Unmarshal([]byte(resp.Body), &body), resp.Body) + } + return resp.StatusCode, body +} + +// crashAfterCreate leaves the row the way a process that died right after the +// AWS create leaves it: pending, holding the attempt's token, with a live +// listing on AWS that nothing recorded. +func (f *marketplaceFixture) crashAfterCreate(t *testing.T) { + t.Helper() + f.db.failRecords = 2 + f.aws.failCancel = true + require.Equal(t, 502, f.list(t)) + require.Equal(t, config.ListingStatePending, f.db.row.ListingState) + require.NotEmpty(t, f.db.token) + require.Equal(t, 1, f.aws.live()) + f.aws.failCancel = false +} + +// Failure mode 3 of issue #525: a pending row could not be canceled or +// recovered through the API (409 on main). A crash between the AWS create and +// the DB write leaves exactly such a row with a live listing on AWS. Cancel +// finds the listing by the persisted token, cancels it and records it. +func TestMarketplaceCancelHTTP_RecoversCrashedAttemptWithLiveListing(t *testing.T) { + f := newMarketplaceFixture(t) + f.crashAfterCreate(t) + + status, body := f.cancel(t) + require.Equal(t, 200, status) + assert.Equal(t, "ril-1", body["listing_id"]) + assert.Equal(t, config.ListingStateCancelled, body["listing_state"]) + assert.Equal(t, 0, f.aws.live(), "the crashed attempt's listing is canceled on AWS") + assert.Equal(t, config.ListingStateCancelled, f.db.row.ListingState) + assert.Equal(t, "ril-1", f.db.row.ListingID) + assert.Empty(t, f.db.token, "the attempt is resolved") + + assert.Equal(t, 200, f.list(t), "the RI can be listed again") + assert.Equal(t, "ril-2", f.db.row.ListingID) + assert.Equal(t, 1, f.aws.live()) +} + +// A pending attempt whose create never reached AWS: cancel finds nothing, +// frees the row and keeps the token, so the next list request resumes the +// same attempt and AWS still deduplicates on it. +func TestMarketplaceCancelHTTP_PendingAttemptWithNothingOnAWSIsReleased(t *testing.T) { + f := newMarketplaceFixture(t) + f.db.row.ListingState = config.ListingStatePending + f.db.token = "tok-crashed" + f.db.schedule = []byte(`[{"term_months":36,"price":380}]`) + + status, body := f.cancel(t) + require.Equal(t, 200, status) + assert.Empty(t, body["listing_state"]) + assert.Contains(t, body["note"], "nothing was canceled") + assert.Empty(t, f.db.row.ListingState, "the row is no longer stuck pending") + assert.Equal(t, "tok-crashed", f.db.token, "the attempt token is kept") + + assert.Equal(t, 200, f.list(t)) + require.Len(t, f.aws.creates, 1) + assert.Equal(t, "tok-crashed", f.aws.creates[0].ClientToken, "the next list resumes the same attempt") + assert.Len(t, f.aws.listings, 1) +} + +// A listing recorded in AWS's own pending state (or after its rollback cancel +// failed) has no token; cancel cancels the recorded listing. +func TestMarketplaceCancelHTTP_PendingRecordedListingIsCanceled(t *testing.T) { + f := newMarketplaceFixture(t) + f.aws.listings = []*fakeAWSListing{{id: "ril-p", state: config.ListingStatePending, token: "old"}} + f.db.row.ListingID, f.db.row.ListingState = "ril-p", config.ListingStatePending + + status, body := f.cancel(t) + require.Equal(t, 200, status) + assert.Equal(t, config.ListingStateCancelled, body["listing_state"]) + assert.Equal(t, config.ListingStateCancelled, f.aws.listings[0].state) + assert.Equal(t, config.ListingStateCancelled, f.db.row.ListingState) + assert.Equal(t, "ril-p", f.db.row.ListingID) +} + +// A pending claim from before migration 000105 has neither a token nor a +// listing; nothing on AWS can be tied to it, so cancel clears it. +func TestMarketplaceCancelHTTP_LegacyPendingClaimIsCleared(t *testing.T) { + f := newMarketplaceFixture(t) + f.db.row.ListingState = config.ListingStatePending + assert.Equal(t, 409, f.list(t), "a legacy pending row blocks listing") + + status, _ := f.cancel(t) + require.Equal(t, 200, status) + assert.Empty(t, f.db.row.ListingState) + assert.Empty(t, f.aws.creates) + + assert.Equal(t, 200, f.list(t)) + assert.Equal(t, 1, f.aws.live()) +} + +// Failure mode 4: the cancel succeeded on AWS but its DB write failed. The +// retry got AWS's error for canceling a canceled listing (400 on main) and the +// row stayed active forever. The retry now reads the listing back and records +// the canceled state. +func TestMarketplaceCancelHTTP_RetryAfterCancelPersistFailureRecordsIt(t *testing.T) { + f := newMarketplaceFixture(t) + require.Equal(t, 200, f.list(t)) + f.db.failRecords = 1 + + status, _ := f.cancel(t) + require.Equal(t, 500, status) + require.Equal(t, 0, f.aws.live(), "AWS canceled the listing") + require.Equal(t, config.ListingStateActive, f.db.row.ListingState, "the DB write was lost") + + status, body := f.cancel(t) + require.Equal(t, 200, status, "the retry must record the cancel, not fail on the canceled listing") + assert.Equal(t, config.ListingStateCancelled, body["listing_state"]) + assert.Equal(t, config.ListingStateCancelled, f.db.row.ListingState) +} + +// Failure mode 5: a create response with an empty status was recorded as an +// empty state on main, which reads as "not listed", so the next request +// created a second listing. The status is read back from AWS instead. +func TestMarketplaceListHTTP_EmptyCreateStatusIsReadBack(t *testing.T) { + f := newMarketplaceFixture(t) + f.blankCreateStatus() + + require.Equal(t, 200, f.list(t)) + assert.Equal(t, config.ListingStateActive, f.db.row.ListingState) + assert.Equal(t, 409, f.list(t), "the live listing blocks a second one") + assert.Len(t, f.aws.listings, 1) +} + +// When the read-back fails too, the listing is recorded as pending: still +// blocking a second listing, and cancelable through the endpoint. +func TestMarketplaceListHTTP_EmptyCreateStatusUnreadableIsPending(t *testing.T) { + f := newMarketplaceFixture(t) + f.blankCreateStatus() + stub := f.stub() + stub.describeFn = func(context.Context, string) (ec2svc.MarketplaceListingResult, error) { + return ec2svc.MarketplaceListingResult{}, errors.New("RequestLimitExceeded") + } + + require.Equal(t, 200, f.list(t)) + assert.Equal(t, config.ListingStatePending, f.db.row.ListingState) + assert.Equal(t, 409, f.list(t)) + assert.Len(t, f.aws.listings, 1) + + stub.describeFn = f.aws.describe + status, _ := f.cancel(t) + require.Equal(t, 200, status) + assert.Equal(t, 0, f.aws.live()) + assert.Equal(t, config.ListingStateCancelled, f.db.row.ListingState) +} + +// A cancel that lands while a list request is between its AWS create and its +// DB write cancels the listing it finds by the token. The list request's +// write must not then record the canceled listing as active. +func TestMarketplaceListHTTP_CancelDuringListIsNotUndone(t *testing.T) { + f := newMarketplaceFixture(t) + stub := f.stub() + create := stub.createFn + var cancelStatus int + stub.createFn = func(ctx context.Context, req ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { + res, err := create(ctx, req) + cancelStatus, _ = f.cancel(t) + return res, err + } + + assert.Equal(t, 409, f.list(t), "the list request reports the concurrent cancel") + assert.Equal(t, 200, cancelStatus) + assert.Equal(t, 0, f.aws.live()) + assert.Equal(t, config.ListingStateCancelled, f.db.row.ListingState, "the canceled listing must not be recorded as active") + assert.Equal(t, 1, stub.cancelCallCount, "no compensating cancel for a listing already canceled") +} + +func (f *marketplaceFixture) stub() *stubMarketplaceEC2 { + s, ok := f.h.buildMarketplaceEC2Client(aws.Config{}).(*stubMarketplaceEC2) + if !ok { + panic("marketplace fixture without a stub EC2 client") + } + return s +} + +func (f *marketplaceFixture) blankCreateStatus() { + s := f.stub() + create := s.createFn + s.createFn = func(ctx context.Context, req ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { + res, err := create(ctx, req) + res.State = "" + return res, err + } +} + +// Failure mode 6: after the rollback cancel succeeded, the release that runs +// when recording the canceled listing fails used the request context, which +// is gone when the request timed out. +func TestMarketplaceList_RollbackReleaseUsesDetachedContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cfgStore := &MockConfigStore{} + authSvc := &MockAuthService{} + adminSession(authSvc) + cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(standardRow(), 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). + Run(func(mock.Arguments) { cancel() }).Return(errors.New("db down")) + cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateCancelled). + Return(errors.New("db still down")) + var releaseCtxErr error + cfgStore.On("ReleaseMarketplaceListingClaim", mock.Anything, validMarketplacePurchaseID, "tok", "", false). + Run(func(args mock.Arguments) { + c, _ := args.Get(0).(context.Context) + releaseCtxErr = c.Err() + }).Return(nil).Once() + + h := newMarketplaceHandler(cfgStore, authSvc, &stubMarketplaceEC2{}) + _, err := h.marketplaceList(ctx, marketplaceReq(), validMarketplacePurchaseID) + require.Error(t, err) + cfgStore.AssertExpectations(t) + assert.NoError(t, releaseCtxErr, "the release must not run on the canceled request context") +} + +// The cancel's DB write runs on a detached context: AWS has already canceled +// the listing, so a caller that disconnects must not lose the record. +func TestMarketplaceCancel_RecordsOnDetachedContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cfgStore := &MockConfigStore{} + authSvc := &MockAuthService{} + adminSession(authSvc) + row := standardRow() + row.ListingID, row.ListingState = "ril-cancel", config.ListingStateActive + cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(row, nil) + var recordCtxErr error + cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-cancel", config.ListingStateCancelled). + Run(func(args mock.Arguments) { + c, _ := args.Get(0).(context.Context) + recordCtxErr = c.Err() + }).Return(nil).Once() + + ec2 := &stubMarketplaceEC2{cancelFn: func(_ context.Context, id string) (ec2svc.MarketplaceListingResult, error) { + cancel() // the caller goes away while AWS cancels + return ec2svc.MarketplaceListingResult{ListingID: id, State: config.ListingStateCancelled}, nil + }} + h := newMarketplaceHandler(cfgStore, authSvc, ec2) + _, err := h.marketplaceCancel(ctx, marketplaceReq(), validMarketplacePurchaseID) + require.NoError(t, err) + cfgStore.AssertExpectations(t) + assert.NoError(t, recordCtxErr) +} + +// A cancel that AWS accepts without a status records it as canceled, never as +// an empty state that reads as "not listed". +func TestMarketplaceCancel_EmptyStatusRecordedAsCanceled(t *testing.T) { + cfgStore := &MockConfigStore{} + authSvc := &MockAuthService{} + adminSession(authSvc) + row := standardRow() + row.ListingID, row.ListingState = "ril-cancel", config.ListingStateActive + cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(row, nil) + cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-cancel", config.ListingStateCancelled). + Return(nil).Once() + + ec2 := &stubMarketplaceEC2{cancelFn: func(_ context.Context, id string) (ec2svc.MarketplaceListingResult, error) { + return ec2svc.MarketplaceListingResult{ListingID: id}, nil + }} + h := newMarketplaceHandler(cfgStore, authSvc, ec2) + _, err := h.marketplaceCancel(context.Background(), marketplaceReq(), validMarketplacePurchaseID) + require.NoError(t, err) + cfgStore.AssertExpectations(t) +} + +// A cancel failure that is not about an ended listing still fails, and a +// still-active listing is not recorded as canceled. +func TestMarketplaceCancel_FailureOnLiveListingIsNotRecorded(t *testing.T) { + cfgStore := &MockConfigStore{} + authSvc := &MockAuthService{} + adminSession(authSvc) + row := standardRow() + row.ListingID, row.ListingState = "ril-cancel", config.ListingStateActive + cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(row, nil) + + ec2 := &stubMarketplaceEC2{cancelFn: func(context.Context, string) (ec2svc.MarketplaceListingResult, error) { + return ec2svc.MarketplaceListingResult{}, errors.New("RequestLimitExceeded") + }} + h := newMarketplaceHandler(cfgStore, authSvc, ec2) + _, err := h.marketplaceCancel(context.Background(), marketplaceReq(), validMarketplacePurchaseID) + ce, ok := IsClientError(err) + require.True(t, ok, "%v", err) + assert.Equal(t, 502, ce.code) + assert.Equal(t, 1, ec2.describeCallCount) + cfgStore.AssertNotCalled(t, "UpdatePurchaseHistoryListing", mock.Anything, mock.Anything, mock.Anything, mock.Anything) +} + +// A pending row that changed between the purchase lookup and the attempt read +// is not touched. +func TestMarketplaceCancel_PendingRowThatChangedIs409(t *testing.T) { + cfgStore := &MockConfigStore{} + authSvc := &MockAuthService{} + adminSession(authSvc) + row := standardRow() + row.ListingState = config.ListingStatePending + cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(row, nil) + cfgStore.On("GetMarketplaceListingAttempt", mock.Anything, validMarketplacePurchaseID). + Return(&config.MarketplaceListingAttempt{ListingID: "ril-new", State: config.ListingStateActive}, nil) + + ec2 := &stubMarketplaceEC2{} + h := newMarketplaceHandler(cfgStore, authSvc, ec2) + _, err := h.marketplaceCancel(context.Background(), marketplaceReq(), validMarketplacePurchaseID) + ce, ok := IsClientError(err) + require.True(t, ok, "%v", err) + assert.Equal(t, 409, ce.code) + assert.Zero(t, ec2.cancelCallCount) +} + +// The recovery write is fenced: if the row moved on (another request recorded +// a different listing), cancel reports a conflict instead of overwriting it. +func TestMarketplaceCancel_FencedRecoveryWriteLosesIs409(t *testing.T) { + cfgStore := &MockConfigStore{} + authSvc := &MockAuthService{} + adminSession(authSvc) + row := standardRow() + row.ListingState = config.ListingStatePending + seen := config.MarketplaceListingAttempt{State: config.ListingStatePending} + cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(row, nil) + cfgStore.On("GetMarketplaceListingAttempt", mock.Anything, validMarketplacePurchaseID).Return(&seen, nil) + cfgStore.On("ResolveMarketplaceListingAttempt", mock.Anything, validMarketplacePurchaseID, seen, "", "").Return(false, nil).Once() + + h := newMarketplaceHandler(cfgStore, authSvc, &stubMarketplaceEC2{}) + _, err := h.marketplaceCancel(context.Background(), marketplaceReq(), validMarketplacePurchaseID) + ce, ok := IsClientError(err) + require.True(t, ok, "%v", err) + assert.Equal(t, 409, ce.code) + cfgStore.AssertExpectations(t) +} + +// The record of a listing whose rollback cancel failed loses to a cancel that +// recorded the listing canceled meanwhile: that is reported as the concurrent +// cancel, not as a row stuck pending. +func TestMarketplaceList_UncanceledRecordAfterConcurrentCancelIs409(t *testing.T) { + cfgStore := &MockConfigStore{} + authSvc := &MockAuthService{} + adminSession(authSvc) + cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).Return(standardRow(), 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). + Return(config.ErrMarketplaceListingEnded).Once() + + ec2 := &stubMarketplaceEC2{cancelFn: func(context.Context, string) (ec2svc.MarketplaceListingResult, error) { + return ec2svc.MarketplaceListingResult{}, errors.New("RequestLimitExceeded") + }} + h := newMarketplaceHandler(cfgStore, authSvc, ec2) + _, err := h.marketplaceList(context.Background(), marketplaceReq(), validMarketplacePurchaseID) + ce, ok := IsClientError(err) + require.True(t, ok, "%v", err) + assert.Equal(t, 409, ce.code) + assert.Contains(t, ce.message, "canceled by another request") + cfgStore.AssertExpectations(t) +} diff --git a/internal/api/handler_marketplace_test.go b/internal/api/handler_marketplace_test.go index 6f877f572..b94c1c863 100644 --- a/internal/api/handler_marketplace_test.go +++ b/internal/api/handler_marketplace_test.go @@ -27,12 +27,15 @@ const validMarketplacePurchaseID = "11111111-1111-1111-1111-111111111111" type stubMarketplaceEC2 struct { createFn func(ctx context.Context, req ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) cancelFn func(ctx context.Context, listingID string) (ec2svc.MarketplaceListingResult, error) + describeFn func(ctx context.Context, listingID string) (ec2svc.MarketplaceListingResult, error) fetchClassFn func(ctx context.Context, reservedInstancesID string) (string, error) lastCreateReq ec2svc.MarketplaceListingRequest createCallCount int cancelCallCount int lastCancelID string + // describeCallCount counts DescribeMarketplaceListing calls (issue #525). + describeCallCount int } func (s *stubMarketplaceEC2) CreateMarketplaceListing(ctx context.Context, req ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) { @@ -44,7 +47,11 @@ func (s *stubMarketplaceEC2) CreateMarketplaceListing(ctx context.Context, req e return ec2svc.MarketplaceListingResult{ListingID: "ril-default", State: config.ListingStateActive}, nil } -func (s *stubMarketplaceEC2) DescribeMarketplaceListing(_ context.Context, listingID string) (ec2svc.MarketplaceListingResult, error) { +func (s *stubMarketplaceEC2) DescribeMarketplaceListing(ctx context.Context, listingID string) (ec2svc.MarketplaceListingResult, error) { + s.describeCallCount++ + if s.describeFn != nil { + return s.describeFn(ctx, listingID) + } return ec2svc.MarketplaceListingResult{ListingID: listingID, State: config.ListingStateActive}, nil } @@ -403,6 +410,8 @@ func TestMarketplaceList_CancelFailureRecordsNonActiveStateAsPending(t *testing. wantNotMsg string }{ {name: "pending", awsState: config.ListingStatePending, wantState: config.ListingStatePending, wantMsg: "needs reconciliation", wantNotMsg: "cancel it again"}, + // An empty status is never recorded (issue #525): with the read-back + // failing too, the listing is recorded as pending from the start. {name: "empty", awsState: "", wantState: config.ListingStatePending, wantMsg: "needs reconciliation", wantNotMsg: "cancel it again"}, } { t.Run(tc.name, func(t *testing.T) { @@ -413,7 +422,7 @@ func TestMarketplaceList_CancelFailureRecordsNonActiveStateAsPending(t *testing. Return(standardRow(), 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). + cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-x", tc.wantState). Return(errors.New("db down")).Once() cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-x", tc.wantState). Return(nil).Once() @@ -425,6 +434,9 @@ func TestMarketplaceList_CancelFailureRecordsNonActiveStateAsPending(t *testing. cancelFn: func(_ context.Context, _ string) (ec2svc.MarketplaceListingResult, error) { return ec2svc.MarketplaceListingResult{}, errors.New("throttled") }, + describeFn: func(_ context.Context, _ string) (ec2svc.MarketplaceListingResult, error) { + return ec2svc.MarketplaceListingResult{}, errors.New("throttled") + }, } h := newMarketplaceHandler(cfgStore, authSvc, ec2) _, err := h.marketplaceList(context.Background(), marketplaceReq(), validMarketplacePurchaseID) diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 9a078709a..e3e903d2a 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -406,9 +406,23 @@ type StoreInterface interface { // CreateReservedInstancesListing succeeds (listing_state="active") and // 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). + // price schedule (issue #525). It never marks a listing the row records as + // canceled or closed live again; that write returns + // ErrMarketplaceListingEnded. UpdatePurchaseHistoryListing(ctx context.Context, purchaseID, listingID, listingState string) error + // GetMarketplaceListingAttempt returns the listing id, state and the token + // of an unresolved list attempt of the row for purchaseID, or (nil, nil) + // when there is none. The cancel endpoint uses it to recover a pending row + // (issue #525). + GetMarketplaceListingAttempt(ctx context.Context, purchaseID string) (*MarketplaceListingAttempt, error) + + // ResolveMarketplaceListingAttempt records listingID in an ended (or, to + // clear a stuck pending row without a listing, empty) state and drops the + // attempt token, only while the row still matches seen or already holds + // listingID. It reports whether the row was written (issue #525). + ResolveMarketplaceListingAttempt(ctx context.Context, purchaseID string, seen MarketplaceListingAttempt, listingID, state string) (bool, error) + // StampOfferingClass writes the offering_class value to a purchase_history // row identified by purchase_id. Called by the marketplace-list handler // when offering_class is absent in the DB (pre-migration 000087 rows and diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 668d1bd27..e11105fc8 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -2369,18 +2369,24 @@ func (s *PostgresStore) SavePurchaseHistory(ctx context.Context, record *Purchas // succeeds (listing_id + listing_state="active") and by the status-poll path on // the later closed/canceled transitions. func (s *PostgresStore) UpdatePurchaseHistoryListing(ctx context.Context, purchaseID, listingID, listingState string) error { + // A listing recorded as canceled or closed is never marked live again + // (issue #525): a list request that lost a race with the cancel endpoint + // gets ErrMarketplaceListingEnded instead of reviving the canceled listing. query := ` UPDATE purchase_history SET listing_id = $1, listing_state = $2, listing_client_token = NULL, listing_price_schedule = NULL WHERE purchase_id = $3 + AND NOT (COALESCE(listing_id, '') = $1 + AND lower(COALESCE(listing_state, '')) IN ($4, $5) + AND lower($2) NOT IN ($4, $5)) ` - tag, err := s.db.Exec(ctx, query, listingID, listingState, purchaseID) + tag, err := s.db.Exec(ctx, query, listingID, listingState, purchaseID, ListingStateCancelled, ListingStateClosed) if err != nil { return fmt.Errorf("failed to update listing state for purchase %s: %w", purchaseID, err) } if tag.RowsAffected() == 0 { - return fmt.Errorf("purchase %s not found in history", purchaseID) + return s.listingEndedOrMissing(ctx, purchaseID) } return nil } diff --git a/internal/config/store_postgres_marketplace_attempt.go b/internal/config/store_postgres_marketplace_attempt.go new file mode 100644 index 000000000..981dbb938 --- /dev/null +++ b/internal/config/store_postgres_marketplace_attempt.go @@ -0,0 +1,97 @@ +package config + +// Store side of recovering a marketplace listing attempt from the cancel +// endpoint (issue #525). A row left pending by a crashed or concurrent +// marketplace-list request is resolved against what the cancel handler read, +// so a cancel never overwrites a listing or attempt it did not see. + +import ( + "context" + "errors" + "fmt" + "strings" + + "github.com/jackc/pgx/v5" +) + +// ErrMarketplaceListingEnded is returned by UpdatePurchaseHistoryListing when +// the row already records the same listing as canceled or closed and the write +// would mark it live again. AWS never revives an ended listing, so such a +// write comes from a request that raced the cancel and lost. +var ErrMarketplaceListingEnded = errors.New("the marketplace listing is already recorded as ended") + +// MarketplaceListingAttempt is the listing state of a purchase_history row as +// the cancel handler read it: the recorded listing, its state, and the client +// token of an unresolved list attempt (empty when none). +type MarketplaceListingAttempt struct { + ListingID string + State string + ClientToken string //nolint:gosec // G117: an idempotency token for AWS, not a credential +} + +// GetMarketplaceListingAttempt returns the listing id, state and unresolved +// attempt token of the row for purchaseID, or (nil, nil) when there is none. +// With duplicate rows for one purchase_id it reads the earliest. +func (s *PostgresStore) GetMarketplaceListingAttempt(ctx context.Context, purchaseID string) (*MarketplaceListingAttempt, error) { + query := ` + SELECT COALESCE(listing_id, ''), COALESCE(listing_state, ''), COALESCE(listing_client_token, '') + FROM purchase_history + WHERE purchase_id = $1 + ORDER BY created_at, id + LIMIT 1 + ` + var a MarketplaceListingAttempt + err := s.db.QueryRow(ctx, query, purchaseID).Scan(&a.ListingID, &a.State, &a.ClientToken) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("failed to read marketplace listing attempt for purchase %s: %w", purchaseID, err) + } + return &a, nil +} + +// ResolveMarketplaceListingAttempt records listingID in state on the row and +// drops any unresolved attempt token, but only while the row still matches +// seen, or already holds listingID (another request recorded the listing the +// caller just canceled). state must be an ended state (canceled or closed) or +// empty, which clears a stuck pending row that has no listing; an empty +// listingID clears the listing id too. It reports whether the row was written. +func (s *PostgresStore) ResolveMarketplaceListingAttempt(ctx context.Context, purchaseID string, seen MarketplaceListingAttempt, listingID, state string) (bool, error) { + if state != "" && !isEndedListingState(state) { + return false, fmt.Errorf("cannot resolve marketplace listing %s for purchase %s to live state %q", listingID, purchaseID, state) + } + query := ` + UPDATE purchase_history + SET listing_id = NULLIF($2, ''), listing_state = NULLIF($3, ''), + listing_client_token = NULL, listing_price_schedule = NULL + WHERE purchase_id = $1 + AND ((lower(COALESCE(listing_state, '')) = lower($4) + AND COALESCE(listing_id, '') = $5 + AND COALESCE(listing_client_token, '') = $6) + OR ($2 <> '' AND listing_id = $2)) + ` + tag, err := s.db.Exec(ctx, query, purchaseID, listingID, state, seen.State, seen.ListingID, seen.ClientToken) + if err != nil { + return false, fmt.Errorf("failed to resolve marketplace listing attempt for purchase %s: %w", purchaseID, err) + } + return tag.RowsAffected() > 0, nil +} + +func isEndedListingState(state string) bool { + return strings.EqualFold(state, ListingStateCancelled) || strings.EqualFold(state, ListingStateClosed) +} + +// listingEndedOrMissing tells apart the two reasons a guarded +// UpdatePurchaseHistoryListing wrote nothing: the row records the listing as +// ended (ErrMarketplaceListingEnded), or there is no row. +func (s *PostgresStore) listingEndedOrMissing(ctx context.Context, purchaseID string) error { + var exists bool + if err := s.db.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM purchase_history WHERE purchase_id = $1)`, purchaseID).Scan(&exists); err != nil { + return fmt.Errorf("failed to update listing state for purchase %s: %w", purchaseID, err) + } + if exists { + return fmt.Errorf("purchase %s: %w", purchaseID, ErrMarketplaceListingEnded) + } + return fmt.Errorf("purchase %s not found in history", purchaseID) +} diff --git a/internal/config/store_postgres_marketplace_attempt_integration_test.go b/internal/config/store_postgres_marketplace_attempt_integration_test.go new file mode 100644 index 000000000..e0f695177 --- /dev/null +++ b/internal/config/store_postgres_marketplace_attempt_integration_test.go @@ -0,0 +1,237 @@ +//go:build integration +// +build integration + +package config + +// Real-DB tests for the cancel endpoint's recovery writes (issue #525): the +// guard that keeps a canceled listing from being recorded live again, the +// fenced ResolveMarketplaceListingAttempt, and the race between a list +// request recording its listing and a cancel recording that listing canceled. + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + + "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" +) + +func TestMarketplaceListingAttempt_PostgresSemantics(t *testing.T) { + ctx := context.Background() + container := testhelpers.RequireMigratedDatabase(ctx, t, getTestMigrationsPath()) + store := NewPostgresStore(container.DB) + pool := container.DB.Pool() + + t.Run("ended listing is never recorded live again", func(t *testing.T) { checkListingEndedGuard(ctx, t, store) }) + t.Run("attempt read", func(t *testing.T) { checkGetMarketplaceListingAttempt(ctx, t, store) }) + t.Run("resolve is fenced", func(t *testing.T) { checkResolveIsFenced(ctx, t, store) }) + t.Run("cancel committed first, list record waits", func(t *testing.T) { checkLateListRecordLoses(ctx, t, store, pool) }) + t.Run("list record committed first, cancel waits", func(t *testing.T) { checkLateCancelStillWins(ctx, t, store, pool) }) + t.Run("free-running race", func(t *testing.T) { checkCancelListRace(ctx, t, store) }) +} + +func listingOf(ctx context.Context, t *testing.T, store *PostgresStore, id string) MarketplaceListingAttempt { + t.Helper() + a, err := store.GetMarketplaceListingAttempt(ctx, id) + require.NoError(t, err) + require.NotNil(t, a) + return *a +} + +// claimedRow seeds a row and claims it with token, as a list request does +// before its AWS create. +func claimedRow(ctx context.Context, t *testing.T, store *PostgresStore, id, token string) { + t.Helper() + seedClaimRow(ctx, t, store, id) + claim, err := store.ClaimMarketplaceListingSlot(ctx, id, token, []byte(claimTestSchedule)) + require.NoError(t, err) + require.NotNil(t, claim) +} + +func checkListingEndedGuard(ctx context.Context, t *testing.T, store *PostgresStore) { + const id = "ri-ended-525" + seedClaimRow(ctx, t, store, id) + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-1", ListingStateActive)) + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-1", ListingStateCancelled)) + + for _, live := range []string{ListingStateActive, ListingStatePending, "ACTIVE"} { + err := store.UpdatePurchaseHistoryListing(ctx, id, "ril-1", live) + require.ErrorIs(t, err, ErrMarketplaceListingEnded, live) + } + got := listingOf(ctx, t, store, id) + assert.Equal(t, ListingStateCancelled, got.State, "the canceled listing must stay canceled") + + // Ended to ended is allowed (a sale recorded after a cancel attempt), and + // so is a new listing. + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-1", ListingStateClosed)) + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-2", ListingStateActive)) + assert.Equal(t, ListingStateActive, listingOf(ctx, t, store, id).State) + + err := store.UpdatePurchaseHistoryListing(ctx, "no-such-ri-525", "ril-1", ListingStateActive) + require.Error(t, err) + assert.NotErrorIs(t, err, ErrMarketplaceListingEnded, "a missing row is not an ended listing") +} + +func checkGetMarketplaceListingAttempt(ctx context.Context, t *testing.T, store *PostgresStore) { + const id = "ri-read-525" + claimedRow(ctx, t, store, id, "tok-read") + assert.Equal(t, MarketplaceListingAttempt{State: ListingStatePending, ClientToken: "tok-read"}, listingOf(ctx, t, store, id)) + + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-r", ListingStateActive)) + assert.Equal(t, MarketplaceListingAttempt{ListingID: "ril-r", State: ListingStateActive}, listingOf(ctx, t, store, id)) + + missing, err := store.GetMarketplaceListingAttempt(ctx, "no-such-ri-525") + require.NoError(t, err) + assert.Nil(t, missing) +} + +func checkResolveIsFenced(ctx context.Context, t *testing.T, store *PostgresStore) { + const id = "ri-resolve-525" + claimedRow(ctx, t, store, id, "tok-r") + seen := listingOf(ctx, t, store, id) + + _, err := store.ResolveMarketplaceListingAttempt(ctx, id, seen, "ril-x", ListingStateActive) + require.Error(t, err, "resolve never records a live state") + + stale := seen + stale.ClientToken = "tok-other" + applied, err := store.ResolveMarketplaceListingAttempt(ctx, id, stale, "ril-x", ListingStateCancelled) + require.NoError(t, err) + assert.False(t, applied, "a stale read must not write") + assert.Equal(t, seen, listingOf(ctx, t, store, id)) + + applied, err = store.ResolveMarketplaceListingAttempt(ctx, id, seen, "ril-x", ListingStateCancelled) + require.NoError(t, err) + assert.True(t, applied) + assert.Equal(t, MarketplaceListingAttempt{ListingID: "ril-x", State: ListingStateCancelled}, listingOf(ctx, t, store, id)) + claim, err := store.ClaimMarketplaceListingSlot(ctx, id, "tok-next", []byte(claimTestSchedule)) + require.NoError(t, err) + require.NotNil(t, claim) + assert.False(t, claim.Resumed, "the resolved attempt's token is gone") + + // Stale read, but the row holds the listing the caller canceled: written. + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-y", ListingStateActive)) + applied, err = store.ResolveMarketplaceListingAttempt(ctx, id, stale, "ril-y", ListingStateCancelled) + require.NoError(t, err) + assert.True(t, applied) + assert.Equal(t, ListingStateCancelled, listingOf(ctx, t, store, id).State) + + // Stale read and another listing on the row: not written. + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-z", ListingStateActive)) + applied, err = store.ResolveMarketplaceListingAttempt(ctx, id, stale, "ril-y", ListingStateCancelled) + require.NoError(t, err) + assert.False(t, applied) + assert.Equal(t, MarketplaceListingAttempt{ListingID: "ril-z", State: ListingStateActive}, listingOf(ctx, t, store, id)) + + // A pending row with neither token nor listing is cleared to no state. + const legacy = "ri-legacy-525" + seedClaimRow(ctx, t, store, legacy) + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, legacy, "", ListingStatePending)) + applied, err = store.ResolveMarketplaceListingAttempt(ctx, legacy, listingOf(ctx, t, store, legacy), "", "") + require.NoError(t, err) + assert.True(t, applied) + assert.Equal(t, MarketplaceListingAttempt{}, listingOf(ctx, t, store, legacy)) +} + +// cancelWriteSQL is the row change the cancel endpoint makes when it records +// the attempt's listing canceled, run inside a held transaction. +const cancelWriteSQL = `UPDATE purchase_history SET listing_id = $2, listing_state = $3, + listing_client_token = NULL, listing_price_schedule = NULL WHERE purchase_id = $1` + +// listWriteSQL is the row change of the list request recording its listing. +const listWriteSQL = `UPDATE purchase_history SET listing_id = $2, listing_state = $3, + listing_client_token = NULL, listing_price_schedule = NULL WHERE purchase_id = $1` + +// The cancel's write commits while the list request's record is queued on +// the row lock. Postgres re-checks the record's guard on the committed row, +// so the record is refused and the listing stays canceled. +func checkLateListRecordLoses(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + const id = "ri-late-list-525" + claimedRow(ctx, t, store, id, "tok-L") + tx, err := pool.Begin(ctx) + require.NoError(t, err) + defer func() { _ = tx.Rollback(ctx) }() + _, err = tx.Exec(ctx, cancelWriteSQL, id, "ril-L", ListingStateCancelled) + require.NoError(t, err) + + done := make(chan error, 1) + go func() { done <- store.UpdatePurchaseHistoryListing(ctx, id, "ril-L", ListingStateActive) }() + testhelpers.WaitForLockWait(ctx, t, pool, "%UPDATE purchase_history%listing_id = $1%") + require.NoError(t, tx.Commit(ctx)) + + require.ErrorIs(t, testhelpers.Await(t, done), ErrMarketplaceListingEnded) + assert.Equal(t, MarketplaceListingAttempt{ListingID: "ril-L", State: ListingStateCancelled}, listingOf(ctx, t, store, id)) +} + +// The list request's record commits while the cancel's fenced write is queued. +// The row no longer matches what the cancel read, but it holds the listing the +// cancel canceled on AWS, so the cancel's write applies. +func checkLateCancelStillWins(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + const id = "ri-late-cancel-525" + claimedRow(ctx, t, store, id, "tok-C") + seen := listingOf(ctx, t, store, id) + tx, err := pool.Begin(ctx) + require.NoError(t, err) + defer func() { _ = tx.Rollback(ctx) }() + _, err = tx.Exec(ctx, listWriteSQL, id, "ril-C", ListingStateActive) + require.NoError(t, err) + + type result struct { + applied bool + err error + } + done := make(chan result, 1) + go func() { + applied, resolveErr := store.ResolveMarketplaceListingAttempt(ctx, id, seen, "ril-C", ListingStateCancelled) + done <- result{applied, resolveErr} + }() + testhelpers.WaitForLockWait(ctx, t, pool, "%UPDATE purchase_history%NULLIF%") + require.NoError(t, tx.Commit(ctx)) + + got := testhelpers.Await(t, done) + require.NoError(t, got.err) + assert.True(t, got.applied) + assert.Equal(t, MarketplaceListingAttempt{ListingID: "ril-C", State: ListingStateCancelled}, listingOf(ctx, t, store, id)) +} + +// Both writes fire at once on many rows. Whatever the order, the listing ends +// recorded as canceled, and the list request learns it lost exactly when it +// ran second. +func checkCancelListRace(ctx context.Context, t *testing.T, store *PostgresStore) { + for i := range 25 { + id := fmt.Sprintf("ri-race-cancel-525-%d", i) + listing := fmt.Sprintf("ril-race-%d", i) + claimedRow(ctx, t, store, id, "tok-race") + seen := listingOf(ctx, t, store, id) + + var wg sync.WaitGroup + start := make(chan struct{}) + var listErr, resolveErr error + var applied bool + wg.Add(2) + go func() { + defer wg.Done() + <-start + listErr = store.UpdatePurchaseHistoryListing(ctx, id, listing, ListingStateActive) + }() + go func() { + defer wg.Done() + <-start + applied, resolveErr = store.ResolveMarketplaceListingAttempt(ctx, id, seen, listing, ListingStateCancelled) + }() + close(start) + wg.Wait() + + require.NoError(t, resolveErr) + assert.True(t, applied, "row %s: the cancel's write applies in either order", id) + if listErr != nil { + require.True(t, errors.Is(listErr, ErrMarketplaceListingEnded), "row %s: %v", id, listErr) + } + assert.Equal(t, MarketplaceListingAttempt{ListingID: listing, State: ListingStateCancelled}, listingOf(ctx, t, store, id), "row %s", id) + } +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index e946ed5f8..669665622 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -858,6 +858,28 @@ func (m *MockConfigStore) UpdatePurchaseHistoryListing(ctx context.Context, purc return args.Error(0) } +// GetMarketplaceListingAttempt mocks reading a row's listing attempt (issue #525). +func (m *MockConfigStore) GetMarketplaceListingAttempt(ctx context.Context, purchaseID string) (*config.MarketplaceListingAttempt, error) { + m.record("GetMarketplaceListingAttempt", ctx, purchaseID) + args := m.Called(ctx, purchaseID) + var attempt *config.MarketplaceListingAttempt + if v := args.Get(0); v != nil { + a, ok := v.(*config.MarketplaceListingAttempt) + if !ok { + panic(fmt.Sprintf("mock: expected *config.MarketplaceListingAttempt, got %T", v)) + } + attempt = a + } + return attempt, args.Error(1) +} + +// ResolveMarketplaceListingAttempt mocks the fenced recovery write of the cancel endpoint (issue #525). +func (m *MockConfigStore) ResolveMarketplaceListingAttempt(ctx context.Context, purchaseID string, seen config.MarketplaceListingAttempt, listingID, state string) (bool, error) { + m.record("ResolveMarketplaceListingAttempt", ctx, purchaseID, seen, listingID, state) + args := m.Called(ctx, purchaseID, seen, listingID, state) + return args.Bool(0), args.Error(1) +} + // StampOfferingClass mocks persisting the offering_class fetched from AWS (issue #292). func (m *MockConfigStore) StampOfferingClass(ctx context.Context, purchaseID, offeringClass string) error { m.record("StampOfferingClass", ctx, purchaseID, offeringClass) diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 845618a1d..b9d863b34 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -481,6 +481,12 @@ func (m *mockConfigStoreForHealth) UpdateGlobalConfigAtomic(_ context.Context, a func (m *mockConfigStoreForHealth) UpdatePurchaseHistoryListing(_ context.Context, _, _, _ string) error { return nil } +func (m *mockConfigStoreForHealth) GetMarketplaceListingAttempt(_ context.Context, _ string) (*config.MarketplaceListingAttempt, error) { + return nil, nil +} +func (m *mockConfigStoreForHealth) ResolveMarketplaceListingAttempt(_ context.Context, _ string, _ config.MarketplaceListingAttempt, _, _ string) (bool, error) { + return false, nil +} func (m *mockConfigStoreForHealth) ClaimMarketplaceListingSlot(_ context.Context, _, _ string, _ []byte) (*config.MarketplaceListingClaim, error) { return &config.MarketplaceListingClaim{}, nil } From 57c97dd4a7866e6eb2450b0a4b881c5a23732c85 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 23 Oct 2026 08:27:11 +0000 Subject: [PATCH 2/2] test(config): rename the marketplace attempt race SQL so it composes with the listing poller tests fix/issue-27's settlement integration test declares its own cancelWriteSQL in the same package; give this file's constants their own names. Refs #525 Co-Authored-By: Claude Opus 5.5 --- ...stgres_marketplace_attempt_integration_test.go | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/internal/config/store_postgres_marketplace_attempt_integration_test.go b/internal/config/store_postgres_marketplace_attempt_integration_test.go index e0f695177..e2b64d9d0 100644 --- a/internal/config/store_postgres_marketplace_attempt_integration_test.go +++ b/internal/config/store_postgres_marketplace_attempt_integration_test.go @@ -138,13 +138,14 @@ func checkResolveIsFenced(ctx context.Context, t *testing.T, store *PostgresStor assert.Equal(t, MarketplaceListingAttempt{}, listingOf(ctx, t, store, legacy)) } -// cancelWriteSQL is the row change the cancel endpoint makes when it records -// the attempt's listing canceled, run inside a held transaction. -const cancelWriteSQL = `UPDATE purchase_history SET listing_id = $2, listing_state = $3, +// attemptCancelWriteSQL is the row change the cancel endpoint makes when it +// records the attempt's listing canceled, run inside a held transaction. +const attemptCancelWriteSQL = `UPDATE purchase_history SET listing_id = $2, listing_state = $3, listing_client_token = NULL, listing_price_schedule = NULL WHERE purchase_id = $1` -// listWriteSQL is the row change of the list request recording its listing. -const listWriteSQL = `UPDATE purchase_history SET listing_id = $2, listing_state = $3, +// attemptListWriteSQL is the row change of the list request recording its +// listing. +const attemptListWriteSQL = `UPDATE purchase_history SET listing_id = $2, listing_state = $3, listing_client_token = NULL, listing_price_schedule = NULL WHERE purchase_id = $1` // The cancel's write commits while the list request's record is queued on @@ -156,7 +157,7 @@ func checkLateListRecordLoses(ctx context.Context, t *testing.T, store *Postgres tx, err := pool.Begin(ctx) require.NoError(t, err) defer func() { _ = tx.Rollback(ctx) }() - _, err = tx.Exec(ctx, cancelWriteSQL, id, "ril-L", ListingStateCancelled) + _, err = tx.Exec(ctx, attemptCancelWriteSQL, id, "ril-L", ListingStateCancelled) require.NoError(t, err) done := make(chan error, 1) @@ -178,7 +179,7 @@ func checkLateCancelStillWins(ctx context.Context, t *testing.T, store *Postgres tx, err := pool.Begin(ctx) require.NoError(t, err) defer func() { _ = tx.Rollback(ctx) }() - _, err = tx.Exec(ctx, listWriteSQL, id, "ril-C", ListingStateActive) + _, err = tx.Exec(ctx, attemptListWriteSQL, id, "ril-C", ListingStateActive) require.NoError(t, err) type result struct {