fix(ri-exchange): bound the direct-execute claim and provider call by one pre-save deadline

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

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,

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(),

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)
+}

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
+}

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")

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.