diff --git a/internal/api/handler.go b/internal/api/handler.go index 15b81e1dc..46981775e 100644 --- a/internal/api/handler.go +++ b/internal/api/handler.go @@ -129,7 +129,7 @@ type Handler struct { commitmentOpts CommitmentOptsInterface // executeExchangeFn is the RI exchange execution function injected by tests. - // When nil (the production default), executeApprovedExchange calls + // When nil (the production default), awsExchangeExecutor returns // exchange.ExecuteExchange directly. Tests inject a stub so the handler // can be exercised end-to-end without live AWS credentials or a real RI. executeExchangeFn func(ctx context.Context, req exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) diff --git a/internal/api/handler_ri_exchange.go b/internal/api/handler_ri_exchange.go index 2d601f6b6..e622b1256 100644 --- a/internal/api/handler_ri_exchange.go +++ b/internal/api/handler_ri_exchange.go @@ -28,6 +28,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" "github.com/LeanerCloud/cloud-commitments-platform/internal/credentials" + "github.com/LeanerCloud/cloud-commitments-platform/internal/ctxutil" ) // reshapeEC2Client is the narrow slice of the EC2 client that @@ -1234,12 +1235,17 @@ func (h *Handler) executeAzureExchange(ctx context.Context, req *events.LambdaFu return nil, err } - result, err := client.ExecuteExchange(ctx, preview.SessionID) + // Past the claim the exchange may commit at Azure; a cancelled request must + // not abort the submit or its long-running-operation poll (#790). + execCtx, cancelExec := ctxutil.DetachedWithTimeout(ctx, riExchangeExecutionTimeout) + defer cancelExec() + result, err := client.ExecuteExchange(execCtx, preview.SessionID) if err != nil { logging.Errorf("azure exchange execution failed: %v", err) // Non-4xx failures here are ambiguous: BeginPost may already have - // submitted the exchange, and a ctx cancellation mid-poll looks - // identical to one that never reached Azure. Say so rather than + // submitted the exchange, and exhausting the riExchangeExecutionTimeout + // budget mid-poll looks identical to a call that never reached Azure + // (a request cancel no longer reaches execCtx). Say so rather than // reporting a flat failure that invites the retry this claim now // refuses (#1642). return nil, mapAzureExchangeError( @@ -1918,7 +1924,11 @@ func (h *Handler) executeExchange(ctx context.Context, req *events.LambdaFunctio return nil, err } - exchangeID, quote, err := exchange.ExecuteExchange(ctx, exchange.ExchangeExecuteRequest{ + // Past the claim the exchange may commit at AWS; a cancelled request must + // not abort it midway (#790). + execCtx, cancelExec := ctxutil.DetachedWithTimeout(ctx, riExchangeExecutionTimeout) + defer cancelExec() + exchangeID, quote, err := h.awsExchangeExecutor()(execCtx, exchange.ExchangeExecuteRequest{ Region: region, ReservedIDs: body.RIIDs, Targets: toExchangeTargets(body.Targets), @@ -2302,6 +2312,9 @@ func (h *Handler) approveRIExchangeViaToken(ctx context.Context, req *events.Lam // stampRIExchangeApprover records session.Email as approved_by. Best-effort: the // exchange already executed, so a stamp failure is logged, not surfaced. func (h *Handler) stampRIExchangeApprover(ctx context.Context, id string, session *Session) { + // Runs after money moved, so the audit write must not die with the request (#790). + ctx, cancel := ctxutil.DetachedWithTimeout(ctx, riExchangeTerminalSaveTimeout) + defer cancel() if err := h.config.StampRIExchangeApprovedBy(ctx, id, session.Email); err != nil { logging.Errorf("failed to stamp approved_by on exchange %s: %v", id, err) } @@ -2566,7 +2579,16 @@ func (h *Handler) validateExchangeApproval(ctx context.Context, id, token string } // failExchange marks an exchange as failed, logging if the DB write also fails. +// +// Every caller runs after the pending -> processing transition committed, so +// the write is detached from the request (#790): a cancel there would +// otherwise strand the record in "processing", where the approval retry gets +// 409 from the transition guard and the row keeps counting against the daily cap. +// +//nolint:unparam // (result, error) shape lets executeApprovedExchange tail-call it; a failed exchange is a 200 "failed" body, not an error func (h *Handler) failExchange(ctx context.Context, id, reason string) (any, error) { + ctx, cancel := ctxutil.DetachedWithTimeout(ctx, riExchangeTerminalSaveTimeout) + defer cancel() if failErr := h.config.FailRIExchange(ctx, id, reason); failErr != nil { logging.Errorf("failed to mark exchange %s as failed (DB may be unavailable): %v", id, failErr) } @@ -2659,19 +2681,21 @@ func (h *Handler) executeApprovedExchange(ctx context.Context, id string, record return h.failExchange(ctx, id, fmt.Sprintf("reserved ceiling %q is not a decimal", ceiling)) } - execFn := exchange.ExecuteExchange - if h.executeExchangeFn != nil { - execFn = h.executeExchangeFn - } - exchangeID, freshQ, execErr := execFn(ctx, exchange.ExchangeExecuteRequest{ + // The provider call and the terminal ledger write that records its outcome + // must survive a cancelled approval request (#790). + execCtx, cancelExec := ctxutil.DetachedWithTimeout(ctx, riExchangeExecutionTimeout) + defer cancelExec() + exchangeID, freshQ, execErr := h.awsExchangeExecutor()(execCtx, exchange.ExchangeExecuteRequest{ Region: region, ReservedIDs: record.SourceRIIDs, TargetOfferingID: record.TargetOfferingID, TargetCount: int32(record.TargetCount), // #nosec G115 -- RI quantity stored from validated API request; AWS limits RI counts well below math.MaxInt32 MaxPaymentDueUSD: effectiveCap, }) + saveCtx, cancelSave := ctxutil.DetachedWithTimeout(ctx, riExchangeTerminalSaveTimeout) + defer cancelSave() if execErr != nil { - return h.failExchange(ctx, id, execErr.Error()) + return h.failExchange(saveCtx, id, execErr.Error()) } // H3: persist the amount AWS actually accepted, not the reserved ceiling. @@ -2680,7 +2704,7 @@ func (h *Handler) executeApprovedExchange(ctx context.Context, id string, record // H4: retry the ledger write via retryCompleteWithPayment; persistent // failure is returned as an error (HTTP 500) so the caller knows money // moved but the record was not updated. - if completeErr := h.retryCompleteWithPayment(ctx, id, exchangeID, acceptedPaymentDue); completeErr != nil { + if completeErr := h.retryCompleteWithPayment(saveCtx, id, exchangeID, acceptedPaymentDue); completeErr != nil { logging.Errorf("all ledger write attempts failed for exchange %s after money moved: %v", id, completeErr) return nil, fmt.Errorf("exchange executed (id=%s) but ledger update failed: %w", diff --git a/internal/api/handler_ri_exchange_azure_test.go b/internal/api/handler_ri_exchange_azure_test.go index 9a221005b..e8f084f32 100644 --- a/internal/api/handler_ri_exchange_azure_test.go +++ b/internal/api/handler_ri_exchange_azure_test.go @@ -579,7 +579,7 @@ func TestExecuteAzureExchange_InScopeSubscriptionAllowed(t *testing.T) { &azurecompute.ExchangePreview{SessionID: "sess-fresh", NetPayable: toPtr(10.00), NetPayableCurrency: "USD"}, []azurecompute.CompatibleOffering{}, nil, ) - opsClient.On("ExecuteExchange", ctx, "sess-fresh").Return( + opsClient.On("ExecuteExchange", mock.Anything, "sess-fresh").Return( &azurecompute.ExchangeResult{SessionID: "sess-fresh", Status: "Succeeded"}, nil, ) t.Cleanup(func() { opsClient.AssertExpectations(t) }) @@ -872,7 +872,7 @@ func TestExecuteAzureExchange_USDCurrencyCapStillEnforced(t *testing.T) { &azurecompute.ExchangePreview{SessionID: "sess-usd-ok", NetPayable: toPtr(50.00), NetPayableCurrency: "USD"}, []azurecompute.CompatibleOffering{}, nil, ) - opsClient.On("ExecuteExchange", ctx, "sess-usd-ok").Return(&azurecompute.ExchangeResult{ + opsClient.On("ExecuteExchange", mock.Anything, "sess-usd-ok").Return(&azurecompute.ExchangeResult{ SessionID: "sess-usd-ok", NetPayable: toPtr(50.00), NetPayableCurrency: "USD", Status: "Succeeded", }, nil) t.Cleanup(func() { opsClient.AssertExpectations(t) }) @@ -917,7 +917,7 @@ func TestExecuteAzureExchange_NonUSDWithoutCapStillAllowed(t *testing.T) { &azurecompute.ExchangePreview{SessionID: "sess-kwd-ok", NetPayable: toPtr(500.00), NetPayableCurrency: "KWD"}, []azurecompute.CompatibleOffering{}, nil, ) - opsClient.On("ExecuteExchange", ctx, "sess-kwd-ok").Return(&azurecompute.ExchangeResult{ + opsClient.On("ExecuteExchange", mock.Anything, "sess-kwd-ok").Return(&azurecompute.ExchangeResult{ SessionID: "sess-kwd-ok", NetPayable: toPtr(500.00), NetPayableCurrency: "KWD", Status: "Succeeded", }, nil) t.Cleanup(func() { opsClient.AssertExpectations(t) }) @@ -1181,7 +1181,7 @@ func TestExecuteAzureExchange_HappyPath(t *testing.T) { }, []azurecompute.CompatibleOffering{}, nil, ) - opsClient.On("ExecuteExchange", ctx, mock.MatchedBy(func(sessionID string) bool { + opsClient.On("ExecuteExchange", mock.Anything, mock.MatchedBy(func(sessionID string) bool { return sessionID == freshSessionID })).Return(&azurecompute.ExchangeResult{ SessionID: freshSessionID, @@ -1267,7 +1267,7 @@ func TestExecuteAzureExchange_ConstraintSetPinsAllDimensions(t *testing.T) { &azurecompute.ExchangePreview{SessionID: "sess-dims", NetPayable: toPtr(10.00), NetPayableCurrency: "USD"}, []azurecompute.CompatibleOffering{}, nil, ) - opsClient.On("ExecuteExchange", ctx, "sess-dims").Return( + opsClient.On("ExecuteExchange", mock.Anything, "sess-dims").Return( &azurecompute.ExchangeResult{SessionID: "sess-dims", Status: "Succeeded"}, nil, ) t.Cleanup(func() { opsClient.AssertExpectations(t) }) @@ -1380,7 +1380,7 @@ func TestExecuteAzureExchange_ConstraintRegionsAreLowerCased(t *testing.T) { &azurecompute.ExchangePreview{SessionID: "sess-case", NetPayable: toPtr(10.00), NetPayableCurrency: "USD"}, []azurecompute.CompatibleOffering{}, nil, ) - opsClient.On("ExecuteExchange", ctx, "sess-case").Return( + opsClient.On("ExecuteExchange", mock.Anything, "sess-case").Return( &azurecompute.ExchangeResult{SessionID: "sess-case", Status: "Succeeded"}, nil, ) t.Cleanup(func() { opsClient.AssertExpectations(t) }) @@ -1439,7 +1439,7 @@ func TestExecuteAzureExchange_LowercaseUSDTakesUSDCapPath(t *testing.T) { &azurecompute.ExchangePreview{SessionID: "sess-lower-usd", NetPayable: toPtr(50.00), NetPayableCurrency: "USD"}, []azurecompute.CompatibleOffering{}, nil, ) - opsClient.On("ExecuteExchange", ctx, "sess-lower-usd").Return( + opsClient.On("ExecuteExchange", mock.Anything, "sess-lower-usd").Return( &azurecompute.ExchangeResult{SessionID: "sess-lower-usd", Status: "Succeeded"}, nil, ) t.Cleanup(func() { opsClient.AssertExpectations(t) }) @@ -1560,7 +1560,7 @@ func TestExecuteAzureExchange_CommitFailureClassification(t *testing.T) { &azurecompute.ExchangePreview{SessionID: "sess-commit-fail", NetPayable: toPtr(10.00), NetPayableCurrency: "USD"}, []azurecompute.CompatibleOffering{}, nil, ) - opsClient.On("ExecuteExchange", ctx, "sess-commit-fail").Return(nil, tt.commitErr) + opsClient.On("ExecuteExchange", mock.Anything, "sess-commit-fail").Return(nil, tt.commitErr) t.Cleanup(func() { opsClient.AssertExpectations(t) }) h := newAzureExecuteMoneyPathHandler(t, opsClient) @@ -2054,7 +2054,7 @@ func TestExecuteAzureExchange_UnreportedSourceRegionAllowedWithoutRegionScope(t &azurecompute.ExchangePreview{SessionID: "sess-shared", NetPayable: toPtr(10.00), NetPayableCurrency: "USD"}, []azurecompute.CompatibleOffering{}, nil, ) - opsClient.On("ExecuteExchange", ctx, "sess-shared").Return( + opsClient.On("ExecuteExchange", mock.Anything, "sess-shared").Return( &azurecompute.ExchangeResult{SessionID: "sess-shared", Status: "Succeeded"}, nil, ) t.Cleanup(func() { opsClient.AssertExpectations(t) }) diff --git a/internal/api/handler_ri_exchange_test.go b/internal/api/handler_ri_exchange_test.go index 188c13e81..4c835b300 100644 --- a/internal/api/handler_ri_exchange_test.go +++ b/internal/api/handler_ri_exchange_test.go @@ -402,8 +402,8 @@ func TestApproveRIExchange_SessionAdmin(t *testing.T) { // executeApprovedExchange will call failExchange (no real AWS SDK) which // returns a non-error result map. Since execErr == nil, StampRIExchangeApprovedBy // is called as best-effort audit trail. - mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil) - mockStore.On("StampRIExchangeApprovedBy", ctx, id, adminSession.Email).Return(nil) + mockStore.On("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil) + mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email).Return(nil) req := &events.LambdaFunctionURLRequest{ Headers: map[string]string{ @@ -417,8 +417,8 @@ func TestApproveRIExchange_SessionAdmin(t *testing.T) { // dispatch reached the session-authed path. mockStore.AssertCalled(t, "GetRIExchangeRecord", ctx, id) mockStore.AssertCalled(t, "TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.Anything) - mockStore.AssertCalled(t, "FailRIExchange", ctx, id, mock.AnythingOfType("string")) - mockStore.AssertCalled(t, "StampRIExchangeApprovedBy", ctx, id, adminSession.Email) + mockStore.AssertCalled(t, "FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")) + mockStore.AssertCalled(t, "StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email) } // TestApproveRIExchange_SessionApproveOwn verifies that a user with approve-own @@ -459,8 +459,8 @@ func TestApproveRIExchange_SessionApproveOwn(t *testing.T) { RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, }, nil) - mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil) - mockStore.On("StampRIExchangeApprovedBy", ctx, id, ownerSession.Email).Return(nil) + mockStore.On("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil) + mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, ownerSession.Email).Return(nil) req := &events.LambdaFunctionURLRequest{ Headers: map[string]string{ @@ -536,8 +536,8 @@ func TestApproveRIExchange_SessionScope(t *testing.T) { mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.Anything). Return(&config.RIExchangeRecord{ID: id, Status: "processing", SourceRIIDs: []string{"ri-1"}, PaymentDue: "50.00"}, nil).Maybe() mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil).Maybe() - mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil).Maybe() - mockStore.On("StampRIExchangeApprovedBy", ctx, id, mock.Anything).Return(nil).Maybe() + mockStore.On("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil).Maybe() + mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, mock.Anything).Return(nil).Maybe() return &Handler{ config: mockStore, @@ -629,8 +629,8 @@ func fourEyesRIExchangeFixture(t *testing.T, creatorID *string, approverID strin mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.Anything). Return(&config.RIExchangeRecord{ID: id, Status: "processing", SourceRIIDs: []string{"ri-1"}, PaymentDue: "50.00"}, nil) mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil) - mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil) - mockStore.On("StampRIExchangeApprovedBy", ctx, id, mock.Anything).Return(nil) + mockStore.On("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil) + mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, mock.Anything).Return(nil) return &Handler{config: mockStore, auth: mockAuth}, mockStore, id, token } @@ -730,7 +730,7 @@ func TestApproveRIExchange_TokenPathAttributesResolvedApprover(t *testing.T) { _, err := h.approveRIExchange(ctx, sessionReq("sess"), id, token) require.NoError(t, err) store.AssertCalled(t, "TransitionRIExchangeStatus", ctx, id, "pending", "processing", &approver) - store.AssertCalled(t, "StampRIExchangeApprovedBy", ctx, id, approver+"@example.com") + store.AssertCalled(t, "StampRIExchangeApprovedBy", mock.Anything, id, approver+"@example.com") }) t.Run("no session leaves the actor unset", func(t *testing.T) { @@ -772,7 +772,7 @@ func TestApproveRIExchange_TokenPathSkipsStampWhenExecutionErrors(t *testing.T) mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil) mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", &approver). Return(&config.RIExchangeRecord{ID: id, Status: "processing"}, nil) - mockStore.On("CompleteRIExchangeWithPayment", ctx, id, "exch-518", "0"). + mockStore.On("CompleteRIExchangeWithPayment", mock.Anything, id, "exch-518", "0"). Return(fmt.Errorf("DB write failed")).Times(3) h := &Handler{ @@ -813,7 +813,7 @@ func TestApproveRIExchange_LegacyTokenStillWorks(t *testing.T) { RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, }, nil) - mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil) + mockStore.On("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil) req := &events.LambdaFunctionURLRequest{} _, err := h.approveRIExchange(ctx, req, id, token) @@ -2244,7 +2244,7 @@ func TestExecuteApprovedExchange_EmptyRecordRegionFails(t *testing.T) { RIExchangeMaxPerExchangeUSD: 500, }, nil) // Record has no region — the handler must call failExchange, not proceed. - mockStore.On("FailRIExchange", ctx, id, mock.MatchedBy(func(reason string) bool { + mockStore.On("FailRIExchange", mock.Anything, id, mock.MatchedBy(func(reason string) bool { return len(reason) > 0 })).Return(nil) @@ -2332,8 +2332,8 @@ func TestApproveRIExchange_SessionActorStamped(t *testing.T) { mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, }, nil) - mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil) - mockStore.On("StampRIExchangeApprovedBy", ctx, id, adminSession.Email).Return(nil) + mockStore.On("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil) + mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email).Return(nil) req := &events.LambdaFunctionURLRequest{ Headers: map[string]string{ @@ -2398,7 +2398,7 @@ func TestExecuteApprovedExchange_EffectiveCap_BoundedByDailyHeadroom(t *testing. }, } - mockStore.On("CompleteRIExchangeWithPayment", ctx, id, "exch-h2-handler", "40.000000").Return(nil) + mockStore.On("CompleteRIExchangeWithPayment", mock.Anything, id, "exch-h2-handler", "40.000000").Return(nil) record := &config.RIExchangeRecord{ ID: id, @@ -2452,7 +2452,7 @@ func TestExecuteApprovedExchange_AcceptedAmountFromFreshQuote(t *testing.T) { // The FRESH amount "35.500000" must reach CompleteRIExchangeWithPayment, // not the stale "30.00" from record.PaymentDue. - mockStore.On("CompleteRIExchangeWithPayment", ctx, id, "exch-h3-fresh", "35.500000").Return(nil) + mockStore.On("CompleteRIExchangeWithPayment", mock.Anything, id, "exch-h3-fresh", "35.500000").Return(nil) record := &config.RIExchangeRecord{ ID: id, @@ -2499,7 +2499,7 @@ func TestExecuteApprovedExchange_LedgerWriteFailure_ReturnsError(t *testing.T) { } // Fail all three retry attempts. - mockStore.On("CompleteRIExchangeWithPayment", ctx, id, "exch-h4-test", "0"). + mockStore.On("CompleteRIExchangeWithPayment", mock.Anything, id, "exch-h4-test", "0"). Return(fmt.Errorf("DB write failed")).Times(3) record := &config.RIExchangeRecord{ @@ -2599,7 +2599,7 @@ func TestExecuteApprovedExchange_ReservationRefused_NoExecute(t *testing.T) { }, nil) mockStore.On("ReserveApprovedRIExchange", ctx, id, "1000.000000", "500.000000"). Return("", fmt.Errorf("%w: initial 600.000000, remaining 400.000000", config.ErrRIExchangeDailyCapExceeded)) - mockStore.On("FailRIExchange", ctx, id, mock.MatchedBy(func(r string) bool { + mockStore.On("FailRIExchange", mock.Anything, id, mock.MatchedBy(func(r string) bool { return strings.Contains(r, "daily cap exceeded") })).Return(nil) @@ -2629,7 +2629,7 @@ func TestExecuteApprovedExchange_SettlesAcceptedAmountWithinReservedCeiling(t *t RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, }, nil) mockStore.On("ReserveApprovedRIExchange", ctx, id, "1000.000000", "500.000000").Return("400.000000", nil) - mockStore.On("CompleteRIExchangeWithPayment", ctx, id, "exch-1", "390.000000").Return(nil) + mockStore.On("CompleteRIExchangeWithPayment", mock.Anything, id, "exch-1", "390.000000").Return(nil) var got exchange.ExchangeExecuteRequest h := &Handler{config: mockStore, executeExchangeFn: func(_ context.Context, req exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { @@ -2658,7 +2658,7 @@ func TestExecuteApprovedExchange_ReservationInfrastructureError_HidesDetail(t *t }, nil) mockStore.On("ReserveApprovedRIExchange", ctx, id, "1000.000000", "500.000000"). Return("", errors.New("failed to lock RI exchange daily spend: connection reset by peer")) - mockStore.On("FailRIExchange", ctx, id, "daily spending cap check failed").Return(nil) + mockStore.On("FailRIExchange", mock.Anything, id, "daily spending cap check failed").Return(nil) called := false h := &Handler{config: mockStore, executeExchangeFn: func(context.Context, exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { diff --git a/internal/api/middleware_test.go b/internal/api/middleware_test.go index 6b046d4a5..11d376095 100644 --- a/internal/api/middleware_test.go +++ b/internal/api/middleware_test.go @@ -522,8 +522,8 @@ func TestApproveRIExchangeViaSession_RequiresCSRF(t *testing.T) { RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, }, nil).Maybe() - mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil).Maybe() - mockStore.On("StampRIExchangeApprovedBy", ctx, id, adminSession.Email).Return(nil).Maybe() + mockStore.On("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil).Maybe() + mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email).Return(nil).Maybe() h := &Handler{config: mockStore, auth: mockAuth} req := &events.LambdaFunctionURLRequest{ @@ -598,8 +598,8 @@ func TestApproveRIExchange_CSRFFailureDoesNotFallThroughToToken(t *testing.T) { RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, }, nil).Maybe() - mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil).Maybe() - mockStore.On("StampRIExchangeApprovedBy", ctx, id, adminSession.Email).Return(nil).Maybe() + mockStore.On("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil).Maybe() + mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email).Return(nil).Maybe() h := &Handler{config: mockStore, auth: mockAuth} req := &events.LambdaFunctionURLRequest{ @@ -654,8 +654,8 @@ func TestApproveRIExchangeViaSession_PassesCSRF(t *testing.T) { RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, }, nil) - mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil) - mockStore.On("StampRIExchangeApprovedBy", ctx, id, adminSession.Email).Return(nil) + mockStore.On("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil) + mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email).Return(nil) h := &Handler{config: mockStore, auth: mockAuth} req := &events.LambdaFunctionURLRequest{ diff --git a/internal/api/ri_exchange_detached.go b/internal/api/ri_exchange_detached.go new file mode 100644 index 000000000..7485306e2 --- /dev/null +++ b/internal/api/ri_exchange_detached.go @@ -0,0 +1,34 @@ +package api + +import ( + "context" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/exchange" +) + +// Budgets for RI exchange work that must outlive the HTTP request (#790). +// +// Once an exchange is submitted to the provider it cannot be undone, so a +// client disconnect or the container server's 30s per-request timeout must not +// cancel the provider call or the terminal ledger write that records it. Both +// run on contexts that keep the request's values but drop its cancellation, +// each bounded by its own budget so a hung provider or database cannot pin the +// goroutine forever. On Lambda the invocation itself is still bounded by the +// function timeout, which no context can extend. +// +// The detaching itself is ctxutil.DetachedWithTimeout, shared with the +// purchase path (#706 / #789). +const ( + riExchangeExecutionTimeout = 4 * time.Minute + riExchangeTerminalSaveTimeout = 30 * time.Second +) + +// awsExchangeExecutor returns the AWS RI exchange execution function: the +// test-injected executeExchangeFn when set, exchange.ExecuteExchange otherwise. +func (h *Handler) awsExchangeExecutor() func(context.Context, exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + if h.executeExchangeFn != nil { + return h.executeExchangeFn + } + return exchange.ExecuteExchange +} diff --git a/internal/api/ri_exchange_detached_test.go b/internal/api/ri_exchange_detached_test.go new file mode 100644 index 000000000..4a82ce5de --- /dev/null +++ b/internal/api/ri_exchange_detached_test.go @@ -0,0 +1,242 @@ +package api + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/exchange" + azurecompute "github.com/LeanerCloud/cloud-commitments-go/providers/azure/services/compute" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// --- RI exchange execution outlives the request context (#790) --- + +// ctxProbe records what the provider call saw of its context at the moment the +// request was canceled underneath it. +type ctxProbe struct { + errAfterCancel error + deadline time.Time + hasDeadline bool +} + +// observe cancels the request (the client disconnecting, or the server's 30s +// per-request timeout firing, while the provider call is in flight) and then +// records whether that cancellation reached the provider call's context. +func (p *ctxProbe) observe(ctx context.Context, cancelRequest context.CancelFunc) { + cancelRequest() + p.errAfterCancel = ctx.Err() + p.deadline, p.hasDeadline = ctx.Deadline() +} + +func (p *ctxProbe) assertDetached(t *testing.T, start time.Time) { + t.Helper() + require.NoError(t, p.errAfterCancel, + "canceling the request must not cancel an exchange already submitted to the provider") + require.True(t, p.hasDeadline, "the detached exchange must still be bounded by its own timeout") + // A literal rather than riExchangeExecutionTimeout, so the test pins the + // budget itself: it must stay inside the documented lambda_timeout >= 270s + // requirement (4m execution + 30s terminal save). + assert.WithinDuration(t, start.Add(4*time.Minute), p.deadline, 5*time.Second) +} + +// liveCtx matches a context that has not been canceled and carries the 30s +// terminal-save budget, so a mock expectation fails when a terminal write is +// attempted on the dead request context, as a real database driver would. +func liveCtx() any { + return mock.MatchedBy(func(ctx context.Context) bool { + deadline, ok := ctx.Deadline() + return ctx.Err() == nil && ok && time.Until(deadline) <= 30*time.Second + }) +} + +func approvedExchangeStore(t *testing.T, id string) *MockConfigStore { + t.Helper() + store := new(MockConfigStore) + t.Cleanup(func() { store.AssertExpectations(t) }) + store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{ + RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, + }, nil) + store.On("ReserveApprovedRIExchange", mock.Anything, id, "1000.000000", "500.000000").Return("500.000000", nil) + return store +} + +func approvedExchangeRecord(id string) *config.RIExchangeRecord { + return &config.RIExchangeRecord{ + ID: id, Region: "us-east-1", SourceRIIDs: []string{"ri-1"}, TargetOfferingID: "o", TargetCount: 1, PaymentDue: "40.00", + } +} + +// TestExecuteApprovedExchange_RequestCancelDoesNotCancelExchangeOrLedger: AWS +// accepted the exchange, the approver's request was canceled mid-call, and the +// ledger write recording the money movement must still land. +func TestExecuteApprovedExchange_RequestCancelDoesNotCancelExchangeOrLedger(t *testing.T) { + const id = "550e8400-e29b-41d4-a716-000000000790" + reqCtx, cancelRequest := context.WithCancel(context.Background()) + defer cancelRequest() + + store := approvedExchangeStore(t, id) + store.On("CompleteRIExchangeWithPayment", liveCtx(), id, "exch-790", "40.000000").Return(nil).Once() + + var probe ctxProbe + h := &Handler{config: store, executeExchangeFn: func(ctx context.Context, _ exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + probe.observe(ctx, cancelRequest) + return "exch-790", &exchange.ExchangeQuoteSummary{PaymentDueUSDStr: "40.000000"}, nil + }} + + start := time.Now() + resp, err := h.executeApprovedExchange(reqCtx, id, approvedExchangeRecord(id)) + require.NoError(t, err) + probe.assertDetached(t, start) + assert.Equal(t, "completed", resp.(map[string]any)["status"]) +} + +// TestExecuteApprovedExchange_RequestCancelStillRecordsFailure: the failure +// branch's terminal write must also survive the canceled request, or the +// record stays "processing" with its ceiling reserved. +func TestExecuteApprovedExchange_RequestCancelStillRecordsFailure(t *testing.T) { + const id = "550e8400-e29b-41d4-a716-000000000791" + reqCtx, cancelRequest := context.WithCancel(context.Background()) + defer cancelRequest() + + store := approvedExchangeStore(t, id) + store.On("FailRIExchange", liveCtx(), id, "quote no longer valid").Return(nil).Once() + + h := &Handler{config: store, executeExchangeFn: func(_ context.Context, _ exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + cancelRequest() + return "", nil, errors.New("quote no longer valid") + }} + + resp, err := h.executeApprovedExchange(reqCtx, id, approvedExchangeRecord(id)) + require.NoError(t, err) + assert.Equal(t, "failed", resp.(map[string]any)["status"]) +} + +// TestExecuteExchange_RequestCancelDoesNotCancelAWSExchange covers the direct +// AWS execute endpoint the issue names. +func TestExecuteExchange_RequestCancelDoesNotCancelAWSExchange(t *testing.T) { + reqCtx, cancelRequest := context.WithCancel(context.Background()) + defer cancelRequest() + + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", mock.Anything, "tok").Return(&Session{UserID: "user-1"}, nil) + mockAuth.On("HasPermissionAPI", mock.Anything, "user-1", "execute", "ri-exchange").Return(true, nil) + allowAnyAccountScope(mockAuth) + mockAuth.On("HasPermissionForConstraintsAPI", mock.Anything, "user-1", "execute", "ri-exchange", mock.Anything).Return(true, nil) + t.Cleanup(func() { mockAuth.AssertExpectations(t) }) + + var probe ctxProbe + h := &Handler{ + auth: mockAuth, + config: newExchangeClaimLedger(), + reshapeAccountResolver: func(context.Context) (string, error) { return "11111111-2222-3333-4444-555555555555", nil }, + executeExchangeFn: func(ctx context.Context, _ exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + probe.observe(ctx, cancelRequest) + return "exch-direct", &exchange.ExchangeQuoteSummary{PaymentDueUSDStr: "10.000000"}, nil + }, + } + + start := time.Now() + res, err := h.executeExchange(reqCtx, &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"authorization": "Bearer tok"}, + Body: `{"ri_ids":["ri-123"],"target_offering_id":"off-1","target_count":1,"max_payment_due_usd":"250.50","region":"eu-central-1"}`, + }) + require.NoError(t, err) + probe.assertDetached(t, start) + assert.Equal(t, "exch-direct", res.(*ExchangeExecuteResponse).ExchangeID) +} + +// TestExecuteAzureExchange_RequestCancelDoesNotCancelExchange covers the Azure +// execute endpoint, whose ExecuteExchange polls a long-running operation and is +// the most exposed to the 30s request timeout. +func TestExecuteAzureExchange_RequestCancelDoesNotCancelExchange(t *testing.T) { + reqCtx, cancelRequest := context.WithCancel(context.Background()) + defer cancelRequest() + + var probe ctxProbe + opsClient := new(mockAzureExchangeOpsClient) + ownsAzureSource(opsClient) + opsClient.On("CalculateExchange", mock.Anything, mock.Anything, mock.Anything).Return( + &azurecompute.ExchangePreview{SessionID: "sess-790", NetPayable: toPtr(75.00), NetPayableCurrency: "USD"}, + []azurecompute.CompatibleOffering{}, nil, + ) + opsClient.On("ExecuteExchange", mock.Anything, "sess-790").Run(func(args mock.Arguments) { + probe.observe(args.Get(0).(context.Context), cancelRequest) + }).Return(&azurecompute.ExchangeResult{SessionID: "sess-790", Status: "Succeeded"}, nil).Once() + t.Cleanup(func() { opsClient.AssertExpectations(t) }) + + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", mock.Anything, "tok").Return(&Session{UserID: "user-1"}, nil) + mockAuth.On("HasPermissionAPI", mock.Anything, "user-1", "execute", "ri-exchange").Return(true, nil) + allowAnyAccountScope(mockAuth) + mockAuth.On("HasPermissionForConstraintsAPI", mock.Anything, "user-1", "execute", "ri-exchange", mock.Anything).Return(true, nil) + t.Cleanup(func() { mockAuth.AssertExpectations(t) }) + + ledger := newExchangeClaimLedger() + ledger.GetCloudAccountByExternalIDFn = func(context.Context, string, string) (*config.CloudAccount, error) { + return &config.CloudAccount{ID: "acct-1"}, nil + } + h := &Handler{ + auth: mockAuth, + config: ledger, + azureExchangeFactory: func(string) azureExchangeClient { return opsClient }, + } + + start := time.Now() + res, err := h.executeAzureExchange(reqCtx, azureExecuteRequest(validAzureExecuteBody)) + require.NoError(t, err) + probe.assertDetached(t, start) + assert.Equal(t, "Succeeded", res.(*AzureExecuteExchangeResponse).Status) +} + +// TestExecuteApprovedExchange_RequestCancelBeforeReserveRecordsFailure covers +// the window between the pending -> processing transition and the provider +// call. A cancel there fails the reservation; the failure write must still land, +// or the record is stranded in "processing": the approval retry gets 409 from +// the transition guard and the row keeps counting against the daily cap. +func TestExecuteApprovedExchange_RequestCancelBeforeReserveRecordsFailure(t *testing.T) { + const id = "550e8400-e29b-41d4-a716-000000000792" + reqCtx, cancelRequest := context.WithCancel(context.Background()) + defer cancelRequest() + + store := new(MockConfigStore) + t.Cleanup(func() { store.AssertExpectations(t) }) + store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{ + RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, + }, nil) + store.On("ReserveApprovedRIExchange", mock.Anything, id, "1000.000000", "500.000000"). + Run(func(mock.Arguments) { cancelRequest() }). + Return("", context.Canceled).Once() + store.On("FailRIExchange", liveCtx(), id, "daily spending cap check failed").Return(nil).Once() + + called := false + h := &Handler{config: store, executeExchangeFn: func(context.Context, exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + called = true + return "", nil, nil + }} + + resp, err := h.executeApprovedExchange(reqCtx, id, approvedExchangeRecord(id)) + require.NoError(t, err) + assert.Equal(t, "failed", resp.(map[string]any)["status"]) + assert.False(t, called, "a failed reservation must never reach the provider") +} + +// TestStampRIExchangeApprover_SurvivesRequestCancel pins the approved_by audit +// write, which runs after money moved, to a detached context. +func TestStampRIExchangeApprover_SurvivesRequestCancel(t *testing.T) { + const id = "550e8400-e29b-41d4-a716-000000000793" + reqCtx, cancelRequest := context.WithCancel(context.Background()) + cancelRequest() + + store := new(MockConfigStore) + t.Cleanup(func() { store.AssertExpectations(t) }) + store.On("StampRIExchangeApprovedBy", liveCtx(), id, "approver@example.com").Return(nil).Once() + + h := &Handler{config: store} + h.stampRIExchangeApprover(reqCtx, id, &Session{Email: "approver@example.com"}) +} diff --git a/internal/api/ri_exchange_idempotency_test.go b/internal/api/ri_exchange_idempotency_test.go index ad0efd192..210e4d02e 100644 --- a/internal/api/ri_exchange_idempotency_test.go +++ b/internal/api/ri_exchange_idempotency_test.go @@ -103,7 +103,7 @@ func azureDoubleSpendHandler(t *testing.T, ledger *exchangeClaimLedger, committe // Registered for any session and without Once() so that a regression // committing twice fails by ASSERTION on the recorded sessions below, // rather than panicking the test binary on an unexpected mock call. - opsClient.On("ExecuteExchange", ctx, mock.Anything).Run(func(args mock.Arguments) { + opsClient.On("ExecuteExchange", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { *committed = append(*committed, args.String(1)) }).Return(&azurecompute.ExchangeResult{SessionID: "sess-1", Status: "Succeeded"}, nil).Maybe() t.Cleanup(func() { opsClient.AssertExpectations(t) }) diff --git a/internal/api/router_handlers_test.go b/internal/api/router_handlers_test.go index 6e8f81fc4..243ea4539 100644 --- a/internal/api/router_handlers_test.go +++ b/internal/api/router_handlers_test.go @@ -805,7 +805,7 @@ func TestHandler_rejectRIExchange_ValidTokenAndRecord(t *testing.T) { func TestHandler_failExchange(t *testing.T) { ctx := context.Background() mockStore := new(MockConfigStore) - mockStore.On("FailRIExchange", ctx, "11111111-1111-1111-1111-111111111111", "test reason").Return(nil) + mockStore.On("FailRIExchange", mock.Anything, "11111111-1111-1111-1111-111111111111", "test reason").Return(nil) h := &Handler{config: mockStore} result, err := h.failExchange(ctx, "11111111-1111-1111-1111-111111111111", "test reason") @@ -819,7 +819,7 @@ func TestHandler_failExchange_DBError(t *testing.T) { ctx := context.Background() mockStore := new(MockConfigStore) // Even when DB fails, failExchange returns the failure result (logging the error) - mockStore.On("FailRIExchange", ctx, "11111111-1111-1111-1111-111111111111", "reason"). + mockStore.On("FailRIExchange", mock.Anything, "11111111-1111-1111-1111-111111111111", "reason"). Return(errors.New("db down")) h := &Handler{config: mockStore} diff --git a/internal/ctxutil/detached.go b/internal/ctxutil/detached.go new file mode 100644 index 000000000..67daf3662 --- /dev/null +++ b/internal/ctxutil/detached.go @@ -0,0 +1,19 @@ +// Package ctxutil holds small context helpers shared across bounded contexts. +// Keep it minimal: add a helper here only when more than one package needs it. +package ctxutil + +import ( + "context" + "time" +) + +// DetachedWithTimeout returns a context that keeps ctx's values but drops its +// cancellation and deadline, bounded instead by a fresh timeout of d. +// +// Use it for work that must outlive the request or caller that started it, +// such as a provider call that commits money and the terminal write recording +// its outcome: a client disconnect or a per-request timeout must not cancel +// either midway. Always defer the returned CancelFunc. +func DetachedWithTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { + return context.WithTimeout(context.WithoutCancel(ctx), d) +} diff --git a/internal/ctxutil/detached_test.go b/internal/ctxutil/detached_test.go new file mode 100644 index 000000000..ffeaacf9c --- /dev/null +++ b/internal/ctxutil/detached_test.go @@ -0,0 +1,37 @@ +package ctxutil + +import ( + "context" + "testing" + "time" +) + +type ctxKey struct{} + +func TestDetachedWithTimeout(t *testing.T) { + parent, cancelParent := context.WithTimeout(context.WithValue(context.Background(), ctxKey{}, "req-1"), time.Millisecond) + cancelParent() + + start := time.Now() + ctx, cancel := DetachedWithTimeout(parent, time.Minute) + defer cancel() + + if err := ctx.Err(); err != nil { + t.Fatalf("parent cancellation must not reach the detached context, got %v", err) + } + deadline, ok := ctx.Deadline() + if !ok { + t.Fatal("detached context must carry its own deadline") + } + if d := deadline.Sub(start); d < 59*time.Second || d > 61*time.Second { + t.Fatalf("deadline must be ~1m from now (not the parent's), got %s", d) + } + if got := ctx.Value(ctxKey{}); got != "req-1" { + t.Fatalf("detached context must keep parent values, got %v", got) + } + + cancel() + if ctx.Err() == nil { + t.Fatal("the returned CancelFunc must cancel the detached context") + } +} diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index 524a90194..645c2a7aa 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -12,6 +12,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-go/pkg/provider" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" "github.com/LeanerCloud/cloud-commitments-platform/internal/credentials" + "github.com/LeanerCloud/cloud-commitments-platform/internal/ctxutil" "github.com/LeanerCloud/cloud-commitments-platform/internal/email" "github.com/LeanerCloud/cloud-commitments-platform/internal/oidc" "github.com/aws/aws-sdk-go-v2/aws" @@ -233,11 +234,6 @@ const ( terminalSaveTimeout = 30 * time.Second ) -// detachedTimeout keeps ctx values but drops its cancellation and deadline. -func detachedTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { - return context.WithTimeout(context.WithoutCancel(ctx), d) -} - // executeAndFinalize runs a purchase and handles status updates, record saving, and progress. // // The root execution row is ALWAYS saved with its finalized status — including @@ -252,7 +248,7 @@ func detachedTimeout(ctx context.Context, d time.Duration) (context.Context, con func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseExecution) error { // A disconnected caller must not cancel a purchase after the claim. // Its terminal outcome must survive an exhausted execution budget too. - execCtx, cancelExec := detachedTimeout(ctx, purchaseExecutionTimeout) + execCtx, cancelExec := ctxutil.DetachedWithTimeout(ctx, purchaseExecutionTimeout) defer cancelExec() // Last line of defense before money moves (issue #1718). Every executor // entry point funnels through here, so one check covers all of them. @@ -264,7 +260,7 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE if execErr != nil { logging.Errorf("Failed to execute purchase %s: %v", exec.ExecutionID, execErr) } - saveCtx, cancelSave := detachedTimeout(ctx, terminalSaveTimeout) + saveCtx, cancelSave := ctxutil.DetachedWithTimeout(ctx, terminalSaveTimeout) defer cancelSave() if err := m.saveOrRecoverTerminal(saveCtx, exec, execErr); err != nil { logging.Errorf("AUDIT LOSS: failed to save execution status: %v", err)