From 91adf0004525351677e68edbd6c6c0f7af01eb71 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 16 Oct 2026 16:15:55 +0200 Subject: [PATCH] fix(exchange): send the RI exchange accept once and halt on an unknown outcome AcceptReservedInstancesExchangeQuote used the default SDK retryer and carries no client token, so a committed exchange whose response was lost came back as a plain error and RunAutoExchange marked its record failed. The accept now makes one attempt, and its error is classified like purchasecfg.ClassifyPurchaseError: 4xx and throttling stay definite, anything else, and a 200 without an ExchangeId, wrap ErrOutcomeUnknown. On an unknown outcome RunAutoExchange leaves the record processing and stops the run instead of failing it. Closes #299 Co-Authored-By: Claude Opus 5.5 --- pkg/exchange/auto.go | 6 + pkg/exchange/exchange.go | 9 +- pkg/exchange/outcome.go | 43 +++++++ pkg/exchange/outcome_test.go | 113 ++++++++++++++++++ pkg/go.mod | 2 +- providers/aws/internal/purchasecfg/outcome.go | 2 + 6 files changed, 172 insertions(+), 3 deletions(-) create mode 100644 pkg/exchange/outcome.go create mode 100644 pkg/exchange/outcome_test.go diff --git a/pkg/exchange/auto.go b/pkg/exchange/auto.go index 1ed28ad8..ea935871 100644 --- a/pkg/exchange/auto.go +++ b/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) } diff --git a/pkg/exchange/exchange.go b/pkg/exchange/exchange.go index df962468..09684636 100644 --- a/pkg/exchange/exchange.go +++ b/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 diff --git a/pkg/exchange/outcome.go b/pkg/exchange/outcome.go new file mode 100644 index 00000000..84cd2e56 --- /dev/null +++ b/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 +} diff --git a/pkg/exchange/outcome_test.go b/pkg/exchange/outcome_test.go new file mode 100644 index 00000000..ba627a29 --- /dev/null +++ b/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) +} diff --git a/pkg/go.mod b/pkg/go.mod index a183ffde..f88dc7c4 100644 --- a/pkg/go.mod +++ b/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 diff --git a/providers/aws/internal/purchasecfg/outcome.go b/providers/aws/internal/purchasecfg/outcome.go index d167f7a8..6cc345b7 100644 --- a/providers/aws/internal/purchasecfg/outcome.go +++ b/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