diff --git a/providers/aws/go.mod b/providers/aws/go.mod index 147e29a7..4dabcff9 100644 --- a/providers/aws/go.mod +++ b/providers/aws/go.mod @@ -3,7 +3,7 @@ module github.com/LeanerCloud/cloud-commitments-go/providers/aws go 1.26.9 require ( - github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261017020613-36d8b1100c05 + github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261018003623-541e05e5fd29 github.com/aws/aws-sdk-go-v2 v1.41.5 github.com/aws/aws-sdk-go-v2/config v1.29.12 github.com/aws/aws-sdk-go-v2/credentials v1.17.65 diff --git a/providers/aws/go.sum b/providers/aws/go.sum index 78318a17..22694c7a 100644 --- a/providers/aws/go.sum +++ b/providers/aws/go.sum @@ -1,5 +1,5 @@ -github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261017020613-36d8b1100c05 h1:YtSRyJHVUTDnxykCGYpGX7Pt/z+mq4/sCphPmO6Pj5M= -github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261017020613-36d8b1100c05/go.mod h1:SCDEgHG6evjPh0RsoXof+J3aMmGXNFMrwbJIle7rMpQ= +github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261018003623-541e05e5fd29 h1:QJkoZBHkwjvMGplDb1TcuBy3HdYPxlZ+AyR/7QEU8Hc= +github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261018003623-541e05e5fd29/go.mod h1:SCDEgHG6evjPh0RsoXof+J3aMmGXNFMrwbJIle7rMpQ= github.com/aws/aws-sdk-go-v2 v1.41.5 h1:dj5kopbwUsVUVFgO4Fi5BIT3t4WyqIDjGKCangnV/yY= github.com/aws/aws-sdk-go-v2 v1.41.5/go.mod h1:mwsPRE8ceUUpiTgF7QmQIJ7lgsKUPQOUl3o72QBrE1o= github.com/aws/aws-sdk-go-v2/config v1.29.12 h1:Y/2a+jLPrPbHpFkpAAYkVEtJmxORlXoo5k2g1fa2sUo= diff --git a/providers/aws/services/redshift/client.go b/providers/aws/services/redshift/client.go index 8d580255..58adf446 100644 --- a/providers/aws/services/redshift/client.go +++ b/providers/aws/services/redshift/client.go @@ -6,16 +6,13 @@ import ( "fmt" "log" "sort" - "sync" "time" "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/redshift" redshifttypes "github.com/aws/aws-sdk-go-v2/service/redshift/types" - "github.com/aws/aws-sdk-go-v2/service/sts" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" - "github.com/LeanerCloud/cloud-commitments-go/pkg/retry" "github.com/LeanerCloud/cloud-commitments-go/providers/aws/internal/offeringprice" "github.com/LeanerCloud/cloud-commitments-go/providers/aws/internal/purchasecfg" "github.com/LeanerCloud/cloud-commitments-go/providers/aws/internal/reservationexpiry" @@ -27,28 +24,12 @@ type API interface { PurchaseReservedNodeOffering(ctx context.Context, params *redshift.PurchaseReservedNodeOfferingInput, optFns ...func(*redshift.Options)) (*redshift.PurchaseReservedNodeOfferingOutput, error) DescribeReservedNodeOfferings(ctx context.Context, params *redshift.DescribeReservedNodeOfferingsInput, optFns ...func(*redshift.Options)) (*redshift.DescribeReservedNodeOfferingsOutput, error) DescribeReservedNodes(ctx context.Context, params *redshift.DescribeReservedNodesInput, optFns ...func(*redshift.Options)) (*redshift.DescribeReservedNodesOutput, error) - CreateTags(ctx context.Context, params *redshift.CreateTagsInput, optFns ...func(*redshift.Options)) (*redshift.CreateTagsOutput, error) - DescribeTags(ctx context.Context, params *redshift.DescribeTagsInput, optFns ...func(*redshift.Options)) (*redshift.DescribeTagsOutput, error) -} - -// STSAPI is the subset of STS this client calls to resolve the caller's -// account ID for ARN construction. Declared at package level so tests can -// substitute a fake. -type STSAPI interface { - GetCallerIdentity(ctx context.Context, params *sts.GetCallerIdentityInput, optFns ...func(*sts.Options)) (*sts.GetCallerIdentityOutput, error) } // Client handles AWS Redshift Reserved Nodes. type Client struct { - client API - stsClient STSAPI - region string - - // accountID is resolved lazily on first tag call via STS. Guarded by - // accountOnce so concurrent purchases don't each call STS. - accountOnce sync.Once - accountID string - accountErr error + client API + region string } // NewClient creates a new Redshift client with purchase-path retry/timeout @@ -56,9 +37,8 @@ type Client struct { func NewClient(cfg aws.Config) *Client { pcfg := purchasecfg.NewConfig(cfg) return &Client{ - client: redshift.NewFromConfig(pcfg), - stsClient: sts.NewFromConfig(pcfg), - region: cfg.Region, + client: redshift.NewFromConfig(pcfg), + region: cfg.Region, } } @@ -67,11 +47,6 @@ func (c *Client) SetRedshiftAPI(api API) { c.client = api } -// SetSTSAPI sets a custom STS client (for testing). -func (c *Client) SetSTSAPI(api STSAPI) { - c.stsClient = api -} - // GetServiceType returns the service type. func (c *Client) GetServiceType() common.ServiceType { return common.ServiceDataWarehouse @@ -137,12 +112,13 @@ func (c *Client) GetExistingCommitments(ctx context.Context) ([]common.Commitmen // PurchaseCommitment purchases a Redshift Reserved Node. // -// PurchaseReservedNodeOfferingInput has no Tags field — tagging happens -// post-purchase via redshift:CreateTags, which requires a full ARN -// (arn:aws:redshift:::reservednode:). The account ID -// is resolved lazily on first tag call via sts:GetCallerIdentity and cached -// for the lifetime of the client. Tagging failure is logged but does NOT -// fail the purchase — the reserved node is already bought. +// The purchase API takes no client token, so a re-drive is recognized by +// matching an owned node on offering ID, node count and start time (see +// findMatchingNodes). With an empty IdempotencyToken (CLI) no lookup runs. +// The tokenized path relies on the caller contract documented on +// findMatchingNodes: no concurrent same-spec purchase for the account and region. +// +// Reserved nodes cannot be tagged, so the purchased node is not tagged. func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendation, opts common.PurchaseOptions) (common.PurchaseResult, error) { result := common.PurchaseResult{ Recommendation: rec, @@ -157,36 +133,10 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, result.Error } - // Idempotency dedupe guard (issue #641). Redshift's - // PurchaseReservedNodeOfferingInput has NO customer-supplied ID and NO - // ClientToken, the ReservedNode resource has no native AlreadyExists fault, - // and DescribeReservedNodes offers no tag filter — so this is an EC2-style - // tag-guard: before buying, list active reserved nodes and check (via - // DescribeTags on each node's ARN) whether one already carries the - // idempotency token tag; if so, this is a re-drive that already succeeded — - // short-circuit. A lookup error must NOT fall through to a purchase. - // - // CAVEAT (documented residual window): the guard's correctness depends on - // the post-purchase CreateTags below actually persisting on a reserved-node - // ARN, which AWS has not confirmed it supports. If tagging is silently - // unsupported the guard cannot recognize the prior purchase and a re-drive - // could double-buy — the same irreducible "purchase-then-tag-fails" window - // EC2 has, but potentially permanent here. This residual is backstopped by - // the recovery sweep's safe-fail + operator-confirm (issue #635), which is - // why #641 does not by itself unblock Redshift auto-re-drive. if opts.IdempotencyToken != "" { - existingID, found, lookupErr := c.findNodeByIdempotencyToken(ctx, opts.IdempotencyToken) - if lookupErr != nil { - result.Error = fmt.Errorf("idempotency lookup failed before Redshift purchase (refusing to purchase to avoid a possible double-buy): %w", lookupErr) + if done := c.guardAgainstRepurchase(ctx, &result, offeringID, rec.Count, opts); done { return result, result.Error } - if found { - log.Printf("Redshift reserved node for idempotency token %s already exists (%s); skipping purchase (issue #641 re-drive)", common.MaskToken(opts.IdempotencyToken), existingID) - result.Success = true - result.ExistingCommitment = true - result.CommitmentID = existingID - return result, nil - } } input := &redshift.PurchaseReservedNodeOfferingInput{ @@ -199,6 +149,9 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati response, err := c.client.PurchaseReservedNodeOffering(ctx, input, func(o *redshift.Options) { o.RetryMaxAttempts = 1 }) if err != nil { result.Error = fmt.Errorf("failed to purchase Redshift Reserved Node: %w", purchasecfg.ClassifyPurchaseError(err)) + if c.adoptAfterUnknownOutcome(ctx, &result, offeringID, rec.Count, opts) { + return result, nil + } return result, result.Error } @@ -213,203 +166,15 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati } else { // A 200 without a node: the buy most likely happened. result.Error = fmt.Errorf("purchase response was empty: %w", common.ErrOutcomeUnknown) + if c.adoptAfterUnknownOutcome(ctx, &result, offeringID, rec.Count, opts) { + return result, nil + } return result, result.Error } - if err := c.tagReservedNode(ctx, result.CommitmentID, rec, opts.Source, opts.IdempotencyToken); err != nil { - log.Printf("WARNING: failed to tag Redshift reserved node %s after purchase (node is bought; tag missing — idempotency guard degrades for this node, issue #641): %v", result.CommitmentID, err) - } - return result, nil } -// maxReservedNodePages bounds the idempotency reserved-node scan. -// An incomplete scan refuses purchase rather than reporting no match. -const maxReservedNodePages = 10 - -// findNodeByIdempotencyToken looks for a nonterminal Redshift -// reserved node tagged with the given idempotency token (issue #641). Redshift -// has no tag filter on DescribeReservedNodes and no reserved-node tag-search, -// so it lists nonterminal nodes and calls DescribeTags per node ARN to read tags -// client-side. Returns the node ID and true on the first match. Retired/expired -// nodes are excluded (same state filter as GetExistingCommitments). A DescribeTags -// error short-circuits as a lookup failure so the caller fails loud rather than -// risk a double-buy. -func (c *Client) findNodeByIdempotencyToken(ctx context.Context, token string) (outID string, outFound bool, outErr error) { - accountID, err := c.resolveAccountID(ctx) - if err != nil { - return "", false, fmt.Errorf("resolve account ID for idempotency check: %w", err) - } - if accountID == "" { - // Without an account ID we cannot build the ARN DescribeTags needs, so - // the tag-guard cannot run. Fail loud: the caller must not silently buy. - return "", false, fmt.Errorf("account ID unavailable for Redshift idempotency check (no STS client)") - } - - var marker *string - for page := 0; page < maxReservedNodePages; page++ { - if err := ctx.Err(); err != nil { - return "", false, err - } - response, err := c.client.DescribeReservedNodes(ctx, &redshift.DescribeReservedNodesInput{ - Marker: marker, - MaxRecords: aws.Int32(100), - }) - if err != nil { - return "", false, fmt.Errorf("failed to describe reserved nodes for idempotency check: %w", err) - } - if nodeID, found, err := c.scanNodesForToken(ctx, response.ReservedNodes, accountID, token); err != nil || found { - return nodeID, found, err - } - if response.Marker == nil || aws.ToString(response.Marker) == "" { - return "", false, nil - } - marker = response.Marker - } - return "", false, fmt.Errorf("pagination cap reached after %d pages for Redshift reserved-node idempotency lookup", maxReservedNodePages) -} - -// scanNodesForToken checks each nonterminal node for the idempotency -// token tag (issue #641), returning the first match. A DescribeTags error -// short-circuits as a lookup failure. -func (c *Client) scanNodesForToken(ctx context.Context, nodes []redshifttypes.ReservedNode, accountID, token string) (outID string, outFound bool, outErr error) { - for i := range nodes { - node := &nodes[i] - if _, owned := ownedNodeState(node.State); !owned { - continue - } - nodeID := aws.ToString(node.ReservedNodeId) - if nodeID == "" { - continue - } - arn := fmt.Sprintf("arn:aws:redshift:%s:%s:reservednode:%s", c.region, accountID, nodeID) - tagged, err := c.nodeHasIdempotencyTag(ctx, arn, token) - if err != nil { - return "", false, fmt.Errorf("failed to read tags for reserved node %s: %w", nodeID, err) - } - if tagged { - return nodeID, true, nil - } - } - return "", false, nil -} - -// nodeHasIdempotencyTag reports whether the reserved node at the given ARN -// carries the idempotency token tag (issue #641). -func (c *Client) nodeHasIdempotencyTag(ctx context.Context, arn, token string) (bool, error) { - out, err := c.client.DescribeTags(ctx, &redshift.DescribeTagsInput{ - ResourceName: aws.String(arn), - TagKeys: []string{common.IdempotencyTagKey}, - TagValues: []string{token}, - }) - if err != nil { - return false, err - } - for _, tr := range out.TaggedResources { - if tr.Tag != nil && - aws.ToString(tr.Tag.Key) == common.IdempotencyTagKey && - aws.ToString(tr.Tag.Value) == token { - return true, nil - } - } - return false, nil -} - -// resolveAccountID fetches the caller's AWS account ID via STS and caches it. -// Returns ("", nil) — i.e. an empty string with no error — when the STS -// client is nil (e.g. a test client that skipped SetSTSAPI). Callers must -// treat empty as "can't tag" and skip the tag call. -func (c *Client) resolveAccountID(ctx context.Context) (string, error) { - c.accountOnce.Do(func() { - if c.stsClient == nil { - return - } - out, err := c.stsClient.GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{}) - if err != nil { - c.accountErr = err - return - } - if out != nil && out.Account != nil { - c.accountID = *out.Account - } - }) - return c.accountID, c.accountErr -} - -// tagReservedNode constructs the reserved-node ARN and calls redshift:CreateTags. -// Retries up to 4 attempts (1s/2s/4s backoff) on validation errors that can -// indicate the node isn't yet visible to the tagging API. Returns nil when -// there is nothing to tag (no source AND no idempotency token) OR when the -// account ID can't be resolved — both mean "don't tag", logged by the caller. -// -// The idempotency token tag (issue #641) is load-bearing for the pre-purchase -// findNodeByIdempotencyToken guard: if it is not written, a re-drive cannot -// recognize this node as already-purchased. -func (c *Client) tagReservedNode(ctx context.Context, nodeID string, rec common.Recommendation, source, idempotencyToken string) error { - if source == "" && idempotencyToken == "" { - return nil - } - accountID, err := c.resolveAccountID(ctx) - if err != nil { - return fmt.Errorf("resolve account ID: %w", err) - } - if accountID == "" { - return fmt.Errorf("account ID unavailable (no STS client)") - } - - arn := fmt.Sprintf("arn:aws:redshift:%s:%s:reservednode:%s", c.region, accountID, nodeID) - // Rich self-describing tag set (issue #687): the Redshift purchase API - // does not accept a customer-supplied node ID, so the only way to make - // the reserved node identifiable from the AWS console alone (without - // cross-referencing CUDly) is to encode the same descriptors that the - // other AWS service clients embed in their reservation name. The Name tag - // uses BuildReservationName to produce the same rich format so operators - // can identify the node without cross-referencing CUDly's purchase audit log. - displayName := common.BuildReservationName(common.ReservationNameFields{ - Service: "redshift", - Region: rec.Region, - ResourceType: rec.ResourceType, - Count: rec.Count, - Term: rec.Term, - Payment: rec.PaymentOption, - Now: time.Now(), - }, "redshift-reserved-") - tags := []redshifttypes.Tag{ - {Key: aws.String("Name"), Value: aws.String(displayName)}, - {Key: aws.String("Purpose"), Value: aws.String("Reserved Node Purchase")}, - {Key: aws.String("NodeType"), Value: aws.String(rec.ResourceType)}, - {Key: aws.String("Region"), Value: aws.String(rec.Region)}, - {Key: aws.String("Count"), Value: aws.String(fmt.Sprintf("%d", rec.Count))}, - {Key: aws.String("Term"), Value: aws.String(rec.Term)}, - {Key: aws.String("PaymentOption"), Value: aws.String(rec.PaymentOption)}, - {Key: aws.String("PurchaseDate"), Value: aws.String(time.Now().Format("2006-01-02"))}, - {Key: aws.String("Tool"), Value: aws.String("CUDly")}, - } - if source != "" { - tags = append(tags, redshifttypes.Tag{Key: aws.String(common.PurchaseTagKey), Value: aws.String(source)}) - } - if idempotencyToken != "" { - tags = append(tags, redshifttypes.Tag{Key: aws.String(common.IdempotencyTagKey), Value: aws.String(idempotencyToken)}) - } - - cfg := retry.Config{MaxAttempts: 4, BaseDelay: time.Second, MaxDelay: 4 * time.Second} - return retry.Do(ctx, cfg, func(perAttemptCtx context.Context, _ int) error { - _, err := c.client.CreateTags(perAttemptCtx, &redshift.CreateTagsInput{ - ResourceName: aws.String(arn), - Tags: tags, - }) - if err == nil { - return nil - } - // All Redshift CreateTags errors that aren't transient (e.g. auth, - // invalid ARN) are permanent. No documented "resource not yet - // visible" retry condition for reserved nodes, so a single pass - // is usually enough; the retry budget exists to absorb network - // flakes not state-machine delays. - return fmt.Errorf("%w: %w", retry.ErrPermanent, err) - }) -} - // maxOfferingPages is the maximum number of DescribeReservedNodeOfferings // pages to walk before giving up. At MaxRecords=100 per page this caps the // search at 500 offerings. Exceeding the cap returns a diagnostic error instead diff --git a/providers/aws/services/redshift/client_test.go b/providers/aws/services/redshift/client_test.go index 8824a0e0..e8e44c8b 100644 --- a/providers/aws/services/redshift/client_test.go +++ b/providers/aws/services/redshift/client_test.go @@ -3,7 +3,6 @@ package redshift import ( "context" "fmt" - "strings" "testing" "time" @@ -12,7 +11,6 @@ import ( "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/redshift" "github.com/aws/aws-sdk-go-v2/service/redshift/types" - "github.com/aws/aws-sdk-go-v2/service/sts" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -47,35 +45,6 @@ func (m *MockRedshiftClient) DescribeReservedNodes(ctx context.Context, params * return args.Get(0).(*redshift.DescribeReservedNodesOutput), args.Error(1) } -func (m *MockRedshiftClient) CreateTags(ctx context.Context, params *redshift.CreateTagsInput, optFns ...func(*redshift.Options)) (*redshift.CreateTagsOutput, error) { - args := m.Called(ctx, params) - if args.Get(0) == nil { - return nil, args.Error(1) - } - return args.Get(0).(*redshift.CreateTagsOutput), args.Error(1) -} - -func (m *MockRedshiftClient) DescribeTags(ctx context.Context, params *redshift.DescribeTagsInput, optFns ...func(*redshift.Options)) (*redshift.DescribeTagsOutput, error) { - args := m.Called(ctx, params) - if args.Get(0) == nil { - return nil, args.Error(1) - } - return args.Get(0).(*redshift.DescribeTagsOutput), args.Error(1) -} - -// MockRedshiftSTSClient implements STSAPI for testing. -type MockRedshiftSTSClient struct { - mock.Mock -} - -func (m *MockRedshiftSTSClient) GetCallerIdentity(ctx context.Context, params *sts.GetCallerIdentityInput, optFns ...func(*sts.Options)) (*sts.GetCallerIdentityOutput, error) { - args := m.Called(ctx, params) - if args.Get(0) == nil { - return nil, args.Error(1) - } - return args.Get(0).(*sts.GetCallerIdentityOutput), args.Error(1) -} - func TestClient_ReservationStatesPreventDuplicatePurchases(t *testing.T) { for _, tc := range []struct { name string @@ -100,7 +69,7 @@ func TestClient_ReservationStatesPreventDuplicatePurchases(t *testing.T) { token := common.DeriveIdempotencyToken("state-regression", 0) id := common.IdempotentReservationID("redshift-id-", token) response := &redshift.DescribeReservedNodesOutput{ReservedNodes: []types.ReservedNode{{ - ReservedNodeId: aws.String(id), State: tc.state, StartTime: aws.Time(time.Now().Add(-time.Hour)), + ReservedNodeId: aws.String(id), ReservedNodeOfferingId: aws.String("offering-1"), State: tc.state, StartTime: aws.Time(time.Now().Add(-time.Hour)), NodeType: aws.String(rec.ResourceType), NodeCount: aws.Int32(1), Duration: aws.Int32(31536000), }}} t.Run("dedupe", func(t *testing.T) { @@ -149,20 +118,13 @@ func TestClient_ReservationStatesPreventDuplicatePurchases(t *testing.T) { m := &MockRedshiftClient{} t.Cleanup(func() { m.AssertExpectations(t) }) expectRSOffering(m) - response.ReservedNodes = append([]types.ReservedNode{{ReservedNodeId: aws.String("unrelated"), State: tc.state}}, response.ReservedNodes...) + response.ReservedNodes[0].StartTime = aws.Time(time.Now()) + response.ReservedNodes = append([]types.ReservedNode{{ReservedNodeId: aws.String("unrelated"), ReservedNodeOfferingId: aws.String("other-offering"), NodeCount: aws.Int32(1), State: tc.state}}, response.ReservedNodes...) m.On("DescribeReservedNodes", mock.Anything, mock.Anything).Return(response, nil) sentinel := fmt.Errorf("purchase boundary reached") m.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything).Return((*redshift.PurchaseReservedNodeOfferingOutput)(nil), sentinel).Maybe() - m.On("DescribeTags", mock.Anything, mock.MatchedBy(func(in *redshift.DescribeTagsInput) bool { - return strings.HasSuffix(aws.ToString(in.ResourceName), ":reservednode:unrelated") - })).Return(&redshift.DescribeTagsOutput{}, nil).Maybe() - m.On("DescribeTags", mock.Anything, mock.MatchedBy(func(in *redshift.DescribeTagsInput) bool { - return strings.HasSuffix(aws.ToString(in.ResourceName), ":reservednode:"+id) - })).Return(&redshift.DescribeTagsOutput{TaggedResources: []types.TaggedResource{{ - Tag: &types.Tag{Key: aws.String(common.IdempotencyTagKey), Value: aws.String(token)}, - }}}, nil).Maybe() - client := rsClientWithAccount(m) - result, err := client.PurchaseCommitment(context.Background(), rec, common.PurchaseOptions{IdempotencyToken: token}) + client := &Client{client: m, region: rec.Region} + result, err := client.PurchaseCommitment(context.Background(), rec, common.PurchaseOptions{IdempotencyToken: token, RequestedAt: time.Now().Add(-time.Hour)}) if tc.owned { assert.NoError(t, err) assert.True(t, result.Success) @@ -172,7 +134,6 @@ func TestClient_ReservationStatesPreventDuplicatePurchases(t *testing.T) { assert.ErrorIs(t, err, sentinel) assert.False(t, result.Success) m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) - m.AssertNotCalled(t, "DescribeTags", mock.Anything, mock.Anything) } }) }) @@ -434,136 +395,6 @@ func TestClient_PurchaseCommitment(t *testing.T) { mockRS.AssertExpectations(t) } -func TestClient_PurchaseCommitment_TagsWithResolvedARN(t *testing.T) { - mockRS := &MockRedshiftClient{} - mockSTS := &MockRedshiftSTSClient{} - client := &Client{ - client: mockRS, - stsClient: mockSTS, - region: "eu-west-1", - } - - rec := common.Recommendation{ - Service: common.ServiceDataWarehouse, - ResourceType: "ra3.xlplus", - Count: 1, - PaymentOption: "all-upfront", - Term: "1yr", - Region: "eu-west-1", - Details: common.DataWarehouseDetails{NodeType: "ra3.xlplus", NumberOfNodes: 1}, - } - - mockRS.On("DescribeReservedNodeOfferings", mock.Anything, mock.Anything). - Return(&redshift.DescribeReservedNodeOfferingsOutput{ - ReservedNodeOfferings: []types.ReservedNodeOffering{{ - ReservedNodeOfferingId: aws.String("off-tag"), - NodeType: aws.String("ra3.xlplus"), - Duration: aws.Int32(31536000), - ReservedNodeOfferingType: types.ReservedNodeOfferingType("Regular"), - FixedPrice: aws.Float64(100.0), - }}, - }, nil) - - mockRS.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything). - Return(&redshift.PurchaseReservedNodeOfferingOutput{ - ReservedNode: &types.ReservedNode{ReservedNodeId: aws.String("rn-tag"), FixedPrice: aws.Float64(100.0)}, - }, nil) - - mockSTS.On("GetCallerIdentity", mock.Anything, mock.Anything). - Return(&sts.GetCallerIdentityOutput{Account: aws.String("123456789012")}, nil) - - expectedARN := "arn:aws:redshift:eu-west-1:123456789012:reservednode:rn-tag" - mockRS.On("CreateTags", mock.Anything, mock.MatchedBy(func(in *redshift.CreateTagsInput) bool { - if aws.ToString(in.ResourceName) != expectedARN { - return false - } - for _, tag := range in.Tags { - if aws.ToString(tag.Key) == common.PurchaseTagKey && aws.ToString(tag.Value) == common.PurchaseSourceCLI { - return true - } - } - return false - })).Return(&redshift.CreateTagsOutput{}, nil) - - result, err := client.PurchaseCommitment(context.Background(), rec, common.PurchaseOptions{Source: common.PurchaseSourceCLI}) - assert.NoError(t, err) - assert.True(t, result.Success) - mockRS.AssertExpectations(t) - mockSTS.AssertExpectations(t) -} - -// TestClient_PurchaseCommitment_TagsCarryRichDescriptors asserts the rich -// self-describing tag set (issue #687): the Redshift purchase API does not -// accept a customer-supplied node ID, so the only way to make the reserved -// node identifiable from the AWS console alone (without cross-referencing -// CUDly's purchase audit log) is to encode service / region / SKU / count / -// term / payment as tags. -func TestClient_PurchaseCommitment_TagsCarryRichDescriptors(t *testing.T) { - mockRS := &MockRedshiftClient{} - mockSTS := &MockRedshiftSTSClient{} - client := &Client{ - client: mockRS, - stsClient: mockSTS, - region: "us-east-1", - } - - rec := common.Recommendation{ - Service: common.ServiceDataWarehouse, - ResourceType: "ra3.xlplus", - Count: 2, - PaymentOption: "partial-upfront", - Term: "3yr", - Region: "us-east-1", - Details: common.DataWarehouseDetails{NodeType: "ra3.xlplus", NumberOfNodes: 2}, - } - - mockRS.On("DescribeReservedNodeOfferings", mock.Anything, mock.Anything). - Return(&redshift.DescribeReservedNodeOfferingsOutput{ - ReservedNodeOfferings: []types.ReservedNodeOffering{{ - ReservedNodeOfferingId: aws.String("off-rich"), - NodeType: aws.String("ra3.xlplus"), - Duration: aws.Int32(94608000), - ReservedNodeOfferingType: types.ReservedNodeOfferingType("Regular"), - // partial-upfront price shape: upfront + recurring (08-H2). - FixedPrice: aws.Float64(500.0), - RecurringCharges: []types.RecurringCharge{{ - RecurringChargeAmount: aws.Float64(0.10), - RecurringChargeFrequency: aws.String("Hourly"), - }}, - }}, - }, nil) - - mockRS.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything). - Return(&redshift.PurchaseReservedNodeOfferingOutput{ - ReservedNode: &types.ReservedNode{ReservedNodeId: aws.String("rn-rich"), FixedPrice: aws.Float64(500.0)}, - }, nil) - - mockSTS.On("GetCallerIdentity", mock.Anything, mock.Anything). - Return(&sts.GetCallerIdentityOutput{Account: aws.String("123456789012")}, nil) - - mockRS.On("CreateTags", mock.Anything, mock.MatchedBy(func(in *redshift.CreateTagsInput) bool { - got := map[string]string{} - for _, tag := range in.Tags { - got[aws.ToString(tag.Key)] = aws.ToString(tag.Value) - } - // Required new descriptors from #687 (the existing NodeType / Region / - // PurchaseDate / Tool / Purpose tags are covered by the existing - // TagsWithResolvedARN test). Name tag carries the rich self-describing - // identifier so the node is findable in the AWS console without CUDly. - name := got["Name"] - return got["Count"] == "2" && - got["Term"] == "3yr" && - got["PaymentOption"] == "partial-upfront" && - len(name) > 0 && - name[:8] == "redshift" - })).Return(&redshift.CreateTagsOutput{}, nil) - - result, err := client.PurchaseCommitment(context.Background(), rec, common.PurchaseOptions{Source: common.PurchaseSourceCLI}) - assert.NoError(t, err) - assert.True(t, result.Success) - mockRS.AssertExpectations(t) -} - func TestClient_MatchesDuration(t *testing.T) { client := &Client{} @@ -1156,113 +987,6 @@ func TestClient_FindOfferingID_UnknownOfferingType(t *testing.T) { mockRS.AssertExpectations(t) } -func TestClient_PurchaseCommitment_Idempotent_ScanBoundaries(t *testing.T) { - for _, tc := range []struct { - name string - pages int - match bool - terminal bool - cancel bool - wantCalls int - wantBuy int - wantSuccess bool - }{ - {name: "cap refuses purchase with page eleven available", pages: 11, terminal: true, wantCalls: 10}, - {name: "cancellation between pages", pages: 2, terminal: true, cancel: true, wantCalls: 1}, - {name: "match on page ten with continuation", pages: 10, match: true, wantCalls: 10, wantSuccess: true}, - {name: "terminal page ten permits purchase", pages: 10, terminal: true, wantCalls: 10, wantBuy: 1, wantSuccess: true}, - } { - t.Run(tc.name, func(t *testing.T) { - m := &MockRedshiftClient{} - client := rsClientWithAccount(m) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - token := common.DeriveIdempotencyToken("scan-boundary", 0) - expectRSOffering(m) - for page := 1; page <= tc.pages; page++ { - input := &redshift.DescribeReservedNodesInput{MaxRecords: aws.Int32(100)} - if page > 1 { - input.Marker = aws.String(fmt.Sprintf("page-%d", page)) - } - out := &redshift.DescribeReservedNodesOutput{ - Marker: aws.String(fmt.Sprintf("page-%d", page+1)), - ReservedNodes: []types.ReservedNode{{ - ReservedNodeId: aws.String(fmt.Sprintf("node-%d", page)), State: aws.String("active"), - }}, - } - if page == tc.pages && tc.terminal { - out.Marker = nil - } - call := m.On("DescribeReservedNodes", mock.Anything, input).Return(out, nil) - tagOut := &redshift.DescribeTagsOutput{} - if page == tc.pages && tc.match { - tagOut.TaggedResources = []types.TaggedResource{{ - Tag: &types.Tag{Key: aws.String(common.IdempotencyTagKey), Value: aws.String(token)}, - }} - } - tagCall := m.On("DescribeTags", mock.Anything, &redshift.DescribeTagsInput{ - ResourceName: aws.String(fmt.Sprintf("arn:aws:redshift:eu-west-1:123456789012:reservednode:node-%d", page)), - TagKeys: []string{common.IdempotencyTagKey}, TagValues: []string{token}, - }).Return(tagOut, nil) - if page > tc.wantCalls { - call.Maybe() - tagCall.Maybe() - } else { - call.Once() - tagCall.Once() - } - if tc.cancel && page == 1 { - call.Run(func(mock.Arguments) { cancel() }) - } - } - purchaseErr := fmt.Errorf("unexpected purchase sentinel") - if tc.wantBuy == 1 { - m.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything). - Return(&redshift.PurchaseReservedNodeOfferingOutput{ - ReservedNode: &types.ReservedNode{ReservedNodeId: aws.String("new"), State: aws.String("payment-pending")}, - }, nil).Once() - m.On("CreateTags", mock.Anything, mock.Anything).Return(&redshift.CreateTagsOutput{}, nil).Once() - } else { - m.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything). - Return((*redshift.PurchaseReservedNodeOfferingOutput)(nil), purchaseErr).Maybe() - } - result, err := client.PurchaseCommitment(ctx, rsIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) - if tc.wantSuccess { - assert.NoError(t, err) - if tc.match { - assert.Equal(t, "node-10", result.CommitmentID) - } else { - assert.Equal(t, "new", result.CommitmentID) - } - } else if tc.cancel { - assert.ErrorIs(t, err, context.Canceled) - } else if assert.Error(t, err) { - assert.Contains(t, err.Error(), "pagination cap reached after 10 pages") - assert.Contains(t, err.Error(), "refusing to purchase") - } - assert.Equal(t, tc.wantSuccess, result.Success) - assert.Equal(t, tc.match, result.ExistingCommitment) - m.AssertNumberOfCalls(t, "DescribeReservedNodes", tc.wantCalls) - m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", tc.wantBuy) - m.AssertExpectations(t) - }) - } -} - -func TestFindNodeByIdempotencyToken_CanceledBeforeScan(t *testing.T) { - m := &MockRedshiftClient{} - client := rsClientWithAccount(m) - ctx, cancel := context.WithCancel(context.Background()) - cancel() - m.On("DescribeReservedNodes", mock.Anything, mock.Anything). - Return(&redshift.DescribeReservedNodesOutput{}, nil).Maybe() - id, found, err := client.findNodeByIdempotencyToken(ctx, "token") - assert.ErrorIs(t, err, context.Canceled) - assert.Empty(t, id) - assert.False(t, found) - m.AssertNotCalled(t, "DescribeReservedNodes", mock.Anything, mock.Anything) -} - func rsIdemRec() common.Recommendation { return common.Recommendation{ Service: common.ServiceDataWarehouse, @@ -1291,96 +1015,6 @@ func expectRSOffering(m *MockRedshiftClient) { }, nil) } -func rsClientWithAccount(mockRS *MockRedshiftClient) *Client { - stsMock := &MockRedshiftSTSClient{} - stsMock.On("GetCallerIdentity", mock.Anything, mock.Anything). - Return(&sts.GetCallerIdentityOutput{Account: aws.String("123456789012")}, nil) - return &Client{client: mockRS, stsClient: stsMock, region: "eu-west-1"} -} - -// TestClient_PurchaseCommitment_Idempotent_TagGuardShortCircuits asserts the -// EC2-style tag-guard short-circuits when a reserved node already carries the -// idempotency token tag (issue #641). -func TestClient_PurchaseCommitment_Idempotent_TagGuardShortCircuits(t *testing.T) { - mockRS := &MockRedshiftClient{} - client := rsClientWithAccount(mockRS) - token := common.DeriveIdempotencyToken("exec-1", 0) - - expectRSOffering(mockRS) - mockRS.On("DescribeReservedNodes", mock.Anything, mock.Anything). - Return(&redshift.DescribeReservedNodesOutput{ - ReservedNodes: []types.ReservedNode{{ReservedNodeId: aws.String("rn-existing"), State: aws.String("active")}}, - }, nil) - mockRS.On("DescribeTags", mock.Anything, mock.Anything). - Return(&redshift.DescribeTagsOutput{ - TaggedResources: []types.TaggedResource{ - {Tag: &types.Tag{Key: aws.String(common.IdempotencyTagKey), Value: aws.String(token)}}, - }, - }, nil) - - result, err := client.PurchaseCommitment(context.Background(), rsIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) - assert.NoError(t, err) - assert.True(t, result.Success) - assert.True(t, result.ExistingCommitment) - assert.Equal(t, "rn-existing", result.CommitmentID) - mockRS.AssertNotCalled(t, "PurchaseReservedNodeOffering", mock.Anything, mock.Anything) -} - -// TestClient_PurchaseCommitment_Idempotent_NoTagProceeds asserts a first-time -// purchase proceeds when no node carries the token tag, and the new node is -// tagged with the idempotency token afterwards. -func TestClient_PurchaseCommitment_Idempotent_NoTagProceeds(t *testing.T) { - mockRS := &MockRedshiftClient{} - client := rsClientWithAccount(mockRS) - token := common.DeriveIdempotencyToken("exec-2", 0) - - expectRSOffering(mockRS) - mockRS.On("DescribeReservedNodes", mock.Anything, mock.Anything). - Return(&redshift.DescribeReservedNodesOutput{}, nil) - mockRS.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything). - Return(&redshift.PurchaseReservedNodeOfferingOutput{ - ReservedNode: &types.ReservedNode{ReservedNodeId: aws.String("rn-new"), State: aws.String("payment-pending")}, - }, nil) - mockRS.On("CreateTags", mock.Anything, mock.MatchedBy(func(in *redshift.CreateTagsInput) bool { - for _, tag := range in.Tags { - if aws.ToString(tag.Key) == common.IdempotencyTagKey && aws.ToString(tag.Value) == token { - return true - } - } - return false - })).Return(&redshift.CreateTagsOutput{}, nil) - - result, err := client.PurchaseCommitment(context.Background(), rsIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) - assert.NoError(t, err) - assert.True(t, result.Success) - assert.False(t, result.ExistingCommitment) - assert.Equal(t, "rn-new", result.CommitmentID) - mockRS.AssertExpectations(t) -} - -// TestClient_PurchaseCommitment_Idempotent_FailLoudOnLookupError asserts a -// DescribeTags failure during the guard fails loud and does NOT purchase. -func TestClient_PurchaseCommitment_Idempotent_FailLoudOnLookupError(t *testing.T) { - mockRS := &MockRedshiftClient{} - client := rsClientWithAccount(mockRS) - token := common.DeriveIdempotencyToken("exec-3", 0) - - expectRSOffering(mockRS) - mockRS.On("DescribeReservedNodes", mock.Anything, mock.Anything). - Return(&redshift.DescribeReservedNodesOutput{ - ReservedNodes: []types.ReservedNode{{ReservedNodeId: aws.String("rn-x"), State: aws.String("active")}}, - }, nil) - mockRS.On("DescribeTags", mock.Anything, mock.Anything). - Return((*redshift.DescribeTagsOutput)(nil), fmt.Errorf("access denied")) - - result, err := client.PurchaseCommitment(context.Background(), rsIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) - assert.Error(t, err) - assert.False(t, result.Success) - assert.False(t, result.ExistingCommitment) - assert.Contains(t, err.Error(), "refusing to purchase") - mockRS.AssertNotCalled(t, "PurchaseReservedNodeOffering", mock.Anything, mock.Anything) -} - // TestFindOfferingID_PaginationCapFires asserts that findOfferingID returns a // "pagination cap reached" error after maxOfferingPages empty pages and does NOT // make a (maxOfferingPages+1)th call (issue #688). @@ -1466,61 +1100,6 @@ func TestFindOfferingID_HappyPath(t *testing.T) { assert.Equal(t, "offering-ok", id) } -// TestClient_tagReservedNode_NameTagPresent asserts that tagReservedNode (issue -// #687) stamps a self-describing Name tag on the Redshift reserved node. The -// Redshift purchase API has no customer-supplied node ID, so the Name tag is -// the only way to identify the node in the AWS console without CUDly. -func TestClient_tagReservedNode_NameTagPresent(t *testing.T) { - t.Parallel() - mockRS := &MockRedshiftClient{} - mockSTS := &MockRedshiftSTSClient{} - client := &Client{ - client: mockRS, - stsClient: mockSTS, - region: "eu-central-1", - } - - rec := common.Recommendation{ - Service: common.ServiceDataWarehouse, - ResourceType: "ra3.4xlarge", - Count: 1, - PaymentOption: "no-upfront", - Term: "1yr", - Region: "eu-central-1", - } - - mockSTS.On("GetCallerIdentity", mock.Anything, mock.Anything). - Return(&sts.GetCallerIdentityOutput{Account: aws.String("111222333444")}, nil) - - var capturedTags []types.Tag - mockRS.On("CreateTags", mock.Anything, mock.MatchedBy(func(in *redshift.CreateTagsInput) bool { - capturedTags = in.Tags - return true - })).Return(&redshift.CreateTagsOutput{}, nil) - - err := client.tagReservedNode(context.Background(), "rn-tag-test", rec, "test-source", "") - assert.NoError(t, err) - - tagMap := make(map[string]string, len(capturedTags)) - for _, tag := range capturedTags { - tagMap[aws.ToString(tag.Key)] = aws.ToString(tag.Value) - } - - name, ok := tagMap["Name"] - assert.True(t, ok, "Name tag must be present in CreateTags call") - assert.True(t, len(name) > 0, "Name tag must be non-empty") - assert.LessOrEqual(t, len(name), 60, "Name must fit the 60-char cap") - // Key segments that make the node self-describing without CUDly: - assert.Contains(t, name, "redshift", "Name must carry the service code") - assert.Contains(t, name, "eu-central-1", "Name must embed the region") - assert.Contains(t, name, "ra3-4xlarge", "Name must embed the SKU (dots->hyphens)") - assert.Contains(t, name, "1x", "Name must embed the count") - assert.Contains(t, name, "1yr", "Name must embed the term") - - mockRS.AssertExpectations(t) - mockSTS.AssertExpectations(t) -} - // rsPaymentRec builds a Redshift recommendation with the given payment option. func rsPaymentRec(paymentOption string) common.Recommendation { return common.Recommendation{ @@ -1735,8 +1314,7 @@ func TestFindOfferingID_EmptyStringTokenEndsPagination(t *testing.T) { } // Redshift spells an in-flight payment "pending-payment" (issue #142). Such a -// node is already owned: it must be returned to duplicate detection and seen by -// the idempotency tag guard, or a retry buys it a second time. +// node is already owned and must be returned to duplicate detection. func TestClient_PendingPaymentNodeIsOwned(t *testing.T) { pending := types.ReservedNode{ ReservedNodeId: aws.String("rn-pending"), @@ -1747,32 +1325,20 @@ func TestClient_PendingPaymentNodeIsOwned(t *testing.T) { StartTime: aws.Time(time.Now()), } mockRS := &MockRedshiftClient{} - client := rsClientWithAccount(mockRS) - token := common.DeriveIdempotencyToken("exec-1", 0) + client := &Client{client: mockRS, region: "eu-west-1"} mockRS.On("DescribeReservedNodes", mock.Anything, mock.Anything). Return(&redshift.DescribeReservedNodesOutput{ReservedNodes: []types.ReservedNode{pending}}, nil) - mockRS.On("DescribeTags", mock.Anything, mock.Anything). - Return(&redshift.DescribeTagsOutput{ - TaggedResources: []types.TaggedResource{ - {Tag: &types.Tag{Key: aws.String(common.IdempotencyTagKey), Value: aws.String(token)}}, - }, - }, nil) commitments, err := client.GetExistingCommitments(context.Background()) require.NoError(t, err) require.Len(t, commitments, 1) assert.Equal(t, common.CommitmentStatePaymentPending, commitments[0].State) - - id, found, err := client.scanNodesForToken(context.Background(), []types.ReservedNode{pending}, "123456789012", token) - require.NoError(t, err) - assert.True(t, found) - assert.Equal(t, "rn-pending", id) } -// Issue #211: a failed purchase after a clean token lookup is not an adoption. +// Issue #211: a failed purchase after a clean lookup is not an adoption. func TestClient_PurchaseCommitment_Idempotent_PurchaseFailureNotFlagged(t *testing.T) { mockRS := &MockRedshiftClient{} - client := rsClientWithAccount(mockRS) + client := &Client{client: mockRS, region: "eu-west-1"} token := common.DeriveIdempotencyToken("exec-211", 0) expectRSOffering(mockRS) @@ -1781,7 +1347,7 @@ func TestClient_PurchaseCommitment_Idempotent_PurchaseFailureNotFlagged(t *testi mockRS.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything). Return((*redshift.PurchaseReservedNodeOfferingOutput)(nil), fmt.Errorf("boom")) - result, err := client.PurchaseCommitment(context.Background(), rsIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) + result, err := client.PurchaseCommitment(context.Background(), rsIdemRec(), common.PurchaseOptions{IdempotencyToken: token, RequestedAt: time.Now().Add(-time.Hour)}) assert.Error(t, err) assert.False(t, result.Success) assert.False(t, result.ExistingCommitment) diff --git a/providers/aws/services/redshift/dedupe.go b/providers/aws/services/redshift/dedupe.go new file mode 100644 index 00000000..789cb1d9 --- /dev/null +++ b/providers/aws/services/redshift/dedupe.go @@ -0,0 +1,134 @@ +package redshift + +import ( + "context" + "errors" + "fmt" + "log" + "strings" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/redshift" + redshifttypes "github.com/aws/aws-sdk-go-v2/service/redshift/types" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" +) + +const ( + // maxReservedNodePages bounds the dedupe scan. An incomplete scan refuses + // the purchase rather than reporting no match. + maxReservedNodePages = 10 + // redshiftClockSkew is the slack between the caller's RequestedAt and AWS's + // node StartTime. The window deliberately has no upper bound: a late + // StartTime must never cause a double buy. + redshiftClockSkew = 5 * time.Minute + // recoverLookupTimeout bounds the post-error lookup, which runs on a + // context detached from the caller's cancellation. + recoverLookupTimeout = 30 * time.Second +) + +// errRequestedAtRequired refuses a tokenized purchase that cannot be matched. +var errRequestedAtRequired = errors.New("RequestedAt is required with an idempotency token: Redshift reserved nodes cannot be tagged, so an earlier purchase is found by start time; refusing to purchase to avoid a possible double-buy") + +// findMatchingNodes lists the IDs of owned reserved nodes with the given +// offering ID and node count whose StartTime is not before requestedAt minus +// redshiftClockSkew. A node missing StartTime, offering ID or +// node count counts as a match on that field (fail closed). The offering ID fixes node type, term and offering type. +// +// Caller contract: because the window is open-ended, the caller must not +// dispatch another Redshift purchase for the same account, region, offering ID +// and node count while a re-drive of an earlier lineage is still possible (the +// platform's execution claim provides this). Otherwise a re-drive may adopt the +// later node or be refused as ambiguous. +func (c *Client) findMatchingNodes(ctx context.Context, offeringID string, count int, requestedAt time.Time) ([]string, error) { + since := requestedAt.Add(-redshiftClockSkew) + var ids []string + var marker *string + for page := 0; page < maxReservedNodePages; page++ { + if err := ctx.Err(); err != nil { + return nil, err + } + response, err := c.client.DescribeReservedNodes(ctx, &redshift.DescribeReservedNodesInput{ + Marker: marker, + MaxRecords: aws.Int32(100), + }) + if err != nil { + return nil, fmt.Errorf("failed to describe reserved nodes for dedupe: %w", err) + } + for i := range response.ReservedNodes { + if nodeMatches(&response.ReservedNodes[i], offeringID, count, since) { + ids = append(ids, aws.ToString(response.ReservedNodes[i].ReservedNodeId)) + } + } + if aws.ToString(response.Marker) == "" { + return ids, nil + } + marker = response.Marker + } + return nil, fmt.Errorf("pagination cap reached after %d pages for Redshift reserved-node dedupe lookup", maxReservedNodePages) +} + +func nodeMatches(node *redshifttypes.ReservedNode, offeringID string, count int, since time.Time) bool { + if _, owned := ownedNodeState(node.State); !owned { + return false + } + if node.ReservedNodeOfferingId != nil && *node.ReservedNodeOfferingId != offeringID { + return false + } + if node.NodeCount != nil && int(*node.NodeCount) != count { + return false + } + return node.StartTime == nil || !node.StartTime.Before(since) +} + +// guardAgainstRepurchase runs before a tokenized purchase. It reports done +// when the purchase must not proceed: result then carries either the adopted +// node (Success) or the refusal in result.Error. +func (c *Client) guardAgainstRepurchase(ctx context.Context, result *common.PurchaseResult, offeringID string, count int, opts common.PurchaseOptions) (done bool) { + if opts.RequestedAt.IsZero() { + result.Error = errRequestedAtRequired + return true + } + ids, err := c.findMatchingNodes(ctx, offeringID, count, opts.RequestedAt) + switch { + case err != nil: + result.Error = fmt.Errorf("idempotency lookup failed before Redshift purchase (refusing to purchase to avoid a possible double-buy): %w", err) + case len(ids) > 1: + result.Error = fmt.Errorf("ambiguous Redshift dedupe: %d reserved nodes (%s) match offering %s, node count %d, started since %s; refusing to purchase to avoid a possible double-buy; check the account and, if the commitment is missing, start a new purchase", + len(ids), strings.Join(ids, ", "), offeringID, count, opts.RequestedAt.Add(-redshiftClockSkew).UTC().Format(time.RFC3339)) + case len(ids) == 1: + log.Printf("Redshift reserved node %s already matches idempotency token %s; skipping purchase", ids[0], common.MaskToken(opts.IdempotencyToken)) + adopt(result, ids[0]) + return true + default: + return false + } + return true +} + +// adoptAfterUnknownOutcome runs after a purchase error. When the outcome is +// unknown and exactly one node matches, the purchase did happen: adopt it. A +// missing or ambiguous match leaves the original error in place. +func (c *Client) adoptAfterUnknownOutcome(ctx context.Context, result *common.PurchaseResult, offeringID string, count int, opts common.PurchaseOptions) bool { + if !errors.Is(result.Error, common.ErrOutcomeUnknown) || opts.IdempotencyToken == "" || opts.RequestedAt.IsZero() { + return false + } + lookupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), recoverLookupTimeout) + defer cancel() + ids, err := c.findMatchingNodes(lookupCtx, offeringID, count, opts.RequestedAt) + if err != nil || len(ids) != 1 { + log.Printf("Redshift purchase outcome unknown and not recoverable (matches=%d, lookup error=%v)", len(ids), err) + return false + } + log.Printf("Redshift purchase for token %s had an unknown outcome; adopted matching node %s", common.MaskToken(opts.IdempotencyToken), ids[0]) + adopt(result, ids[0]) + return true +} + +func adopt(result *common.PurchaseResult, nodeID string) { + result.Success = true + result.ExistingCommitment = true + result.CommitmentID = nodeID + result.Error = nil +} diff --git a/providers/aws/services/redshift/dedupe_test.go b/providers/aws/services/redshift/dedupe_test.go new file mode 100644 index 00000000..2c226c5f --- /dev/null +++ b/providers/aws/services/redshift/dedupe_test.go @@ -0,0 +1,313 @@ +package redshift + +import ( + "context" + "errors" + "fmt" + "net/http" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "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/redshift" + "github.com/aws/aws-sdk-go-v2/service/redshift/types" + "github.com/aws/smithy-go" + smithyhttp "github.com/aws/smithy-go/transport/http" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// All tests here are fixture-based: mocks only, no AWS calls. + +var dedupeReqAt = time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + +func dedupeOpts() common.PurchaseOptions { + return common.PurchaseOptions{IdempotencyToken: common.DeriveIdempotencyToken("dedupe", 0), RequestedAt: dedupeReqAt} +} + +// dnode builds a node matching rsIdemRec (offering-1, count 1) unless altered. +func dnode(id, state string, start time.Time, alter ...func(*types.ReservedNode)) types.ReservedNode { + n := types.ReservedNode{ + ReservedNodeId: aws.String(id), ReservedNodeOfferingId: aws.String("offering-1"), + NodeCount: aws.Int32(1), State: aws.String(state), StartTime: aws.Time(start), + } + for _, f := range alter { + f(&n) + } + return n +} + +// expectNodePages serves pages of DescribeReservedNodes keyed by marker. +func expectNodePages(m *MockRedshiftClient, pages ...[]types.ReservedNode) { + for i, nodes := range pages { + want := "" + if i > 0 { + want = fmt.Sprintf("p%d", i) + } + out := &redshift.DescribeReservedNodesOutput{ReservedNodes: nodes} + if i < len(pages)-1 { + out.Marker = aws.String(fmt.Sprintf("p%d", i+1)) + } + m.On("DescribeReservedNodes", mock.Anything, mock.MatchedBy(func(in *redshift.DescribeReservedNodesInput) bool { + return aws.ToString(in.Marker) == want + })).Return(out, nil).Maybe() + } +} + +func expectBuy(m *MockRedshiftClient) { + m.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything). + Return(&redshift.PurchaseReservedNodeOfferingOutput{ReservedNode: &types.ReservedNode{ReservedNodeId: aws.String("rn-new")}}, nil).Maybe() +} + +func newDedupe(pages ...[]types.ReservedNode) (*Client, *MockRedshiftClient) { + m := &MockRedshiftClient{} + expectRSOffering(m) + expectNodePages(m, pages...) + expectBuy(m) + return &Client{client: m, region: "eu-west-1"}, m +} + +func TestDedupe_Found(t *testing.T) { + inWindow := dedupeReqAt.Add(time.Minute) + for _, tc := range []struct { + name string + node types.ReservedNode + }{ + {"active in window", dnode("rn-1", "active", inWindow)}, + {"start exactly at lower bound", dnode("rn-1", "active", dedupeReqAt.Add(-redshiftClockSkew))}, + {"start long after request, no upper bound", dnode("rn-1", "active", dedupeReqAt.Add(72*time.Hour))}, + {"pending-payment", dnode("rn-1", "pending-payment", inWindow)}, + {"exchanging", dnode("rn-1", "exchanging", inWindow)}, + {"nil offering ID", dnode("rn-1", "active", inWindow, func(n *types.ReservedNode) { n.ReservedNodeOfferingId = nil })}, + {"nil node count", dnode("rn-1", "active", inWindow, func(n *types.ReservedNode) { n.NodeCount = nil })}, + {"nil StartTime", dnode("rn-1", "active", inWindow, func(n *types.ReservedNode) { n.StartTime = nil })}, + } { + t.Run(tc.name, func(t *testing.T) { + c, m := newDedupe([]types.ReservedNode{tc.node}) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), dedupeOpts()) + require.NoError(t, err) + assert.True(t, result.Success) + assert.True(t, result.ExistingCommitment) + assert.Equal(t, "rn-1", result.CommitmentID) + m.AssertNotCalled(t, "PurchaseReservedNodeOffering", mock.Anything, mock.Anything) + }) + } +} + +func TestDedupe_NotFoundPurchases(t *testing.T) { + inWindow := dedupeReqAt.Add(time.Minute) + for _, tc := range []struct { + name string + node types.ReservedNode + }{ + {"different offering", dnode("rn-1", "active", inWindow, func(n *types.ReservedNode) { n.ReservedNodeOfferingId = aws.String("offering-2") })}, + {"different count", dnode("rn-1", "active", inWindow, func(n *types.ReservedNode) { n.NodeCount = aws.Int32(2) })}, + {"started before window", dnode("rn-1", "active", dedupeReqAt.Add(-redshiftClockSkew-time.Second))}, + {"retired", dnode("rn-1", "retired", inWindow)}, + {"payment-failed", dnode("rn-1", "payment-failed", inWindow)}, + } { + t.Run(tc.name, func(t *testing.T) { + c, m := newDedupe([]types.ReservedNode{tc.node}) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), dedupeOpts()) + require.NoError(t, err) + assert.False(t, result.ExistingCommitment) + assert.Equal(t, "rn-new", result.CommitmentID) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) + }) + } +} + +func TestDedupe_AmbiguousRefuses(t *testing.T) { + c, m := newDedupe([]types.ReservedNode{dnode("rn-a", "active", dedupeReqAt), dnode("rn-b", "pending-payment", dedupeReqAt)}) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), dedupeOpts()) + require.Error(t, err) + assert.False(t, result.Success) + assert.Contains(t, err.Error(), "rn-a") + assert.Contains(t, err.Error(), "rn-b") + assert.Contains(t, err.Error(), "check the account") + assert.Contains(t, err.Error(), "start a new purchase") + m.AssertNotCalled(t, "PurchaseReservedNodeOffering", mock.Anything, mock.Anything) +} + +func TestDedupe_LookupErrorRefuses(t *testing.T) { + m := &MockRedshiftClient{} + expectRSOffering(m) + expectBuy(m) + m.On("DescribeReservedNodes", mock.Anything, mock.Anything).Return((*redshift.DescribeReservedNodesOutput)(nil), errors.New("access denied")) + c := &Client{client: m, region: "eu-west-1"} + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), dedupeOpts()) + require.Error(t, err) + assert.False(t, result.Success) + assert.Contains(t, err.Error(), "refusing to purchase to avoid a possible double-buy") + assert.Contains(t, err.Error(), "access denied") + m.AssertNotCalled(t, "PurchaseReservedNodeOffering", mock.Anything, mock.Anything) +} + +func TestDedupe_PageCap(t *testing.T) { + other := []types.ReservedNode{dnode("rn-x", "active", dedupeReqAt, func(n *types.ReservedNode) { n.ReservedNodeOfferingId = aws.String("other") })} + t.Run("cap refuses", func(t *testing.T) { + pages := make([][]types.ReservedNode, maxReservedNodePages+1) + for i := range pages { + pages[i] = other + } + c, m := newDedupe(pages...) + _, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), dedupeOpts()) + require.Error(t, err) + assert.Contains(t, err.Error(), "pagination cap reached after 10 pages") + assert.Contains(t, err.Error(), "refusing to purchase") + m.AssertNumberOfCalls(t, "DescribeReservedNodes", maxReservedNodePages) + m.AssertNotCalled(t, "PurchaseReservedNodeOffering", mock.Anything, mock.Anything) + }) + t.Run("match on page two", func(t *testing.T) { + c, m := newDedupe(other, []types.ReservedNode{dnode("rn-2", "active", dedupeReqAt)}) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), dedupeOpts()) + require.NoError(t, err) + assert.Equal(t, "rn-2", result.CommitmentID) + m.AssertNotCalled(t, "PurchaseReservedNodeOffering", mock.Anything, mock.Anything) + }) +} + +func TestDedupe_TokenWithoutRequestedAtRefuses(t *testing.T) { + c, m := newDedupe() + opts := dedupeOpts() + opts.RequestedAt = time.Time{} + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), opts) + assert.EqualError(t, err, "RequestedAt is required with an idempotency token: Redshift reserved nodes cannot be tagged, so an earlier purchase is found by start time; refusing to purchase to avoid a possible double-buy") + assert.False(t, result.Success) + m.AssertNotCalled(t, "DescribeReservedNodes", mock.Anything, mock.Anything) + m.AssertNotCalled(t, "PurchaseReservedNodeOffering", mock.Anything, mock.Anything) +} + +func TestDedupe_EmptyTokenDoesNoLookup(t *testing.T) { + c, m := newDedupe() + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), common.PurchaseOptions{Source: common.PurchaseSourceCLI}) + require.NoError(t, err) + assert.Equal(t, "rn-new", result.CommitmentID) + m.AssertNotCalled(t, "DescribeReservedNodes", mock.Anything, mock.Anything) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) +} + +func TestDedupe_CanceledContextRefusesBeforeScan(t *testing.T) { + c, m := newDedupe() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err := c.findMatchingNodes(ctx, "offering-1", 1, dedupeReqAt) + assert.ErrorIs(t, err, context.Canceled) + m.AssertNotCalled(t, "DescribeReservedNodes", mock.Anything, mock.Anything) +} + +func unknownOutcome() error { + return fmt.Errorf("%w: connection reset", common.ErrOutcomeUnknown) +} + +func clean4xx() error { + return &awshttp.ResponseError{ResponseError: &smithyhttp.ResponseError{ + Response: &smithyhttp.Response{Response: &http.Response{StatusCode: http.StatusBadRequest}}, + Err: &smithy.GenericAPIError{Code: "InvalidParameterValue", Message: "bad"}, + }} +} + +// recoverMock returns an empty guard lookup first, then the post-error lookup. +func recoverMock(buy func(*MockRedshiftClient), second ...types.ReservedNode) (*Client, *MockRedshiftClient) { + m := &MockRedshiftClient{} + expectRSOffering(m) + buy(m) + m.On("DescribeReservedNodes", mock.Anything, mock.Anything).Return(&redshift.DescribeReservedNodesOutput{}, nil).Once() + m.On("DescribeReservedNodes", mock.Anything, mock.Anything).Return(&redshift.DescribeReservedNodesOutput{ReservedNodes: second}, nil).Maybe() + return &Client{client: m, region: "eu-west-1"}, m +} + +func buyFails(err error) func(*MockRedshiftClient) { + return func(m *MockRedshiftClient) { + m.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything).Return((*redshift.PurchaseReservedNodeOfferingOutput)(nil), err) + } +} + +func buyEmpty(m *MockRedshiftClient) { + m.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything).Return(&redshift.PurchaseReservedNodeOfferingOutput{}, nil) +} + +func TestDedupe_RecoverAfterUnknownOutcome(t *testing.T) { + one := dnode("rn-bought", "payment-pending", dedupeReqAt.Add(time.Second)) + two := dnode("rn-bought2", "active", dedupeReqAt.Add(time.Second)) + for _, tc := range []struct { + name string + buy func(*MockRedshiftClient) + nodes []types.ReservedNode + adopt bool + }{ + {"timeout then one match adopts", buyFails(unknownOutcome()), []types.ReservedNode{one}, true}, + {"empty 200 then one match adopts", buyEmpty, []types.ReservedNode{one}, true}, + {"timeout then none keeps original error", buyFails(unknownOutcome()), nil, false}, + {"timeout then ambiguous keeps original error", buyFails(unknownOutcome()), []types.ReservedNode{one, two}, false}, + {"empty 200 then none keeps original error", buyEmpty, nil, false}, + } { + t.Run(tc.name, func(t *testing.T) { + c, m := recoverMock(tc.buy, tc.nodes...) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), dedupeOpts()) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) + m.AssertNumberOfCalls(t, "DescribeReservedNodes", 2) + if tc.adopt { + require.NoError(t, err) + assert.True(t, result.Success) + assert.True(t, result.ExistingCommitment) + assert.Equal(t, "rn-bought", result.CommitmentID) + assert.NoError(t, result.Error) + return + } + assert.ErrorIs(t, err, common.ErrOutcomeUnknown) + assert.False(t, result.Success) + }) + } +} + +func TestDedupe_RecoverLookupErrorKeepsOriginalError(t *testing.T) { + c, m := recoverMock(buyFails(unknownOutcome())) + m.ExpectedCalls = m.ExpectedCalls[:len(m.ExpectedCalls)-1] + m.On("DescribeReservedNodes", mock.Anything, mock.Anything).Return((*redshift.DescribeReservedNodesOutput)(nil), errors.New("throttled")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), dedupeOpts()) + assert.ErrorIs(t, err, common.ErrOutcomeUnknown) + assert.False(t, result.Success) +} + +func TestDedupe_Clean4xxDoesNoRecoveryLookup(t *testing.T) { + c, m := recoverMock(buyFails(clean4xx()), dnode("rn-bought", "active", dedupeReqAt)) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), dedupeOpts()) + require.Error(t, err) + assert.NotErrorIs(t, err, common.ErrOutcomeUnknown) + assert.False(t, result.Success) + m.AssertNumberOfCalls(t, "DescribeReservedNodes", 1) +} + +func TestDedupe_RecoverLookupIsDetachedAndBounded(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + m := &MockRedshiftClient{} + expectRSOffering(m) + m.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything). + Run(func(mock.Arguments) { cancel() }). + Return((*redshift.PurchaseReservedNodeOfferingOutput)(nil), unknownOutcome()) + m.On("DescribeReservedNodes", mock.Anything, mock.Anything).Return(&redshift.DescribeReservedNodesOutput{}, nil).Once() + var lookupErr error + var deadline time.Time + var hasDeadline bool + m.On("DescribeReservedNodes", mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { + lookupCtx := args.Get(0).(context.Context) + lookupErr = lookupCtx.Err() + deadline, hasDeadline = lookupCtx.Deadline() + }). + Return(&redshift.DescribeReservedNodesOutput{ReservedNodes: []types.ReservedNode{dnode("rn-bought", "active", dedupeReqAt)}}, nil) + c := &Client{client: m, region: "eu-west-1"} + + result, err := c.PurchaseCommitment(ctx, rsIdemRec(), dedupeOpts()) + require.NoError(t, err) + assert.Equal(t, "rn-bought", result.CommitmentID) + assert.NoError(t, lookupErr, "the lookup must survive the caller's cancellation") + require.True(t, hasDeadline, "the lookup must carry a deadline") + assert.LessOrEqual(t, time.Until(deadline), recoverLookupTimeout) + assert.Equal(t, 30*time.Second, recoverLookupTimeout) +}