fix(aws): send empty-token reservation purchases once and report lost responses as unknown

providers/aws/internal/purchasecfg/purchasetest/purchasetest.go

@@ -0,0 +1,79 @@
+// Package purchasetest is imported only by tests. It holds the shared empty-token PurchaseCommitment cases
+// for the services whose purchase API has no idempotency token of its own
+// (see go#299). It is test support only.
+package purchasetest
+
+import (
+	"errors"
+	"net"
+	"net/http"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
+	"github.com/aws/smithy-go"
+	smithyhttp "github.com/aws/smithy-go/transport/http"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Call runs one PurchaseCommitment against a fake whose Purchase API returns
+// apiErr (or an empty success when emptyResp is set, or a normal success when
+// both are zero). It returns the RetryMaxAttempts the call asked for (0 when
+// no per-operation option was passed).
+type Call func(t *testing.T, token string, apiErr error, emptyResp bool) (attempts int, result common.PurchaseResult, err error)
+
+// RespErr wraps inner with an HTTP status; a bare typed fault classifies as unknown.
+func RespErr(status int, inner error) error {
+	return &awshttp.ResponseError{ResponseError: &smithyhttp.ResponseError{
+		Response: &smithyhttp.Response{Response: &http.Response{StatusCode: status}},
+		Err:      inner,
+	}}
+}
+
+// Pick returns the fake output: nil on apiErr, empty if emptyResp, else ok.
+func Pick[T any](ok, empty *T, apiErr error, emptyResp bool) *T {
+	switch {
+	case apiErr != nil:
+		return nil
+	case emptyResp:
+		return empty
+	}
+	return ok
+}
+
+// RunEmptyToken checks the go#299 behavior. alreadyExists is the service's
+// typed AlreadyExists fault.
+func RunEmptyToken(t *testing.T, call Call, alreadyExists error) {
+	t.Helper()
+	attempts, result, err := call(t, "", nil, false)
+	require.NoError(t, err)
+	assert.True(t, result.Success)
+	assert.Equal(t, 1, attempts, "empty token must send once")
+	attempts, _, err = call(t, "exec-token", nil, false)
+	require.NoError(t, err)
+	assert.Equal(t, 0, attempts, "token path must keep the client default retries")
+
+	lost := &net.OpError{Op: "read", Err: errors.New("connection reset")}
+	for _, tc := range []struct {
+		name, token string
+		apiErr      error
+		emptyResp   bool
+		unknown     bool
+	}{
+		{"transport error", "", lost, false, true},
+		{"5xx", "", RespErr(http.StatusInternalServerError, &smithy.GenericAPIError{Code: "InternalFailure"}), false, true},
+		{"empty response", "", nil, true, true},
+		{"400 AlreadyExists", "", RespErr(http.StatusBadRequest, alreadyExists), false, false},
+		{"throttle", "", RespErr(http.StatusServiceUnavailable, &smithy.GenericAPIError{Code: "Throttling"}), false, false},
+		{"token: transport error", "exec-token", lost, false, false},
+		{"token: empty response", "exec-token", nil, true, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			_, result, err := call(t, tc.token, tc.apiErr, tc.emptyResp)
+			require.Error(t, err)
+			assert.False(t, result.Success)
+			assert.Equal(t, tc.unknown, errors.Is(err, common.ErrOutcomeUnknown))
+		})
+	}
+}

providers/aws/internal/purchasecfg/tokenless.go

@@ -0,0 +1,34 @@
+package purchasecfg
+
+import (
+	"fmt"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+)
+
+// These apply only without an idempotency token (go#299; precedent: ec2).
+
+// SingleAttemptIf returns the per-operation option that disables SDK retries
+// when token is empty, and no option otherwise.
+func SingleAttemptIf[O any](token string, set func(*O)) []func(*O) {
+	if token != "" {
+		return nil
+	}
+	return []func(*O){set}
+}
+
+// ClassifyUnlessToken returns err unchanged when token is set, else ClassifyPurchaseError(err).
+func ClassifyUnlessToken(token string, err error) error {
+	if token != "" {
+		return err
+	}
+	return ClassifyPurchaseError(err)
+}
+
+// UnknownUnlessToken wraps err with common.ErrOutcomeUnknown when token is empty.
+func UnknownUnlessToken(token string, err error) error {
+	if token != "" {
+		return err
+	}
+	return fmt.Errorf("%w: %w", common.ErrOutcomeUnknown, err)
+}

providers/aws/internal/purchasecfg/tokenless_test.go

@@ -0,0 +1,23 @@
+package purchasecfg
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/stretchr/testify/assert"
+)
+
+type opts struct{ n int }
+
+func TestTokenlessHelpers(t *testing.T) {
+	set := func(o *opts) { o.n = 1 }
+	assert.Len(t, SingleAttemptIf("", set), 1)
+	assert.Nil(t, SingleAttemptIf("tok", set))
+
+	plain := errors.New("boom")
+	assert.ErrorIs(t, ClassifyUnlessToken("", plain), common.ErrOutcomeUnknown)
+	assert.Equal(t, plain, ClassifyUnlessToken("tok", plain))
+	assert.ErrorIs(t, UnknownUnlessToken("", plain), common.ErrOutcomeUnknown)
+	assert.Equal(t, plain, UnknownUnlessToken("tok", plain))
+}

providers/aws/services/elasticache/client.go

@@ -171,14 +171,18 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati
 		Tags:                         c.createPurchaseTags(rec, opts.Source),
 	}
 
-	response, err := c.client.PurchaseReservedCacheNodesOffering(ctx, input)
+	// With no token the API cannot dedupe, so an SDK retry after a lost response
+	// could re-send the purchase: send once and report a lost response as
+	// unknown (precedent: ec2/client.go).
+	response, err := c.client.PurchaseReservedCacheNodesOffering(ctx, input, purchasecfg.SingleAttemptIf(opts.IdempotencyToken, func(o *elasticache.Options) { o.RetryMaxAttempts = 1 })...)
 	if err != nil {
 		if existingID, recovered := c.recoverAlreadyExists(ctx, opts.IdempotencyToken, reservationID, err); recovered {
 			result.Success = true
 			result.ExistingCommitment = true
 			result.CommitmentID = existingID
 			return result, nil
 		}
+		err = purchasecfg.ClassifyUnlessToken(opts.IdempotencyToken, err)
 		result.Error = fmt.Errorf("failed to purchase Reserved Cache Node: %w", err)
 		return result, result.Error
 	}
@@ -192,7 +196,7 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati
 			result.Cost = &total
 		}
 	} else {
-		result.Error = fmt.Errorf("purchase response was empty")
+		result.Error = purchasecfg.UnknownUnlessToken(opts.IdempotencyToken, fmt.Errorf("purchase response was empty"))
 		return result, result.Error
 	}
 

providers/aws/services/elasticache/purchase_empty_token_test.go

@@ -0,0 +1,58 @@
+package elasticache
+
+import (
+	"context"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/LeanerCloud/cloud-commitments-go/providers/aws/internal/purchasecfg/purchasetest"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/elasticache"
+	"github.com/aws/aws-sdk-go-v2/service/elasticache/types"
+	"github.com/stretchr/testify/mock"
+)
+
+// optRecordingElastiCache records the RetryMaxAttempts the per-operation options set.
+type optRecordingElastiCache struct {
+	*MockElastiCacheClient
+	attempts int
+}
+
+func (r *optRecordingElastiCache) PurchaseReservedCacheNodesOffering(ctx context.Context, p *elasticache.PurchaseReservedCacheNodesOfferingInput, optFns ...func(*elasticache.Options)) (*elasticache.PurchaseReservedCacheNodesOfferingOutput, error) {
+	var o elasticache.Options
+	for _, f := range optFns {
+		f(&o)
+	}
+	r.attempts = o.RetryMaxAttempts
+	return r.MockElastiCacheClient.PurchaseReservedCacheNodesOffering(ctx, p, optFns...)
+}
+
+func TestPurchaseCommitment_EmptyToken(t *testing.T) {
+	call := func(t *testing.T, token string, apiErr error, emptyResp bool) (int, common.PurchaseResult, error) {
+		m := &MockElastiCacheClient{}
+		rec := common.Recommendation{
+			Service: common.ServiceCache, ResourceType: "cache.m5.large", Count: 1,
+			PaymentOption: "all-upfront", Term: "1yr",
+			Details: &common.CacheDetails{Engine: "redis", NodeType: "cache.m5.large"},
+		}
+		m.On("DescribeReservedCacheNodesOfferings", mock.Anything, mock.Anything).
+			Return(&elasticache.DescribeReservedCacheNodesOfferingsOutput{
+				ReservedCacheNodesOfferings: []types.ReservedCacheNodesOffering{{
+					ReservedCacheNodesOfferingId: aws.String("offering-123"),
+					CacheNodeType:                aws.String("cache.m5.large"),
+					ProductDescription:           aws.String("redis"),
+					OfferingType:                 aws.String("All Upfront"),
+					Duration:                     aws.Int32(31536000),
+				}},
+			}, nil)
+		m.On("DescribeReservedCacheNodes", mock.Anything, mock.Anything).
+			Return(&elasticache.DescribeReservedCacheNodesOutput{}, nil)
+		out := purchasetest.Pick(&elasticache.PurchaseReservedCacheNodesOfferingOutput{ReservedCacheNode: &types.ReservedCacheNode{ReservedCacheNodeId: aws.String("rn-1")}},
+			&elasticache.PurchaseReservedCacheNodesOfferingOutput{}, apiErr, emptyResp)
+		m.On("PurchaseReservedCacheNodesOffering", mock.Anything, mock.Anything).Return(out, apiErr)
+		rec2 := &optRecordingElastiCache{MockElastiCacheClient: m}
+		res, err := (&Client{client: rec2, region: "us-east-1"}).PurchaseCommitment(context.Background(), rec, common.PurchaseOptions{IdempotencyToken: token})
+		return rec2.attempts, res, err
+	}
+	purchasetest.RunEmptyToken(t, call, &types.ReservedCacheNodeAlreadyExistsFault{})
+}

providers/aws/services/memorydb/client.go

@@ -166,14 +166,18 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati
 		Tags:                    c.createPurchaseTags(rec, opts.Source),
 	}
 
-	response, err := c.client.PurchaseReservedNodesOffering(ctx, input)
+	// With no token the API cannot dedupe, so an SDK retry after a lost response
+	// could re-send the purchase: send once and report a lost response as
+	// unknown (precedent: ec2/client.go).
+	response, err := c.client.PurchaseReservedNodesOffering(ctx, input, purchasecfg.SingleAttemptIf(opts.IdempotencyToken, func(o *memorydb.Options) { o.RetryMaxAttempts = 1 })...)
 	if err != nil {
 		if existingID, recovered := c.recoverAlreadyExists(ctx, opts.IdempotencyToken, reservationID, err); recovered {
 			result.Success = true
 			result.ExistingCommitment = true
 			result.CommitmentID = existingID
 			return result, nil
 		}
+		err = purchasecfg.ClassifyUnlessToken(opts.IdempotencyToken, err)
 		result.Error = fmt.Errorf("failed to purchase MemoryDB Reserved Nodes: %w", err)
 		return result, result.Error
 	}
@@ -188,7 +192,7 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati
 			result.Cost = &total
 		}
 	} else {
-		result.Error = fmt.Errorf("purchase response was empty")
+		result.Error = purchasecfg.UnknownUnlessToken(opts.IdempotencyToken, fmt.Errorf("purchase response was empty"))
 		return result, result.Error
 	}
 

providers/aws/services/memorydb/purchase_empty_token_test.go

@@ -0,0 +1,57 @@
+package memorydb
+
+import (
+	"context"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/LeanerCloud/cloud-commitments-go/providers/aws/internal/purchasecfg/purchasetest"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/memorydb"
+	"github.com/aws/aws-sdk-go-v2/service/memorydb/types"
+	"github.com/stretchr/testify/mock"
+)
+
+// optRecordingMemoryDB records the RetryMaxAttempts the per-operation options set.
+type optRecordingMemoryDB struct {
+	*MockMemoryDBClient
+	attempts int
+}
+
+func (r *optRecordingMemoryDB) PurchaseReservedNodesOffering(ctx context.Context, p *memorydb.PurchaseReservedNodesOfferingInput, optFns ...func(*memorydb.Options)) (*memorydb.PurchaseReservedNodesOfferingOutput, error) {
+	var o memorydb.Options
+	for _, f := range optFns {
+		f(&o)
+	}
+	r.attempts = o.RetryMaxAttempts
+	return r.MockMemoryDBClient.PurchaseReservedNodesOffering(ctx, p, optFns...)
+}
+
+func TestPurchaseCommitment_EmptyToken(t *testing.T) {
+	call := func(t *testing.T, token string, apiErr error, emptyResp bool) (int, common.PurchaseResult, error) {
+		m := &MockMemoryDBClient{}
+		rec := common.Recommendation{
+			Service: common.ServiceCache, ResourceType: "db.r6gd.xlarge", Count: 1,
+			PaymentOption: "partial-upfront", Term: "1yr",
+			Details: &common.CacheDetails{Engine: "redis", NodeType: "db.r6gd.xlarge"},
+		}
+		m.On("DescribeReservedNodesOfferings", mock.Anything, mock.Anything).
+			Return(&memorydb.DescribeReservedNodesOfferingsOutput{
+				ReservedNodesOfferings: []types.ReservedNodesOffering{{
+					ReservedNodesOfferingId: aws.String("offering-123"),
+					NodeType:                aws.String("db.r6gd.xlarge"),
+					Duration:                31536000,
+					OfferingType:            aws.String("Partial Upfront"),
+				}},
+			}, nil)
+		m.On("DescribeReservedNodes", mock.Anything, mock.Anything).
+			Return(&memorydb.DescribeReservedNodesOutput{}, nil)
+		out := purchasetest.Pick(&memorydb.PurchaseReservedNodesOfferingOutput{ReservedNode: &types.ReservedNode{ReservationId: aws.String("rn-1")}},
+			&memorydb.PurchaseReservedNodesOfferingOutput{}, apiErr, emptyResp)
+		m.On("PurchaseReservedNodesOffering", mock.Anything, mock.Anything).Return(out, apiErr)
+		rec2 := &optRecordingMemoryDB{MockMemoryDBClient: m}
+		res, err := (&Client{client: rec2, region: "us-east-1"}).PurchaseCommitment(context.Background(), rec, common.PurchaseOptions{IdempotencyToken: token})
+		return rec2.attempts, res, err
+	}
+	purchasetest.RunEmptyToken(t, call, &types.ReservedNodeAlreadyExistsFault{})
+}

providers/aws/services/opensearch/client.go

@@ -208,14 +208,18 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati
 		InstanceCount:              aws.Int32(instanceCount),
 	}
 
-	response, err := c.client.PurchaseReservedInstanceOffering(ctx, input)
+	// With no token the API cannot dedupe, so an SDK retry after a lost response
+	// could re-send the purchase: send once and report a lost response as
+	// unknown (precedent: ec2/client.go).
+	response, err := c.client.PurchaseReservedInstanceOffering(ctx, input, purchasecfg.SingleAttemptIf(opts.IdempotencyToken, func(o *opensearch.Options) { o.RetryMaxAttempts = 1 })...)
 	if err != nil {
 		if existingID, recovered := c.recoverAlreadyExists(ctx, opts.IdempotencyToken, reservationName, err); recovered {
 			result.Success = true
 			result.ExistingCommitment = true
 			result.CommitmentID = existingID
 			return result, nil
 		}
+		err = purchasecfg.ClassifyUnlessToken(opts.IdempotencyToken, err)
 		result.Error = fmt.Errorf("failed to purchase OpenSearch RI: %w", err)
 		return result, result.Error
 	}
@@ -224,7 +228,7 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati
 		result.Success = true
 		result.CommitmentID = aws.ToString(response.ReservedInstanceId)
 	} else {
-		result.Error = fmt.Errorf("purchase response was empty")
+		result.Error = purchasecfg.UnknownUnlessToken(opts.IdempotencyToken, fmt.Errorf("purchase response was empty"))
 		return result, result.Error
 	}