fix(marketplace): recover pending listings through cancel and never record an empty or revived state

internal/analytics/collector_test.go

@@ -532,6 +532,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
 }

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)

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,14 +220,19 @@ 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()
 		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)
+		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)
 	}
 }
 

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 }