fix(exchange): send the RI exchange accept once and halt on an unknown outcome

pkg/exchange/auto.go

@@ -2,6 +2,7 @@ package exchange
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math/big"
 	"time"
@@ -551,6 +552,11 @@ func processAutoExchange(ctx context.Context, params RunAutoExchangeParams, rec
 	if execErr != nil {
 		logging.Errorf("exchange execution failed for %s: %v", rec.SourceRIID, execErr)
 		outcome.Error = execErr.Error()
+		if errors.Is(execErr, common.ErrOutcomeUnknown) {
+			// The exchange may have happened: keep the record processing (still
+			// counted against the cap) and stop, instead of marking it failed.
+			return outcome, true
+		}
 		return outcome, failLedgerRecord(ctx, params.Store, record.ID, outcome.Error)
 	}
 

pkg/exchange/exchange.go

@@ -10,6 +10,7 @@ import (
 	"math/big"
 	"strings"
 
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
 	sdkaws "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/ec2"
@@ -446,9 +447,13 @@ func executeWithAPI(ctx context.Context, client EC2ExchangeAPI, identity stsIden
 	out, err := client.AcceptReservedInstancesExchangeQuote(ctx, &ec2.AcceptReservedInstancesExchangeQuoteInput{
 		ReservedInstanceIds:  req.ReservedIDs,
 		TargetConfigurations: req.targetConfigs(),
-	})
+		// No client token: a retried accept after a lost response cannot be deduped.
+	}, func(o *ec2.Options) { o.RetryMaxAttempts = 1 })
 	if err != nil {
-		return "", freshQ, err
+		return "", freshQ, classifyAcceptError(err)
+	}
+	if sdkaws.ToString(out.ExchangeId) == "" {
+		return "", freshQ, fmt.Errorf("%w: accept returned no exchange ID", common.ErrOutcomeUnknown)
 	}
 
 	return sdkaws.ToString(out.ExchangeId), freshQ, nil

pkg/exchange/outcome.go

@@ -0,0 +1,43 @@
+package exchange
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/aws/aws-sdk-go-v2/aws/retry"
+	awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
+	"github.com/aws/smithy-go"
+)
+
+// classifyAcceptError wraps err with common.ErrOutcomeUnknown unless AWS
+// definitely rejected the accept before processing it (an API error with HTTP
+// 4xx or a throttle code). Accept has no client token, so after a lost
+// response the exchange may have happened.
+//
+// Mirrors providers/aws/internal/purchasecfg/outcome.go ClassifyPurchaseError;
+// pkg cannot import it. Keep the two rules in step.
+func classifyAcceptError(err error) error {
+	if err == nil || errors.Is(err, common.ErrOutcomeUnknown) {
+		return err
+	}
+	var apiErr smithy.APIError
+	if errors.As(err, &apiErr) {
+		if _, throttled := retry.DefaultThrottleErrorCodes[apiErr.ErrorCode()]; throttled {
+			return err
+		}
+		if status := acceptHTTPStatus(err); status >= http.StatusBadRequest && status < http.StatusInternalServerError {
+			return err
+		}
+	}
+	return fmt.Errorf("%w: the accept may have been processed and the exchange may exist; reconcile before retrying: %w", common.ErrOutcomeUnknown, err)
+}
+
+func acceptHTTPStatus(err error) int {
+	var re *awshttp.ResponseError
+	if !errors.As(err, &re) || re == nil || re.ResponseError == nil || re.Response == nil || re.Response.Response == nil {
+		return 0
+	}
+	return re.Response.StatusCode
+}

pkg/exchange/outcome_test.go

@@ -0,0 +1,113 @@
+package exchange
+
+import (
+	"context"
+	"errors"
+	"math/big"
+	"net"
+	"net/http"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	sdkaws "github.com/aws/aws-sdk-go-v2/aws"
+	awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
+	"github.com/aws/aws-sdk-go-v2/service/ec2"
+	"github.com/aws/smithy-go"
+	smithyhttp "github.com/aws/smithy-go/transport/http"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func statusErr(status int, code string) error {
+	return &awshttp.ResponseError{ResponseError: &smithyhttp.ResponseError{
+		Response: &smithyhttp.Response{Response: &http.Response{StatusCode: status}},
+		Err:      &smithy.GenericAPIError{Code: code},
+	}}
+}
+
+// acceptRecorder applies the per-call options to a real ec2.Options.
+type acceptRecorder struct {
+	*sequentialFakeEC2
+	attempts int
+}
+
+func (a *acceptRecorder) AcceptReservedInstancesExchangeQuote(ctx context.Context, in *ec2.AcceptReservedInstancesExchangeQuoteInput, optFns ...func(*ec2.Options)) (*ec2.AcceptReservedInstancesExchangeQuoteOutput, error) {
+	var o ec2.Options
+	for _, f := range optFns {
+		f(&o)
+	}
+	a.attempts = o.RetryMaxAttempts
+	return a.sequentialFakeEC2.AcceptReservedInstancesExchangeQuote(ctx, in, optFns...)
+}
+
+func runAccept(t *testing.T, out *ec2.AcceptReservedInstancesExchangeQuoteOutput, acceptErr error) (*acceptRecorder, string, error) {
+	t.Helper()
+	f := &acceptRecorder{sequentialFakeEC2: &sequentialFakeEC2{
+		quoteOutputs: []*ec2.GetReservedInstancesExchangeQuoteOutput{seqQuoteOut("40.00"), seqQuoteOut("40.00")},
+		quoteErrors:  []error{nil, nil},
+		acceptOutput: out,
+		acceptErr:    acceptErr,
+	}}
+	id, _, err := NewExchangeClientFromAPI(f).Execute(context.Background(), ExchangeExecuteRequest{
+		ReservedIDs: []string{"ri-1"}, TargetOfferingID: "off-A", TargetCount: 1,
+		MaxPaymentDueUSD: new(big.Rat).SetInt64(100),
+	})
+	return f, id, err
+}
+
+func TestExecute_AcceptSendsOnceAndClassifies(t *testing.T) {
+	t.Parallel()
+	ok := &ec2.AcceptReservedInstancesExchangeQuoteOutput{ExchangeId: sdkaws.String("exch-1")}
+	lost := &net.OpError{Op: "read", Err: errors.New("connection reset")}
+	for _, tc := range []struct {
+		name    string
+		out     *ec2.AcceptReservedInstancesExchangeQuoteOutput
+		err     error
+		unknown bool
+		fails   bool
+	}{
+		{"success", ok, nil, false, false},
+		{"transport error", nil, lost, true, true},
+		{"5xx", nil, statusErr(500, "InternalError"), true, true},
+		{"400 definite", nil, statusErr(400, "InvalidParameterValue"), false, true},
+		{"throttle definite", nil, statusErr(503, "Throttling"), false, true},
+		{"empty exchange id", &ec2.AcceptReservedInstancesExchangeQuoteOutput{}, nil, true, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			f, _, err := runAccept(t, tc.out, tc.err)
+			assert.Equal(t, 1, f.attempts, "accept has no client token and must be sent once")
+			assert.Equal(t, tc.fails, err != nil)
+			assert.Equal(t, tc.unknown, errors.Is(err, common.ErrOutcomeUnknown))
+		})
+	}
+}
+
+func TestClassifyAcceptError(t *testing.T) {
+	t.Parallel()
+	assert.NoError(t, classifyAcceptError(nil))
+	wrapped := classifyAcceptError(context.DeadlineExceeded)
+	assert.ErrorIs(t, wrapped, common.ErrOutcomeUnknown)
+	assert.ErrorIs(t, wrapped, context.DeadlineExceeded)
+	assert.Equal(t, wrapped, classifyAcceptError(wrapped), "must not double wrap")
+}
+
+func TestRunAutoExchange_UnknownAcceptLeavesRecordProcessingAndHalts(t *testing.T) {
+	t.Parallel()
+	store := &mockExchangeStore{dailySpend: "0"}
+	client := &mockExchangeClient{quoteResult: defaultQuote(), executeErr: classifyAcceptError(context.DeadlineExceeded)}
+	params := defaultParams(store, client)
+	params.Config.Mode = "auto"
+	params.RIs = append(params.RIs, RIInfo{ID: "ri-002", InstanceType: "m5.xlarge", InstanceCount: 1, OfferingClass: "convertible", NormalizationFactor: 8})
+	params.Utilization = append(params.Utilization, UtilizationInfo{RIID: "ri-002", UtilizationPercent: 50.0})
+	params.RIMetadata["ri-002"] = params.RIMetadata["ri-001"]
+
+	result, err := RunAutoExchange(context.Background(), params)
+	require.NoError(t, err)
+	assert.Equal(t, 1, client.executeCalls, "loop must halt after an unknown accept")
+	assert.Zero(t, store.failCalls, "an unknown accept must not release the reservation")
+	require.Len(t, result.Failed, 1)
+	assert.Contains(t, result.Failed[0].Error, "outcome unknown")
+	require.NotEmpty(t, store.savedRecords)
+	assert.NotEqual(t, "failed", store.savedRecords[0].Status)
+}

pkg/go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/config v1.29.12
 	github.com/aws/aws-sdk-go-v2/service/ec2 v1.251.2
 	github.com/aws/aws-sdk-go-v2/service/sts v1.33.17
+	github.com/aws/smithy-go v1.24.2
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.11.1
 	golang.org/x/sync v0.21.0
@@ -27,7 +28,6 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sso v1.25.2 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.30.0 // indirect
-	github.com/aws/smithy-go v1.24.2 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect

providers/aws/internal/purchasecfg/outcome.go

@@ -22,6 +22,8 @@ import (
 // Unknown (wrapped): everything else - transport errors, timeouts, context
 // cancellation, 5xx, a 200 whose body could not be deserialized, and pre-send
 // failures (credentials, DNS, endpoint), which are treated conservatively.
+//
+// Mirrored in pkg/exchange/outcome.go classifyAcceptError; keep the two in step.
 func ClassifyPurchaseError(err error) error {
 	if err == nil || errors.Is(err, common.ErrOutcomeUnknown) {
 		return err