fix(api): keep marketplace listing tracked when rollback cancel fails

internal/api/handler_marketplace.go

@@ -208,11 +208,16 @@ func (h *Handler) marketplaceList(ctx context.Context, req *events.LambdaFunctio
 	}, nil
 }
 
+// marketplaceCompensationTimeout bounds the detached cancel and recording write
+// run after a created listing failed to persist.
+const marketplaceCompensationTimeout = 10 * time.Second
+
 // reserveAndCreateListing atomically claims the marketplace-listing slot for the
 // row, creates the AWS listing, and persists it, releasing the claim on every
-// failure path so a failed attempt never leaves the row stuck in the transient
-// pending state. Extracted from marketplaceList to keep that handler under the
-// gocyclo budget. Returns the persisted listing result on success.
+// failure path that leaves no live listing so a failed attempt does not leave
+// the row stuck in the transient pending state. Extracted from marketplaceList
+// to keep that handler under the gocyclo budget. Returns the persisted listing
+// result on success.
 func (h *Handler) reserveAndCreateListing(ctx context.Context, purchaseID string, row *config.PurchaseHistoryRecord, ec2Client marketplaceEC2Client, awsSchedule []ec2svc.MarketplacePriceTier) (ec2svc.MarketplaceListingResult, error) {
 	// List every RI in the row: a row of N Standard RIs must list all N, not a
 	// single unit (issue #292 multi-count fix). Floor at 1 for legacy rows that
@@ -260,18 +265,38 @@ func (h *Handler) reserveAndCreateListing(ctx context.Context, purchaseID string
 	// release the claim so the row does not stay stuck in the pending state.
 	if dbErr := h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, result.ListingID, result.State); dbErr != nil {
 		logging.Errorf("marketplace: listing created (%s / %s) but DB update failed: %v; attempting rollback", result.ListingID, result.State, dbErr)
-		if _, rollbackErr := ec2Client.CancelMarketplaceListing(ctx, result.ListingID); rollbackErr != nil {
-			logging.Errorf("marketplace: rollback cancel for listing %s also failed: %v", result.ListingID, rollbackErr)
-		} else {
-			logging.Warnf("marketplace: listing %s rolled back (canceled) after DB failure", result.ListingID)
+		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.releaseMarketplaceClaim(ctx, purchaseID, row)
 		return ec2svc.MarketplaceListingResult{}, fmt.Errorf("listing created but could not be persisted; listing has been rolled back: %w", dbErr)
 	}
 
 	return result, nil
 }
 
+// keepUncanceledListing records a listing that could be neither persisted nor
+// canceled; releasing the claim instead would hide it from the cancel endpoint.
+func (h *Handler) keepUncanceledListing(ctx context.Context, purchaseID string, listing ec2svc.MarketplaceListingResult, cancelErr error) error {
+	logging.Errorf("marketplace: rollback cancel for listing %s failed: %v", listing.ListingID, cancelErr)
+	state := listing.State
+	if state == "" {
+		state = config.ListingStatePending
+	}
+	msg := fmt.Sprintf("marketplace listing %s was created but could not be saved, and canceling it failed (%v); the listing may still be active on AWS", listing.ListingID, cancelErr)
+	if err := h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, listing.ListingID, state); err != nil {
+		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")
+	}
+	if state != config.ListingStateActive {
+		return NewClientError(502, fmt.Sprintf("%s; it is recorded on this RI as %s and needs reconciliation (listing %s)", msg, state, listing.ListingID))
+	}
+	return NewClientError(502, msg+"; it is now recorded on this RI, so cancel it again")
+}
+
 // populateOfferingClass calls AWS DescribeReservedInstances to determine the
 // offering class for an RI whose purchase_history row lacks one, then persists
 // the result so subsequent requests are served from the DB. Returns the class

internal/api/handler_marketplace_compensation_integration_test.go

@@ -0,0 +1,192 @@
+//go:build integration
+
+package api
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	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/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+// failFirstListingWriteStore fails only the first UpdatePurchaseHistoryListing,
+// simulating a transient DB error right after AWS created the listing.
+type failFirstListingWriteStore struct {
+	config.StoreInterface
+	failed bool
+}
+
+func (s *failFirstListingWriteStore) UpdatePurchaseHistoryListing(ctx context.Context, purchaseID, listingID, listingState string) error {
+	if !s.failed {
+		s.failed = true
+		return errors.New("transient db failure")
+	}
+	return s.StoreInterface.UpdatePurchaseHistoryListing(ctx, purchaseID, listingID, listingState)
+}
+
+func TestMarketplaceListCancelFailureKeepsListingRecorded(t *testing.T) {
+	ctx := t.Context()
+	pg, err := testhelpers.SetupPostgresContainer(ctx, t)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, pg.Cleanup(context.Background())) })
+	require.NoError(t, migrations.RunMigrations(ctx, pg.DB.Pool(), "../database/postgres/migrations", "", ""))
+	store := config.NewPostgresStore(pg.DB)
+
+	account := uuid.NewString()
+	require.NoError(t, store.CreateCloudAccount(ctx, &config.CloudAccount{ID: account, Name: account, Provider: "aws", ExternalID: "111111111111", Enabled: true}))
+	row := standardRow()
+	row.PurchaseID, row.CloudAccountID, row.Provider, row.Service = uuid.NewString(), &account, "aws", "ec2"
+	require.NoError(t, store.SavePurchaseHistory(ctx, row))
+
+	authSvc := &MockAuthService{}
+	authSvc.On("ValidateSession", mock.Anything, "test-token").Return(&Session{UserID: "admin", Email: "admin@test.com"}, nil)
+	authSvc.On("HasPermissionAPI", mock.Anything, "admin", auth.ActionSellAny, auth.ResourcePurchases).Return(true, nil)
+	authSvc.On("GetAllowedAccountsAPI", mock.Anything, "admin").Return([]string{"*"}, nil)
+	authSvc.On("HasPermissionForConstraintsAPI", mock.Anything, "admin", auth.ActionSellAny, auth.ResourcePurchases, mock.Anything).Return(true, nil)
+
+	cancelErr := errors.New("RequestLimitExceeded")
+	ec2 := &stubMarketplaceEC2{cancelFn: func(_ context.Context, id string) (ec2svc.MarketplaceListingResult, error) {
+		if cancelErr != nil {
+			return ec2svc.MarketplaceListingResult{}, cancelErr
+		}
+		return ec2svc.MarketplaceListingResult{ListingID: id, State: config.ListingStateCancelled}, nil
+	}}
+	h := &Handler{config: &failFirstListingWriteStore{StoreInterface: store}, auth: authSvc,
+		marketplaceEC2Factory: func(_ aws.Config) marketplaceEC2Client { return ec2 }}
+	h.awsCfgOnce.Do(func() { h.awsCfg = aws.Config{Region: "us-east-1"} })
+
+	_, err = h.marketplaceList(ctx, marketplaceReq(), row.PurchaseID)
+	require.Error(t, err)
+	ce, ok := IsClientError(err)
+	require.True(t, ok, "%v", err)
+	assert.Equal(t, 502, ce.code)
+	assert.NotContains(t, ce.message, "rolled back")
+	assert.Contains(t, ce.message, "RequestLimitExceeded")
+
+	stored, err := store.GetPurchaseHistoryByPurchaseID(ctx, row.PurchaseID)
+	require.NoError(t, err)
+	assert.Equal(t, "ril-default", stored.ListingID)
+	assert.Equal(t, config.ListingStateActive, stored.ListingState)
+
+	_, err = h.marketplaceList(ctx, marketplaceReq(), row.PurchaseID)
+	require.Error(t, err, "a second listing must not be created while the first is live")
+	assert.Equal(t, 1, ec2.createCallCount)
+
+	cancelErr = nil
+	_, err = h.marketplaceCancel(ctx, marketplaceReq(), row.PurchaseID)
+	require.NoError(t, err)
+	stored, err = store.GetPurchaseHistoryByPurchaseID(ctx, row.PurchaseID)
+	require.NoError(t, err)
+	assert.Equal(t, "ril-default", stored.ListingID)
+	assert.Equal(t, config.ListingStateCancelled, stored.ListingState)
+}
+
+// cancelOnFirstWriteStore cancels the request context while failing the first
+// listing write, as a request deadline would, then delegates to the real store.
+type cancelOnFirstWriteStore struct {
+	config.StoreInterface
+	cancel context.CancelFunc
+	failed bool
+}
+
+func (s *cancelOnFirstWriteStore) UpdatePurchaseHistoryListing(ctx context.Context, purchaseID, listingID, listingState string) error {
+	if !s.failed {
+		s.failed = true
+		s.cancel()
+		return context.Canceled
+	}
+	return s.StoreInterface.UpdatePurchaseHistoryListing(ctx, purchaseID, listingID, listingState)
+}
+
+func setupCompensationRow(t *testing.T, store config.StoreInterface) *config.PurchaseHistoryRecord {
+	t.Helper()
+	ctx := t.Context()
+	account := uuid.NewString()
+	require.NoError(t, store.CreateCloudAccount(ctx, &config.CloudAccount{ID: account, Name: account, Provider: "aws", ExternalID: "111111111111", Enabled: true}))
+	row := standardRow()
+	row.PurchaseID, row.CloudAccountID, row.Provider, row.Service = uuid.NewString(), &account, "aws", "ec2"
+	require.NoError(t, store.SavePurchaseHistory(ctx, row))
+	return row
+}
+
+func newCompensationAuth() *MockAuthService {
+	authSvc := &MockAuthService{}
+	authSvc.On("ValidateSession", mock.Anything, "test-token").Return(&Session{UserID: "admin", Email: "admin@test.com"}, nil)
+	authSvc.On("HasPermissionAPI", mock.Anything, "admin", auth.ActionSellAny, auth.ResourcePurchases).Return(true, nil)
+	authSvc.On("GetAllowedAccountsAPI", mock.Anything, "admin").Return([]string{"*"}, nil)
+	authSvc.On("HasPermissionForConstraintsAPI", mock.Anything, "admin", auth.ActionSellAny, auth.ResourcePurchases, mock.Anything).Return(true, nil)
+	return authSvc
+}
+
+func TestMarketplaceListCancelFailureAfterContextCanceledStillRecordsListing(t *testing.T) {
+	ctx := t.Context()
+	pg, err := testhelpers.SetupPostgresContainer(ctx, t)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, pg.Cleanup(context.Background())) })
+	require.NoError(t, migrations.RunMigrations(ctx, pg.DB.Pool(), "../database/postgres/migrations", "", ""))
+	store := config.NewPostgresStore(pg.DB)
+	row := setupCompensationRow(t, store)
+
+	reqCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	ec2 := &stubMarketplaceEC2{cancelFn: func(_ context.Context, _ string) (ec2svc.MarketplaceListingResult, error) {
+		return ec2svc.MarketplaceListingResult{}, errors.New("RequestLimitExceeded")
+	}}
+	h := &Handler{config: &cancelOnFirstWriteStore{StoreInterface: store, cancel: cancel}, auth: newCompensationAuth(),
+		marketplaceEC2Factory: func(_ aws.Config) marketplaceEC2Client { return ec2 }}
+	h.awsCfgOnce.Do(func() { h.awsCfg = aws.Config{Region: "us-east-1"} })
+
+	_, err = h.marketplaceList(reqCtx, marketplaceReq(), row.PurchaseID)
+	require.Error(t, err)
+
+	stored, err := store.GetPurchaseHistoryByPurchaseID(ctx, row.PurchaseID)
+	require.NoError(t, err)
+	assert.Equal(t, "ril-default", stored.ListingID)
+	assert.Equal(t, config.ListingStateActive, stored.ListingState)
+}
+
+func TestMarketplaceListCancelFailureRecordsEmptyAWSStateAsPending(t *testing.T) {
+	ctx := t.Context()
+	pg, err := testhelpers.SetupPostgresContainer(ctx, t)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, pg.Cleanup(context.Background())) })
+	require.NoError(t, migrations.RunMigrations(ctx, pg.DB.Pool(), "../database/postgres/migrations", "", ""))
+	store := config.NewPostgresStore(pg.DB)
+	row := setupCompensationRow(t, store)
+
+	ec2 := &stubMarketplaceEC2{
+		createFn: func(_ context.Context, _ ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) {
+			return ec2svc.MarketplaceListingResult{ListingID: "ril-empty"}, nil
+		},
+		cancelFn: 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 }}
+	h.awsCfgOnce.Do(func() { h.awsCfg = aws.Config{Region: "us-east-1"} })
+
+	_, err = h.marketplaceList(ctx, marketplaceReq(), row.PurchaseID)
+	ce, ok := IsClientError(err)
+	require.True(t, ok, "%v", err)
+	assert.Contains(t, ce.message, "needs reconciliation")
+
+	stored, err := store.GetPurchaseHistoryByPurchaseID(ctx, row.PurchaseID)
+	require.NoError(t, err)
+	assert.Equal(t, "ril-empty", stored.ListingID)
+	assert.Equal(t, config.ListingStatePending, stored.ListingState)
+
+	_, err = h.marketplaceList(ctx, marketplaceReq(), row.PurchaseID)
+	require.Error(t, err, "a duplicate listing must stay blocked")
+	assert.Equal(t, 1, ec2.createCallCount)
+}

internal/api/handler_marketplace_test.go

@@ -348,6 +348,128 @@ func TestMarketplaceList_DBFailureCompensatingRollback(t *testing.T) {
 	cfgStore.AssertExpectations(t)
 }
 
+func TestMarketplaceList_DBFailureAndCancelFailureKeepsListing(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		recordErr error
+		wantMsg   string
+	}{
+		{name: "listing recorded", wantMsg: "now recorded on this RI"},
+		{name: "db still down", recordErr: errors.New("db still down"), wantMsg: "stays locked as pending"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			cfgStore := &MockConfigStore{}
+			authSvc := &MockAuthService{}
+			adminSession(authSvc)
+			cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).
+				Return(standardRow(), nil)
+			cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID).
+				Return(true, nil)
+			cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive).
+				Return(errors.New("db down")).Once()
+			cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive).
+				Return(tc.recordErr).Once()
+
+			ec2 := &stubMarketplaceEC2{
+				cancelFn: 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)
+
+			require.Error(t, err)
+			ce, ok := IsClientError(err)
+			require.True(t, ok, "operator must see the message, not a generic 500: %v", err)
+			assert.Equal(t, 502, ce.code)
+			assert.NotContains(t, ce.message, "rolled back")
+			assert.Contains(t, ce.message, "ril-default")
+			assert.Contains(t, ce.message, "throttled")
+			assert.Contains(t, ce.message, tc.wantMsg)
+			assert.Equal(t, 1, ec2.cancelCallCount)
+			cfgStore.AssertNotCalled(t, "UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "", "")
+			cfgStore.AssertExpectations(t)
+		})
+	}
+}
+
+func TestMarketplaceList_CancelFailureRecordsNonActiveStateAsPending(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		awsState   string
+		wantState  string
+		wantMsg    string
+		wantNotMsg string
+	}{
+		{name: "pending", awsState: config.ListingStatePending, wantState: config.ListingStatePending, wantMsg: "needs reconciliation", wantNotMsg: "cancel it again"},
+		{name: "empty", awsState: "", wantState: config.ListingStatePending, wantMsg: "needs reconciliation", wantNotMsg: "cancel it again"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			cfgStore := &MockConfigStore{}
+			authSvc := &MockAuthService{}
+			adminSession(authSvc)
+			cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).
+				Return(standardRow(), nil)
+			cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID).
+				Return(true, nil)
+			cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-x", tc.awsState).
+				Return(errors.New("db down")).Once()
+			cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-x", tc.wantState).
+				Return(nil).Once()
+
+			ec2 := &stubMarketplaceEC2{
+				createFn: func(_ context.Context, _ ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error) {
+					return ec2svc.MarketplaceListingResult{ListingID: "ril-x", State: tc.awsState}, nil
+				},
+				cancelFn: 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)
+
+			ce, ok := IsClientError(err)
+			require.True(t, ok, "%v", err)
+			assert.Equal(t, 502, ce.code)
+			assert.Contains(t, ce.message, tc.wantMsg)
+			assert.Contains(t, ce.message, "ril-x")
+			assert.NotContains(t, ce.message, tc.wantNotMsg)
+			cfgStore.AssertExpectations(t)
+		})
+	}
+}
+
+func TestMarketplaceList_CancelFailureAfterRequestContextCanceledStillRecords(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cfgStore := &MockConfigStore{}
+	authSvc := &MockAuthService{}
+	adminSession(authSvc)
+	cfgStore.On("GetPurchaseHistoryByPurchaseID", mock.Anything, validMarketplacePurchaseID).
+		Return(standardRow(), nil)
+	cfgStore.On("ClaimMarketplaceListingSlot", mock.Anything, validMarketplacePurchaseID).
+		Return(true, nil)
+	// The request context dies during the first write, as on a deadline.
+	cfgStore.On("UpdatePurchaseHistoryListing", mock.Anything, validMarketplacePurchaseID, "ril-default", config.ListingStateActive).
+		Run(func(mock.Arguments) { cancel() }).Return(errors.New("db down")).Once()
+	liveCtx := mock.MatchedBy(func(c context.Context) bool { return c.Err() == nil })
+	cfgStore.On("UpdatePurchaseHistoryListing", liveCtx, validMarketplacePurchaseID, "ril-default", config.ListingStateActive).
+		Return(nil).Once()
+
+	ec2 := &stubMarketplaceEC2{
+		cancelFn: func(c context.Context, _ string) (ec2svc.MarketplaceListingResult, error) {
+			require.NoError(t, c.Err(), "compensating cancel must not run on the canceled request context")
+			return ec2svc.MarketplaceListingResult{}, errors.New("throttled")
+		},
+	}
+	h := newMarketplaceHandler(cfgStore, authSvc, ec2)
+	_, err := h.marketplaceList(ctx, marketplaceReq(), validMarketplacePurchaseID)
+
+	ce, ok := IsClientError(err)
+	require.True(t, ok, "%v", err)
+	assert.Contains(t, ce.message, "now recorded on this RI")
+	cfgStore.AssertExpectations(t)
+}
+
 func TestMarketplaceList_DefaultScheduleProrationMath(t *testing.T) {
 	// remaining=12, term=12, count=1, upfront=1200 => per-unit residual
 	// 1200*12/12/1 = 1200; default price = 1200*0.95 = 1140.