fix(ri-exchange): guard direct execute against an RI held by an in-flight exchange

internal/api/handler_ri_exchange.go

@@ -2076,10 +2076,10 @@ func (h *Handler) executeExchange(ctx context.Context, req *events.LambdaFunctio
 		if errors.Is(err, common.ErrOutcomeUnknown) {
 			mapped = directExchangeOutcomeUnknown(body.RIIDs)
 		}
-		h.finalizeDirectExchange(ctx, auditID, awsDirectExchangeOutcome(exchangeID, quote, err, mapped))
+		h.finalizeDirectExchange(ctx, auditID, awsDirectExchangeOutcome(exchangeID, quote, err))
 		return nil, mapped
 	}
-	h.finalizeDirectExchange(ctx, auditID, awsDirectExchangeOutcome(exchangeID, quote, nil, nil))
+	h.finalizeDirectExchange(ctx, auditID, awsDirectExchangeOutcome(exchangeID, quote, nil))
 
 	return &ExchangeExecuteResponse{
 		ExchangeID: exchangeID,
@@ -2120,8 +2120,9 @@ func mapAWSExchangeError(opMsg string, err error) error {
 
 // directExchangeOutcomeUnknown is the error for an accept whose response was
 // lost: AWS may have exchanged the RIs, so the operator must check before
-// retrying (#1050). It stays 5xx so the audit records the outcome as unknown,
-// and it names only the caller's own RI ids, never provider text.
+// retrying (#1050). The audit status is decided from the library's error,
+// which must keep its wrapping (awsDirectFailureStatus), not from this status.
+// It names only the caller's own RI ids, never provider text.
 func directExchangeOutcomeUnknown(riIDs []string) error {
 	return NewClientError(502, fmt.Sprintf("the exchange outcome is unknown: AWS may already have exchanged RI(s) %s; check them at AWS before retrying", strings.Join(riIDs, ", ")))
 }

internal/api/handler_ri_exchange_direct_guard_integration_test.go

@@ -0,0 +1,98 @@
+//go:build integration
+
+package api
+
+import (
+	"context"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/exchange"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const guardAcct = "210987654321"
+
+// #1829 against a real database: the issue's repro and its reverse.
+
+func directGuardHandler(t *testing.T, store *config.PostgresStore, executor awsExecutor) (*Handler, *int) {
+	t.Helper()
+	calls := new(int)
+	return &Handler{
+		auth:                   auditSessionAuth(t),
+		config:                 store,
+		reshapeAccountResolver: func(context.Context) (string, error) { return auditCloudAccount, nil },
+		executeExchangeFn: func(ctx context.Context, req exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) {
+			*calls++
+			return executor(ctx, req)
+		},
+	}, calls
+}
+
+func countRows(ctx context.Context, t *testing.T, c *testhelpers.PostgresContainer, table string) int {
+	t.Helper()
+	var n int
+	require.NoError(t, c.DB.QueryRow(ctx, "SELECT count(*) FROM "+table).Scan(&n))
+	return n
+}
+
+// Repro: a processing history row holds ri-0a1b2c3d, so the direct execute is
+// refused with a 409 and leaves no audit row, no claim and no AWS call.
+func TestExecuteExchange_DirectGuard_Repro(t *testing.T) {
+	ctx := context.Background()
+	container := testhelpers.RequireMigratedDatabase(ctx, t, getMigrationsPath())
+	store := config.NewPostgresStore(container.DB)
+	require.NoError(t, store.SaveRIExchangeRecord(ctx, &config.RIExchangeRecord{
+		AccountID: guardAcct, Region: "eu-central-1", SourceRIIDs: []string{"ri-0a1b2c3d"}, SourceInstanceType: "m5.large",
+		SourceCount: 1, TargetOfferingID: "o", TargetInstanceType: "m6i.large", TargetCount: 1, PaymentDue: "10.00", Status: "processing", Mode: "auto",
+	}))
+	h, calls := directGuardHandler(t, store, func(context.Context, exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) {
+		return "riex-1", nil, nil
+	})
+
+	_, err := h.executeExchange(auditRequest(ctx, awsAuditBody))
+	ce, ok := IsClientError(err)
+	require.True(t, ok, "got %v", err)
+	assert.Equal(t, 409, ce.code)
+	assert.Contains(t, ce.message, "ri-0a1b2c3d")
+	assert.Zero(t, *calls, "AWS must not be called")
+	assert.Zero(t, countRows(ctx, t, container, "ri_exchange_direct_executions"))
+	assert.Zero(t, countRows(ctx, t, container, "ri_exchange_idempotency"))
+}
+
+// Reverse: a direct execute whose accept response is lost leaves an unknown
+// row, which then holds its RIs against AUTO and approval reservations.
+func TestExecuteExchange_DirectGuard_Reverse(t *testing.T) {
+	ctx := context.Background()
+	container := testhelpers.RequireMigratedDatabase(ctx, t, getMigrationsPath())
+	store := config.NewPostgresStore(container.DB)
+	h, calls := directGuardHandler(t, store, func(context.Context, exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) {
+		return "", nil, common.ErrOutcomeUnknown
+	})
+
+	_, err := h.executeExchange(auditRequest(ctx, awsAuditBody))
+	ce, ok := IsClientError(err)
+	require.True(t, ok, "got %v", err)
+	assert.Equal(t, 502, ce.code)
+	assert.Equal(t, 1, *calls)
+	var status string
+	require.NoError(t, container.DB.QueryRow(ctx, "SELECT status FROM ri_exchange_direct_executions").Scan(&status))
+	assert.Equal(t, config.RIExchangeDirectStatusUnknown, status)
+
+	record := func(ri string) *config.RIExchangeRecord {
+		return &config.RIExchangeRecord{AccountID: guardAcct, Region: "eu-central-1", SourceRIIDs: []string{ri}, SourceInstanceType: "m5.large",
+			SourceCount: 1, TargetOfferingID: "o", TargetInstanceType: "m6i.large", TargetCount: 1, PaymentDue: "10.00", Status: "processing", Mode: "auto"}
+	}
+	_, err = store.ReserveRIExchange(ctx, record("ri-0a1b2c3d"), "1000", "1000")
+	require.ErrorIs(t, err, config.ErrRIExchangeSourceInFlight, "AUTO reservation of the held RI")
+	_, err = store.ReserveRIExchange(ctx, record("ri-other"), "1000", "1000")
+	require.NoError(t, err, "a different RI still reserves")
+
+	approving := record("ri-0a1b2c3d")
+	require.NoError(t, store.SaveRIExchangeRecord(ctx, approving))
+	_, err = store.ReserveApprovedRIExchange(ctx, approving.ID, "5000", "1000")
+	require.ErrorIs(t, err, config.ErrRIExchangeSourceInFlight, "approval of the held RI")
+}

internal/api/handler_ri_exchange_direct_guard_test.go

@@ -0,0 +1,50 @@
+package api
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/exchange"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// #1829: a direct execute whose source RI is held by another exchange is
+// refused with a 409 before the submit claim and before any AWS call.
+func TestExecuteExchange_DirectGuard_RefusesWith409BeforeClaimAndAWS(t *testing.T) {
+	store := newDirectExchangeAuditStore()
+	store.saveErr = fmt.Errorf("%w: source RI(s) ri-0a1b2c3d are held by a direct exchange that may have reached AWS; an operator must verify the outcome at AWS and settle the record", config.ErrRIExchangeSourceInFlight)
+	h := awsAuditHandler(t, store, func(context.Context, exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) {
+		t.Fatal("AWS must not be called for a refused direct execute")
+		return "", nil, nil
+	})
+
+	_, err := h.executeExchange(auditRequest(context.Background(), awsAuditBody))
+	ce, ok := IsClientError(err)
+	require.True(t, ok, "got %v", err)
+	assert.Equal(t, 409, ce.code)
+	assert.Contains(t, ce.message, "ri-0a1b2c3d")
+	// The handler passes the store's text through, so the no-blocker-id property
+	// belongs to the store; it is asserted there (config guard integration tests).
+	assert.Equal(t, []string{"save"}, store.recordedSteps(), "no claim, no execute, no finalize")
+	assert.Empty(t, store.claimedKeys())
+}
+
+// Any other save failure still refuses the exchange, but as a server error.
+func TestExecuteExchange_DirectGuard_OtherSaveErrorsStay5xx(t *testing.T) {
+	store := newDirectExchangeAuditStore()
+	store.saveErr = fmt.Errorf("connection refused")
+	h := awsAuditHandler(t, store, func(context.Context, exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) {
+		t.Fatal("AWS must not be called when the audit row cannot be written")
+		return "", nil, nil
+	})
+
+	_, err := h.executeExchange(auditRequest(context.Background(), awsAuditBody))
+	require.Error(t, err)
+	if ce, ok := IsClientError(err); ok {
+		assert.GreaterOrEqual(t, ce.code, 500)
+	}
+	assert.Equal(t, []string{"save"}, store.recordedSteps())
+}

internal/api/openapi.yaml

@@ -2545,6 +2545,11 @@ paths:
         not part of that fingerprint: raising the cap does not make it a
         different purchase. See the 409 response: it does not assert that the
         earlier submit succeeded, only that it claimed the window.
+        The 409 also means a source RI is held by an exchange in progress: an
+        in-flight approve or AUTO exchange row, or an unresolved direct
+        execution (submitted, or unknown after a lost accept response). Nothing
+        is sent to AWS or claimed in that case; verify the outcome at AWS and
+        have an operator settle the held record before retrying.
         The exchange runs in the deployment's own AWS account, so a session
         whose allowed accounts do not cover that account gets 403.
         A quote that is not denominated in USD, or names no currency, is
@@ -2624,7 +2629,14 @@ paths:
               schema:
                 $ref: '#/components/schemas/Error'
         '409':
-          $ref: '#/components/responses/Conflict'
+          description: >-
+            Either an identical submit is already inside the 15 minute dedup
+            window, or a source RI is held by an exchange in progress (an
+            in-flight approve or AUTO row, or an unresolved direct execution).
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'
         '422':
           $ref: '#/components/responses/UnprocessableEntity'
 

internal/api/ri_exchange_audit.go

@@ -3,11 +3,13 @@ package api
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"math"
 	"math/big"
 	"strconv"
 
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/exchange"
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/logging"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
@@ -44,6 +46,10 @@ func (h *Handler) beginDirectExchange(ctx context.Context, audit *config.RIExcha
 		return "", fmt.Errorf("invalid RI exchange audit record, exchange not executed: %w", err)
 	}
 	if err := h.config.SaveRIExchangeDirectExecution(ctx, audit); err != nil {
+		if errors.Is(err, config.ErrRIExchangeSourceInFlight) {
+			logging.Warnf("ri-exchange direct execute refused: %v", err)
+			return "", NewClientError(409, err.Error())
+		}
 		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)
 	}
@@ -72,8 +78,8 @@ func (h *Handler) finalizeDirectExchange(ctx context.Context, id string, outcome
 // directExchangeFailureStatus classifies a failed provider call for the audit
 // trail from the error the handler returns for it: a 4xx means the provider
 // rejected the request as a client fault, so the exchange did not happen;
-// anything else (timeout, 5xx, transport error) does not prove that, so the
-// exchange may have committed and the outcome is recorded as unknown.
+// anything else does not prove that, so the exchange may have committed and
+// the outcome is recorded as unknown. Azure only; AWS uses awsDirectFailureStatus.
 func directExchangeFailureStatus(mapped error) string {
 	if ce, ok := IsClientError(mapped); ok && ce.code >= 400 && ce.code < 500 {
 		return config.RIExchangeDirectStatusFailed
@@ -181,7 +187,7 @@ func awsDirectExchangeAudit(session *Session, cloudAccountID string, body Exchan
 // awsDirectExchangeOutcome builds the outcome of a direct AWS execute.
 // ExecuteExchange may return its quote alongside an error (a guardrail
 // refusal), so the quoted amount is recorded whenever there is one.
-func awsDirectExchangeOutcome(exchangeID string, quote *exchange.ExchangeQuoteSummary, providerErr, mapped error) config.RIExchangeDirectOutcome {
+func awsDirectExchangeOutcome(exchangeID string, quote *exchange.ExchangeQuoteSummary, providerErr error) config.RIExchangeDirectOutcome {
 	outcome := config.RIExchangeDirectOutcome{
 		Status:            config.RIExchangeDirectStatusCompleted,
 		ProviderReference: exchangeID,
@@ -191,12 +197,27 @@ func awsDirectExchangeOutcome(exchangeID string, quote *exchange.ExchangeQuoteSu
 		outcome.NetPayableCurrency = quote.CurrencyCode
 	}
 	if providerErr != nil {
-		outcome.Status = directExchangeFailureStatus(mapped)
+		outcome.Status = awsDirectFailureStatus(providerErr)
 		outcome.Error = providerErr.Error()
 	}
 	return outcome
 }
 
+// awsDirectFailureStatus is unknown only for an error the library marked
+// common.ErrOutcomeUnknown, because only then may the accept have committed.
+// This depends on classifyAcceptError (pkg/exchange/outcome.go:21-35 in the
+// pinned cloud-commitments-go/pkg) wrapping every accept error except a
+// definite 4xx or throttle, and on pre-accept errors (quote, payment over
+// max, invalid exchange, STS) staying unwrapped. The in-flight guard blocks
+// on unknown rows, so a plain error must be failed or it would block the RI.
+// TestAWSDirectFailureStatus pins that wrapping.
+func awsDirectFailureStatus(providerErr error) string {
+	if errors.Is(providerErr, common.ErrOutcomeUnknown) {
+		return config.RIExchangeDirectStatusUnknown
+	}
+	return config.RIExchangeDirectStatusFailed
+}
+
 // azureDirectExchangeAudit builds the audit row for a direct Azure execute
 // from the request and the fresh re-quote that cleared the guardrails.
 func azureDirectExchangeAudit(session *Session, accountID string, body AzureExecuteExchangeRequestBody, maxRat *big.Rat, preview *azureExchangePreview) *config.RIExchangeDirectExecution {

internal/api/ri_exchange_audit_test.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
 	"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"
@@ -234,8 +235,9 @@ func TestExecuteExchange_AuditInsertFailureRefusesExchange(t *testing.T) {
 }
 
 // TestExecuteExchange_FailureOutcomeClassification: a provider client fault
-// proves the exchange did not happen (failed); any other error does not, so
-// the trail must not claim it failed (unknown).
+// proves the exchange did not happen (failed), and so does any error from
+// before the accept; only an error the library marks ErrOutcomeUnknown may
+// have committed (unknown).
 func TestExecuteExchange_FailureOutcomeClassification(t *testing.T) {
 	tests := []struct {
 		name       string
@@ -250,17 +252,37 @@ func TestExecuteExchange_FailureOutcomeClassification(t *testing.T) {
 			wantStatus: config.RIExchangeDirectStatusFailed,
 		},
 		{
-			name:       "transport failure",
-			err:        errors.New("AcceptReservedInstancesExchangeQuote: context deadline exceeded"),
+			name:       "transport failure wrapped as the library does",
+			err:        fmt.Errorf("%w: the accept may have been processed: %w", common.ErrOutcomeUnknown, errors.New("AcceptReservedInstancesExchangeQuote: context deadline exceeded")),
 			wantStatus: config.RIExchangeDirectStatusUnknown,
 		},
+		{
+			name:       "unwrapped pre-accept error did not reach the accept",
+			err:        errors.New("GetReservedInstancesExchangeQuote: context deadline exceeded"),
+			wantStatus: config.RIExchangeDirectStatusFailed,
+		},
 		{
 			name:       "guardrail refusal keeps the quoted amount",
 			err:        errors.New("payment due 900 exceeds max 500"),
 			quote:      &exchange.ExchangeQuoteSummary{PaymentDueRaw: "900.000000", CurrencyCode: "USD"},
-			wantStatus: config.RIExchangeDirectStatusUnknown,
+			wantStatus: config.RIExchangeDirectStatusFailed,
 			wantPay:    toPtr("900"),
 		},
+		{
+			name:       "re-quote over the cap",
+			err:        errors.New("re-quoted payment due 1200 exceeds cap 1000"),
+			wantStatus: config.RIExchangeDirectStatusFailed,
+		},
+		{
+			name:       "invalid exchange quote",
+			err:        errors.New("exchange is not valid: target offering unavailable"),
+			wantStatus: config.RIExchangeDirectStatusFailed,
+		},
+		{
+			name:       "throttled accept is a definite reject",
+			err:        &smithy.GenericAPIError{Code: "RequestLimitExceeded", Message: "throttled", Fault: smithy.FaultServer},
+			wantStatus: config.RIExchangeDirectStatusFailed,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -533,3 +555,15 @@ func TestAWSDirectExchangeAudit_LegacySingletonTarget(t *testing.T) {
 	assert.Equal(t, "250", audit.MaxPaymentDue)
 	require.NoError(t, audit.Validate())
 }
+
+// TestAWSDirectFailureStatus pins the dependency on the library's
+// classifyAcceptError wrapping: a wrapped ErrOutcomeUnknown, however nested,
+// is unknown; a plain error is failed. If the library stops wrapping, the
+// in-flight guard silently stops protecting, and this is where it shows.
+func TestAWSDirectFailureStatus(t *testing.T) {
+	wrapped := fmt.Errorf("%w: the accept may have been processed: %w", common.ErrOutcomeUnknown, errors.New("send request failed"))
+	assert.Equal(t, config.RIExchangeDirectStatusUnknown, awsDirectFailureStatus(wrapped))
+	assert.Equal(t, config.RIExchangeDirectStatusUnknown, awsDirectFailureStatus(fmt.Errorf("outer: %w", wrapped)))
+	assert.Equal(t, config.RIExchangeDirectStatusFailed, awsDirectFailureStatus(errors.New("send request failed")))
+	assert.Equal(t, config.RIExchangeDirectStatusFailed, awsDirectFailureStatus(&smithy.GenericAPIError{Code: "InternalError", Fault: smithy.FaultServer}))
+}

internal/api/ri_exchange_direct_outcome_unknown_test.go

@@ -33,5 +33,5 @@ func TestExecuteExchange_OutcomeUnknownIs502NamingSourceRIs(t *testing.T) {
 
 	_, outcome, finalized := store.onlyRow(t)
 	require.True(t, finalized)
-	assert.Equal(t, config.RIExchangeDirectStatusUnknown, outcome.Status, "a 5xx keeps the audit outcome unknown, never failed")
+	assert.Equal(t, config.RIExchangeDirectStatusUnknown, outcome.Status, "a lost accept response (ErrOutcomeUnknown) keeps the audit outcome unknown, never failed")
 }

internal/api/ri_exchange_refusals.go

@@ -33,9 +33,11 @@ var exchangeRefusals = []error{
 // exchangeRefusalError returns a 422 carrying err's message when err wraps
 // one of exchangeRefusals, or nil otherwise. 422 matches the Azure execute
 // guardrails (checkAzureExchangeMoneyGuardrails), which refuse a currency
-// mismatch the same way, and a 4xx makes directExchangeFailureStatus record
-// the audit row as failed rather than unknown, which is correct because these
-// refusals all happen before the provider's commit call.
+// mismatch the same way. On Azure a 4xx makes directExchangeFailureStatus
+// record the audit row as failed rather than unknown.
+// On AWS the row is failed because the refusal error is not wrapped in
+// common.ErrOutcomeUnknown (awsDirectFailureStatus). Both are correct because
+// these refusals all happen before the provider's commit call.
 func exchangeRefusalError(err error) error {
 	for _, sentinel := range exchangeRefusals {
 		if errors.Is(err, sentinel) {

internal/config/ri_exchange_direct_execution.go

@@ -12,7 +12,10 @@ import (
 // Statuses of an ri_exchange_direct_executions row (issue #82). A row is
 // written as RIExchangeDirectStatusSubmitted before the provider call and
 // moved exactly once to one of the three terminal statuses after it; see
-// migration 000131 for what each one asserts about the money.
+// migration 000131 for what each one asserts about the money. On AWS,
+// unknown means only that the accept response was lost (the library wrapped
+// the error in common.ErrOutcomeUnknown); migration 000131's "timeout, 5xx,
+// transport error" wording is stale for AWS.
 const (
 	RIExchangeDirectStatusSubmitted = "submitted"
 	RIExchangeDirectStatusCompleted = "completed"