diff --git a/internal/api/handler.go b/internal/api/handler.go index 746692a01..58736ed8e 100644 --- a/internal/api/handler.go +++ b/internal/api/handler.go @@ -144,6 +144,9 @@ type Handler struct { // 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) + // executionTimeoutOverride, when positive, replaces riExchangeExecutionTimeout + // for the direct executes. Tests only. + executionTimeoutOverride time.Duration // exchangeQuoteFn is the RI exchange quote function injected by tests. // When nil (the production default), awsExchangeQuoter returns diff --git a/internal/api/handler_ri_exchange.go b/internal/api/handler_ri_exchange.go index 236125fe4..52b526dff 100644 --- a/internal/api/handler_ri_exchange.go +++ b/internal/api/handler_ri_exchange.go @@ -1294,16 +1294,18 @@ func (h *Handler) executeAzureExchange(ctx context.Context, req *events.LambdaFu // mid-LRO and retries commits the exchange twice, each half individually // under the cap. Taken here, last, so no gate rejection ever leaves a // claim behind. + // The execution window is measured from just before the audit save (#1840). + deadline := time.Now().Add(h.exchangeExecutionTimeout()) auditID, err := h.beginDirectExchange(ctx, azureDirectExchangeAudit(session, accountID, body, maxRat, preview), - azureExchangeIdempotencyKey(body)) + azureExchangeIdempotencyKey(body), deadline) if err != nil { return nil, err } // 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) + execCtx, cancelExec := ctxutil.DetachedWithDeadline(ctx, deadline) defer cancelExec() result, err := client.ExecuteExchange(execCtx, preview.SessionID) if err != nil { @@ -2051,16 +2053,20 @@ func (h *Handler) executeExchange(ctx context.Context, req *events.LambdaFunctio // retries otherwise double-spends with each half individually under the // cap. Taken last, after every gate, so no rejected request leaves a // claim behind. + // + // The execution window is measured from just before the audit save (#1840): + // one deadline bounds both the claim and the provider call. + deadline := time.Now().Add(h.exchangeExecutionTimeout()) auditID, err := h.beginDirectExchange(ctx, awsDirectExchangeAudit(session, cloudAccountID, body, maxRat), - awsExchangeIdempotencyKey(cloudAccountID, body)) + awsExchangeIdempotencyKey(cloudAccountID, body), deadline) if err != nil { return nil, err } // Past the claim the exchange may commit at AWS; a cancelled request must // not abort it midway (#790). - execCtx, cancelExec := ctxutil.DetachedWithTimeout(ctx, riExchangeExecutionTimeout) + execCtx, cancelExec := ctxutil.DetachedWithDeadline(ctx, deadline) defer cancelExec() exchangeID, quote, err := h.awsExchangeExecutor()(execCtx, exchange.ExchangeExecuteRequest{ Region: region, diff --git a/internal/api/ri_exchange_audit.go b/internal/api/ri_exchange_audit.go index 763796862..9e30e13db 100644 --- a/internal/api/ri_exchange_audit.go +++ b/internal/api/ri_exchange_audit.go @@ -8,6 +8,7 @@ import ( "math" "math/big" "strconv" + "time" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" "github.com/LeanerCloud/cloud-commitments-go/pkg/exchange" @@ -37,11 +38,19 @@ import ( // the submit-time idempotency claim for claimKey. It returns the audit row's // ID, which the caller finalizes with the provider's outcome. // +// deadline is the end of the whole execution window, computed by the caller +// just before the audit save (#1840): the claim is bounded by it, and the +// caller runs the provider call to the same instant, so the exchange cannot +// outlive created_at + riExchangeExecutionTimeout however slow the claim is +// (the runbook's settle age relies on that bound). A claim that has not +// answered by then fails like any other claim failure and never reaches the +// provider; the row is finalized as failed. +// // A claim refusal (an identical submit inside the window, or a claim store // failure) finalizes the row as failed: the request never reached the // provider, and the row records that an identical execute was attempted and // refused. -func (h *Handler) beginDirectExchange(ctx context.Context, audit *config.RIExchangeDirectExecution, claimKey string) (string, error) { +func (h *Handler) beginDirectExchange(ctx context.Context, audit *config.RIExchangeDirectExecution, claimKey string, deadline time.Time) (string, error) { if err := audit.Validate(); err != nil { return "", fmt.Errorf("invalid RI exchange audit record, exchange not executed: %w", err) } @@ -53,7 +62,14 @@ func (h *Handler) beginDirectExchange(ctx context.Context, audit *config.RIExcha logging.Errorf("ri-exchange direct execute: failed to record audit row, refusing to execute: %v", err) return "", fmt.Errorf("failed to record the RI exchange audit trail, exchange not executed: %w", err) } - if err := h.claimExchangeSubmit(ctx, claimKey); err != nil { + claimCtx, cancelClaim := context.WithDeadline(ctx, deadline) + defer cancelClaim() + err := h.claimExchangeSubmit(claimCtx, claimKey) + if err == nil && claimCtx.Err() != nil { + // Claimed, but too late to leave the provider call any of its window. + err = fmt.Errorf("RI exchange claim finished after the execution deadline: %w", claimCtx.Err()) + } + if err != nil { h.finalizeDirectExchange(ctx, audit.ID, config.RIExchangeDirectOutcome{ Status: config.RIExchangeDirectStatusFailed, Error: "not submitted to the provider: " + err.Error(), diff --git a/internal/api/ri_exchange_audit_test.go b/internal/api/ri_exchange_audit_test.go index a8af0ce18..7a69b8f95 100644 --- a/internal/api/ri_exchange_audit_test.go +++ b/internal/api/ri_exchange_audit_test.go @@ -567,3 +567,199 @@ func TestAWSDirectFailureStatus(t *testing.T) { assert.Equal(t, config.RIExchangeDirectStatusFailed, awsDirectFailureStatus(errors.New("send request failed"))) assert.Equal(t, config.RIExchangeDirectStatusFailed, awsDirectFailureStatus(&smithy.GenericAPIError{Code: "InternalError", Fault: smithy.FaultServer})) } + +// --- #1840: one deadline for the claim and the provider call --- + +// slowClaimStore blocks the submit claim until release closes or the claim's +// own context ends, recording the deadline the claim ran under. +type slowClaimStore struct { + *directExchangeAuditStore + release chan struct{} + claimDeadline time.Time + claimHasDL bool + // lateSuccess makes the claim wait for its context to end and then still + // report success, as a store that answers just after the deadline would. + lateSuccess bool + // saveDelay slows the audit save. + saveDelay time.Duration + // saveStartedAt is taken at the start of the save, before saveDelay, and + // claimReturnedAt when the claim answers; the handler's deadline is + // computed before the first, so tests compare against them, not slack. + saveStartedAt time.Time + claimReturnedAt time.Time +} + +func (s *slowClaimStore) SaveRIExchangeDirectExecution(ctx context.Context, record *config.RIExchangeDirectExecution) error { + s.mu.Lock() + s.saveStartedAt = time.Now() + s.mu.Unlock() + time.Sleep(s.saveDelay) + return s.directExchangeAuditStore.SaveRIExchangeDirectExecution(ctx, record) +} + +func (s *slowClaimStore) ClaimRIExchangeIdempotencyKey(ctx context.Context, key string, window time.Duration) (bool, error) { + s.mu.Lock() + s.claimDeadline, s.claimHasDL = ctx.Deadline() + s.mu.Unlock() + defer func() { + s.mu.Lock() + s.claimReturnedAt = time.Now() + s.mu.Unlock() + }() + if s.lateSuccess { + <-ctx.Done() + return s.directExchangeAuditStore.ClaimRIExchangeIdempotencyKey(context.WithoutCancel(ctx), key, window) + } + select { + case <-s.release: + return s.directExchangeAuditStore.ClaimRIExchangeIdempotencyKey(ctx, key, window) + case <-ctx.Done(): + return false, ctx.Err() + case <-time.After(2 * time.Second): + // An unbounded claim (the regression) must fail the test, not hang it. + return true, nil + } +} + +// A slow claim must not push the provider call past created_at + the budget: +// the exec context ends at the deadline taken before the save, not at +// claim-end + the budget. +func TestExecuteExchange_DirectExecDeadlineAnchoredBeforeSave(t *testing.T) { + const budget = 400 * time.Millisecond + const claimDelay = 150 * time.Millisecond + store := &slowClaimStore{directExchangeAuditStore: newDirectExchangeAuditStore(), release: make(chan struct{})} + var execDeadline time.Time + var execHasDL bool + h := awsAuditHandler(t, store.directExchangeAuditStore, func(ctx context.Context, _ exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + execDeadline, execHasDL = ctx.Deadline() + return "riex-1840", &exchange.ExchangeQuoteSummary{PaymentDueUSDStr: "1.000000", PaymentDueRaw: "1", CurrencyCode: "USD"}, nil + }) + h.config = store + h.executionTimeoutOverride = budget + time.AfterFunc(claimDelay, func() { close(store.release) }) + + _, err := h.executeExchange(auditRequest(context.Background(), awsAuditBody)) + require.NoError(t, err) + + require.True(t, execHasDL) + assert.False(t, execDeadline.After(store.saveStartedAt.Add(budget)), + "exec deadline must be taken no later than the save start") + assert.True(t, execDeadline.Before(store.claimReturnedAt.Add(budget)), + "exec deadline must not be anchored after the %v claim", claimDelay) + require.True(t, store.claimHasDL, "the claim must be bounded") + assert.WithinDuration(t, execDeadline, store.claimDeadline, time.Millisecond, "claim and exec share one deadline") +} + +// A claim that outlives the deadline never reaches the provider, fails as a +// claim failure, and leaves the row finalized as failed. +func TestExecuteExchange_DirectClaimOutlivingDeadlineNeverExecutes(t *testing.T) { + store := &slowClaimStore{directExchangeAuditStore: newDirectExchangeAuditStore(), release: make(chan struct{})} + executed := false + h := awsAuditHandler(t, store.directExchangeAuditStore, func(context.Context, exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + executed = true + return "", nil, nil + }) + h.config = store + h.executionTimeoutOverride = 100 * time.Millisecond + + _, err := h.executeExchange(auditRequest(context.Background(), awsAuditBody)) + require.Error(t, err) + assert.ErrorIs(t, err, context.DeadlineExceeded) + assert.False(t, executed, "a timed-out claim must not reach the executor") + + _, outcome, finalized := store.onlyRow(t) + require.True(t, finalized) + assert.Equal(t, config.RIExchangeDirectStatusFailed, outcome.Status) + assert.Contains(t, outcome.Error, "not submitted to the provider") +} + +// A claim that reports success only after the deadline leaves the provider +// call none of its window: it must not execute, and the row is finalized failed. +func TestExecuteExchange_DirectLateClaimSuccessNeverExecutes(t *testing.T) { + store := &slowClaimStore{directExchangeAuditStore: newDirectExchangeAuditStore(), lateSuccess: true} + executed := false + h := awsAuditHandler(t, store.directExchangeAuditStore, func(context.Context, exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + executed = true + return "", nil, nil + }) + h.config = store + h.executionTimeoutOverride = 100 * time.Millisecond + + _, err := h.executeExchange(auditRequest(context.Background(), awsAuditBody)) + require.Error(t, err) + assert.False(t, executed, "a claim that succeeded after the deadline must not reach the executor") + + _, outcome, finalized := store.onlyRow(t) + require.True(t, finalized) + assert.Equal(t, config.RIExchangeDirectStatusFailed, outcome.Status) +} + +// The deadline is taken before the save: a slow save must not push the exec +// deadline later by the save's duration. +func TestExecuteExchange_DirectExecDeadlineNotMovedBySlowSave(t *testing.T) { + const budget = 600 * time.Millisecond + store := &slowClaimStore{directExchangeAuditStore: newDirectExchangeAuditStore(), release: make(chan struct{}), saveDelay: 250 * time.Millisecond} + close(store.release) + var execDeadline time.Time + h := awsAuditHandler(t, store.directExchangeAuditStore, func(ctx context.Context, _ exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + execDeadline, _ = ctx.Deadline() + return "riex-1840", &exchange.ExchangeQuoteSummary{PaymentDueUSDStr: "1.000000", PaymentDueRaw: "1", CurrencyCode: "USD"}, nil + }) + h.config = store + h.executionTimeoutOverride = budget + + _, err := h.executeExchange(auditRequest(context.Background(), awsAuditBody)) + require.NoError(t, err) + assert.False(t, execDeadline.After(store.saveStartedAt.Add(budget)), + "exec deadline must not move later by the 250ms save") +} + +func azureSlowClaimHandler(t *testing.T, store *slowClaimStore, opsClient *mockAzureExchangeOpsClient, budget time.Duration) *Handler { + h := azureAuditHandler(t, store.directExchangeAuditStore, opsClient) + h.config = store + h.executionTimeoutOverride = budget + return h +} + +// Azure shares beginDirectExchange: its exec must end at the pre-save deadline +// shared with the claim, not claim-end + the budget. +func TestExecuteAzureExchange_DirectExecDeadlineAnchoredBeforeSave(t *testing.T) { + const budget = 400 * time.Millisecond + const claimDelay = 150 * time.Millisecond + store := &slowClaimStore{directExchangeAuditStore: newDirectExchangeAuditStore(), release: make(chan struct{})} + var execDeadline time.Time + var execHasDL bool + opsClient := new(mockAzureExchangeOpsClient) + opsClient.On("ExecuteExchange", mock.Anything, "sess-quoted").Run(func(args mock.Arguments) { + execDeadline, execHasDL = args.Get(0).(context.Context).Deadline() + }).Return(&azurecompute.ExchangeResult{SessionID: "sess-quoted", Status: "Succeeded"}, nil).Once() + h := azureSlowClaimHandler(t, store, opsClient, budget) + time.AfterFunc(claimDelay, func() { close(store.release) }) + + _, err := h.executeAzureExchange(auditRequest(context.Background(), azureAuditBody)) + require.NoError(t, err) + + require.True(t, execHasDL) + assert.False(t, execDeadline.After(store.saveStartedAt.Add(budget)), + "exec deadline must be taken no later than the save start") + assert.True(t, execDeadline.Before(store.claimReturnedAt.Add(budget)), + "exec deadline must not be anchored after the %v claim", claimDelay) + require.True(t, store.claimHasDL, "the claim must be bounded") + assert.WithinDuration(t, execDeadline, store.claimDeadline, time.Millisecond, "claim and exec share one deadline") +} + +// An Azure claim that outlives the deadline never reaches ExecuteExchange. +func TestExecuteAzureExchange_DirectClaimOutlivingDeadlineNeverExecutes(t *testing.T) { + store := &slowClaimStore{directExchangeAuditStore: newDirectExchangeAuditStore(), release: make(chan struct{})} + opsClient := new(mockAzureExchangeOpsClient) // no ExecuteExchange expectation: a call panics + h := azureSlowClaimHandler(t, store, opsClient, 100*time.Millisecond) + + _, err := h.executeAzureExchange(auditRequest(context.Background(), azureAuditBody)) + require.Error(t, err) + assert.ErrorIs(t, err, context.DeadlineExceeded) + opsClient.AssertNotCalled(t, "ExecuteExchange", mock.Anything, mock.Anything) + + _, outcome, finalized := store.onlyRow(t) + require.True(t, finalized) + assert.Equal(t, config.RIExchangeDirectStatusFailed, outcome.Status) +} diff --git a/internal/api/ri_exchange_detached.go b/internal/api/ri_exchange_detached.go index 03c16c3ce..a07e7196f 100644 --- a/internal/api/ri_exchange_detached.go +++ b/internal/api/ri_exchange_detached.go @@ -19,6 +19,11 @@ import ( // the sum of both budgets plus a 30s margin (lambda_timeout >= 300, enforced // by Terraform validation and TestLambdaTimeoutCoversRIExchangeBudget, #913). // +// For the direct executes the execution budget is a deadline, not a fresh +// timeout: it is computed just before the audit row is saved and bounds the +// submit claim and the provider call together (#1840), so the call ends about +// riExchangeExecutionTimeout after the row's created_at. +// // The detaching itself is ctxutil.DetachedWithTimeout, shared with the // purchase path (#706 / #789). const ( @@ -43,3 +48,12 @@ func (h *Handler) awsExchangeQuoter() func(context.Context, exchange.ExchangeQuo } return exchange.GetExchangeQuote } + +// exchangeExecutionTimeout is riExchangeExecutionTimeout unless a test shrank +// it through the handler. +func (h *Handler) exchangeExecutionTimeout() time.Duration { + if h.executionTimeoutOverride > 0 { + return h.executionTimeoutOverride + } + return riExchangeExecutionTimeout +} diff --git a/internal/config/store_postgres_ri_exchange_direct_runbook_integration_test.go b/internal/config/store_postgres_ri_exchange_direct_runbook_integration_test.go index 90831620d..0546f1491 100644 --- a/internal/config/store_postgres_ri_exchange_direct_runbook_integration_test.go +++ b/internal/config/store_postgres_ri_exchange_direct_runbook_integration_test.go @@ -49,12 +49,12 @@ func TestRIExchangeDirectRunbook_SettleStatement(t *testing.T) { require.NoError(t, env.reserve(t, testAccountID(850000000001), "ri-g1829-rb-unknown"), "a settled row stops blocking the guard") old := env.directRow(t, "aws", "submitted", "ri-g1829-rb-old") - age(old, "6 minutes") - assert.EqualValues(t, 1, settle(old), "a submitted row older than 5 minutes is settled") + age(old, "11 minutes") + assert.EqualValues(t, 1, settle(old), "a submitted row older than 10 minutes is settled") young := env.directRow(t, "aws", "submitted", "ri-g1829-rb-young") - age(young, "1 minute") - assert.EqualValues(t, 0, settle(young), "a submitted row that may still be in flight is not settled") + age(young, "6 minutes") + assert.EqualValues(t, 0, settle(young), "a submitted row inside the 10 minute window is not settled") assert.EqualValues(t, 0, settle(env.directRow(t, "azure", "unknown", "ri-g1829-rb-azure")), "Azure rows are untouched") assert.EqualValues(t, 0, settle(env.directRow(t, "aws", "completed", "ri-g1829-rb-done")), "a completed row is final") diff --git a/runbooks/ri-exchange-direct-execution-settle.md b/runbooks/ri-exchange-direct-execution-settle.md index 805923d7e..fb3acb1f2 100644 --- a/runbooks/ri-exchange-direct-execution-settle.md +++ b/runbooks/ri-exchange-direct-execution-settle.md @@ -51,17 +51,18 @@ UPDATE ri_exchange_direct_executions WHERE id = :'id' AND provider = 'aws' AND status IN ('submitted', 'unknown') - AND (status = 'unknown' OR created_at < now() - interval '5 minutes'); + AND (status = 'unknown' OR created_at < now() - interval '10 minutes'); -- Expect exactly 1 row updated. 0 rows means: wrong id, already settled, or a 'submitted' row younger --- than 5 minutes that may still be in flight. Do NOT widen the WHERE clause; wait and re-check. +-- than 10 minutes that may still be in flight. Do NOT widen the WHERE clause; wait and re-check. ``` -- **Why the age check:** the AWS call can run for 4 minutes (`riExchangeExecutionTimeout`), - so a `submitted` row younger than 5 minutes (4 minutes plus 1 minute of margin) may - still be in flight. Marking it `failed` would release the RI while AWS might still +- **Why the age check:** the claim and the AWS call share one deadline of 4 minutes + (`riExchangeExecutionTimeout`) measured from just before the row is saved, so a + `submitted` row younger than 10 minutes (about 4 minutes plus 6 minutes of margin for + clock skew and finalize) may still be in flight. Marking it `failed` would release the RI while AWS might still commit the accept, and the handler's own finalize would then lose the real outcome. - **Why `created_at` can be trusted:** it is taken with `clock_timestamp()` after the guard lock is acquired, so a lock wait does not age the row. A `submitted` row older - than 5 minutes is past the execution window plus the claim. + than 10 minutes is past the execution window, claim included. - **Why `error` is appended:** it keeps the original failure text. - An `unknown` row is settled at any age: its accept already returned.