From 15e3ec89107a8cbc8f69d28a894015995868b255 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 15 Oct 2026 23:10:36 +0100 Subject: [PATCH 1/2] fix(ri-exchange): run ExecuteExchange outside the request context The RI exchange execute paths called the provider on the HTTP request context, so a client disconnect or the container server's 30s request timeout could cancel the exchange after the provider had committed it, and the terminal ledger write recording the outcome then failed on the same dead context. Same class as #706 (fixed for purchases in #789). Run the provider call on a detached context bounded by its own 4m budget, and the terminal writes (CompleteRIExchangeWithPayment, FailRIExchange after execution, approved_by stamp) on a separate 30s detached context. Covers the direct AWS endpoint, the approval flow and the Azure execute endpoint. The direct AWS path now goes through the same injectable executor as the approval flow so it can be tested without AWS. Closes #790 Co-Authored-By: Claude Opus 5.5 --- internal/api/handler.go | 2 +- internal/api/handler_ri_exchange.go | 33 ++- .../api/handler_ri_exchange_azure_test.go | 18 +- internal/api/handler_ri_exchange_test.go | 24 +-- internal/api/middleware_test.go | 6 +- internal/api/ri_exchange_detached.go | 41 ++++ internal/api/ri_exchange_detached_test.go | 189 ++++++++++++++++++ internal/api/ri_exchange_idempotency_test.go | 2 +- 8 files changed, 280 insertions(+), 35 deletions(-) create mode 100644 internal/api/ri_exchange_detached.go create mode 100644 internal/api/ri_exchange_detached_test.go 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..2add78f19 100644 --- a/internal/api/handler_ri_exchange.go +++ b/internal/api/handler_ri_exchange.go @@ -1234,7 +1234,11 @@ 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 := detachedRIExchangeContext(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 @@ -1918,7 +1922,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 := detachedRIExchangeContext(ctx, riExchangeExecutionTimeout) + defer cancelExec() + exchangeID, quote, err := h.awsExchangeExecutor()(execCtx, exchange.ExchangeExecuteRequest{ Region: region, ReservedIDs: body.RIIDs, Targets: toExchangeTargets(body.Targets), @@ -2302,6 +2310,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 := detachedRIExchangeContext(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,6 +2577,8 @@ func (h *Handler) validateExchangeApproval(ctx context.Context, id, token string } // failExchange marks an exchange as failed, logging if the DB write also fails. +// +//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) { 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 +2672,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 := detachedRIExchangeContext(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 := detachedRIExchangeContext(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 +2695,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..50c2ca63a 100644 --- a/internal/api/handler_ri_exchange_test.go +++ b/internal/api/handler_ri_exchange_test.go @@ -403,7 +403,7 @@ func TestApproveRIExchange_SessionAdmin(t *testing.T) { // 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("StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email).Return(nil) req := &events.LambdaFunctionURLRequest{ Headers: map[string]string{ @@ -418,7 +418,7 @@ func TestApproveRIExchange_SessionAdmin(t *testing.T) { 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, "StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email) } // TestApproveRIExchange_SessionApproveOwn verifies that a user with approve-own @@ -460,7 +460,7 @@ func TestApproveRIExchange_SessionApproveOwn(t *testing.T) { RIExchangeMaxPerExchangeUSD: 500, }, nil) mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil) - mockStore.On("StampRIExchangeApprovedBy", ctx, id, ownerSession.Email).Return(nil) + mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, ownerSession.Email).Return(nil) req := &events.LambdaFunctionURLRequest{ Headers: map[string]string{ @@ -537,7 +537,7 @@ func TestApproveRIExchange_SessionScope(t *testing.T) { 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("StampRIExchangeApprovedBy", mock.Anything, id, mock.Anything).Return(nil).Maybe() return &Handler{ config: mockStore, @@ -630,7 +630,7 @@ func fourEyesRIExchangeFixture(t *testing.T, creatorID *string, approverID strin 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("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{ @@ -2333,7 +2333,7 @@ func TestApproveRIExchange_SessionActorStamped(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("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{ @@ -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) { diff --git a/internal/api/middleware_test.go b/internal/api/middleware_test.go index 6b046d4a5..1ee33d39f 100644 --- a/internal/api/middleware_test.go +++ b/internal/api/middleware_test.go @@ -523,7 +523,7 @@ func TestApproveRIExchangeViaSession_RequiresCSRF(t *testing.T) { 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("StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email).Return(nil).Maybe() h := &Handler{config: mockStore, auth: mockAuth} req := &events.LambdaFunctionURLRequest{ @@ -599,7 +599,7 @@ func TestApproveRIExchange_CSRFFailureDoesNotFallThroughToToken(t *testing.T) { 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("StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email).Return(nil).Maybe() h := &Handler{config: mockStore, auth: mockAuth} req := &events.LambdaFunctionURLRequest{ @@ -655,7 +655,7 @@ func TestApproveRIExchangeViaSession_PassesCSRF(t *testing.T) { RIExchangeMaxPerExchangeUSD: 500, }, nil) mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil) - mockStore.On("StampRIExchangeApprovedBy", ctx, id, adminSession.Email).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..a8117e8eb --- /dev/null +++ b/internal/api/ri_exchange_detached.go @@ -0,0 +1,41 @@ +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. +// +// Same pattern as purchase.detachedTimeout (#706 / #789); kept local to the +// exchange path rather than shared so the two money paths can be reconciled +// deliberately. +const ( + riExchangeExecutionTimeout = 4 * time.Minute + riExchangeTerminalSaveTimeout = 30 * time.Second +) + +// detachedRIExchangeContext keeps ctx values but drops its cancellation and +// deadline, replacing them with a fresh timeout of d. +func detachedRIExchangeContext(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { + return context.WithTimeout(context.WithoutCancel(ctx), d) +} + +// 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..6d8fe27ec --- /dev/null +++ b/internal/api/ri_exchange_detached_test.go @@ -0,0 +1,189 @@ +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") + assert.WithinDuration(t, start.Add(riExchangeExecutionTimeout), p.deadline, 5*time.Second) +} + +// liveCtx matches a context that has not been canceled, 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 { return ctx.Err() == nil }) +} + +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) +} 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) }) From d4fd03decbc4fcddb97665df6afc5da8dc53ef95 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 16 Oct 2026 00:16:41 +0100 Subject: [PATCH 2/2] fix(ri-exchange): detach pre-commit failure writes; share detached-context helper Address review on #897: - failExchange now detaches its write. Every caller runs after the pending -> processing transition committed, so a request cancel before ReserveApprovedRIExchange could strand the record in "processing" (409 on retry, still counted against the daily cap). - Pin the approved_by stamp detach with a test. - Assert the 4m execution and 30s terminal-save budgets against literals rather than the constants under test. - Update the stale Azure comment: a request cancel no longer reaches the poll, only the execution budget can. - Introduce ctxutil.DetachedWithTimeout as the single detached-context helper and use it from the purchase and RI exchange paths, replacing purchase.detachedTimeout and the exchange-local copy. Co-Authored-By: Claude Opus 5.5 --- internal/api/handler_ri_exchange.go | 23 ++++++--- internal/api/handler_ri_exchange_test.go | 20 +++---- internal/api/middleware_test.go | 6 +-- internal/api/ri_exchange_detached.go | 11 +--- internal/api/ri_exchange_detached_test.go | 63 +++++++++++++++++++++-- internal/api/router_handlers_test.go | 4 +- internal/ctxutil/detached.go | 19 +++++++ internal/ctxutil/detached_test.go | 37 +++++++++++++ internal/purchase/manager.go | 10 ++-- 9 files changed, 150 insertions(+), 43 deletions(-) create mode 100644 internal/ctxutil/detached.go create mode 100644 internal/ctxutil/detached_test.go diff --git a/internal/api/handler_ri_exchange.go b/internal/api/handler_ri_exchange.go index 2add78f19..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 @@ -1236,14 +1237,15 @@ func (h *Handler) executeAzureExchange(ctx context.Context, req *events.LambdaFu // 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 := detachedRIExchangeContext(ctx, riExchangeExecutionTimeout) + 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( @@ -1924,7 +1926,7 @@ func (h *Handler) executeExchange(ctx context.Context, req *events.LambdaFunctio // Past the claim the exchange may commit at AWS; a cancelled request must // not abort it midway (#790). - execCtx, cancelExec := detachedRIExchangeContext(ctx, riExchangeExecutionTimeout) + execCtx, cancelExec := ctxutil.DetachedWithTimeout(ctx, riExchangeExecutionTimeout) defer cancelExec() exchangeID, quote, err := h.awsExchangeExecutor()(execCtx, exchange.ExchangeExecuteRequest{ Region: region, @@ -2311,7 +2313,7 @@ func (h *Handler) approveRIExchangeViaToken(ctx context.Context, req *events.Lam // 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 := detachedRIExchangeContext(ctx, riExchangeTerminalSaveTimeout) + 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) @@ -2578,8 +2580,15 @@ 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) } @@ -2674,7 +2683,7 @@ func (h *Handler) executeApprovedExchange(ctx context.Context, id string, record // The provider call and the terminal ledger write that records its outcome // must survive a cancelled approval request (#790). - execCtx, cancelExec := detachedRIExchangeContext(ctx, riExchangeExecutionTimeout) + execCtx, cancelExec := ctxutil.DetachedWithTimeout(ctx, riExchangeExecutionTimeout) defer cancelExec() exchangeID, freshQ, execErr := h.awsExchangeExecutor()(execCtx, exchange.ExchangeExecuteRequest{ Region: region, @@ -2683,7 +2692,7 @@ func (h *Handler) executeApprovedExchange(ctx context.Context, id string, record 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 := detachedRIExchangeContext(ctx, riExchangeTerminalSaveTimeout) + saveCtx, cancelSave := ctxutil.DetachedWithTimeout(ctx, riExchangeTerminalSaveTimeout) defer cancelSave() if execErr != nil { return h.failExchange(saveCtx, id, execErr.Error()) diff --git a/internal/api/handler_ri_exchange_test.go b/internal/api/handler_ri_exchange_test.go index 50c2ca63a..4c835b300 100644 --- a/internal/api/handler_ri_exchange_test.go +++ b/internal/api/handler_ri_exchange_test.go @@ -402,7 +402,7 @@ 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("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil) mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email).Return(nil) req := &events.LambdaFunctionURLRequest{ @@ -417,7 +417,7 @@ 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, "FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")) mockStore.AssertCalled(t, "StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email) } @@ -459,7 +459,7 @@ func TestApproveRIExchange_SessionApproveOwn(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) mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, ownerSession.Email).Return(nil) req := &events.LambdaFunctionURLRequest{ @@ -536,7 +536,7 @@ 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("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil).Maybe() mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, mock.Anything).Return(nil).Maybe() return &Handler{ @@ -629,7 +629,7 @@ 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("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 @@ -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,7 +2332,7 @@ 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("FailRIExchange", mock.Anything, id, mock.AnythingOfType("string")).Return(nil) mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email).Return(nil) req := &events.LambdaFunctionURLRequest{ @@ -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) @@ -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 1ee33d39f..11d376095 100644 --- a/internal/api/middleware_test.go +++ b/internal/api/middleware_test.go @@ -522,7 +522,7 @@ 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("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} @@ -598,7 +598,7 @@ 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("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} @@ -654,7 +654,7 @@ func TestApproveRIExchangeViaSession_PassesCSRF(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) mockStore.On("StampRIExchangeApprovedBy", mock.Anything, id, adminSession.Email).Return(nil) h := &Handler{config: mockStore, auth: mockAuth} diff --git a/internal/api/ri_exchange_detached.go b/internal/api/ri_exchange_detached.go index a8117e8eb..7485306e2 100644 --- a/internal/api/ri_exchange_detached.go +++ b/internal/api/ri_exchange_detached.go @@ -17,20 +17,13 @@ import ( // goroutine forever. On Lambda the invocation itself is still bounded by the // function timeout, which no context can extend. // -// Same pattern as purchase.detachedTimeout (#706 / #789); kept local to the -// exchange path rather than shared so the two money paths can be reconciled -// deliberately. +// The detaching itself is ctxutil.DetachedWithTimeout, shared with the +// purchase path (#706 / #789). const ( riExchangeExecutionTimeout = 4 * time.Minute riExchangeTerminalSaveTimeout = 30 * time.Second ) -// detachedRIExchangeContext keeps ctx values but drops its cancellation and -// deadline, replacing them with a fresh timeout of d. -func detachedRIExchangeContext(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { - return context.WithTimeout(context.WithoutCancel(ctx), d) -} - // 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) { diff --git a/internal/api/ri_exchange_detached_test.go b/internal/api/ri_exchange_detached_test.go index 6d8fe27ec..4a82ce5de 100644 --- a/internal/api/ri_exchange_detached_test.go +++ b/internal/api/ri_exchange_detached_test.go @@ -39,14 +39,20 @@ func (p *ctxProbe) assertDetached(t *testing.T, start time.Time) { 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") - assert.WithinDuration(t, start.Add(riExchangeExecutionTimeout), p.deadline, 5*time.Second) + // 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, so a mock expectation -// fails when a terminal write is attempted on the dead request context, as a -// real database driver would. +// 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 { return ctx.Err() == nil }) + 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 { @@ -187,3 +193,50 @@ func TestExecuteAzureExchange_RequestCancelDoesNotCancelExchange(t *testing.T) { 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/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)