diff --git a/internal/api/handler_marketplace.go b/internal/api/handler_marketplace.go index 285578bfe..cacdfc876 100644 --- a/internal/api/handler_marketplace.go +++ b/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,11 +265,12 @@ 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) } @@ -272,6 +278,25 @@ func (h *Handler) reserveAndCreateListing(ctx context.Context, purchaseID string 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 diff --git a/internal/api/handler_marketplace_compensation_integration_test.go b/internal/api/handler_marketplace_compensation_integration_test.go new file mode 100644 index 000000000..8cb4aed22 --- /dev/null +++ b/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) +} diff --git a/internal/api/handler_marketplace_test.go b/internal/api/handler_marketplace_test.go index 0fb8ec684..909c7005c 100644 --- a/internal/api/handler_marketplace_test.go +++ b/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.