diff --git a/pkg/common/identifiers.go b/pkg/common/identifiers.go index 599172195..875f3e793 100644 --- a/pkg/common/identifiers.go +++ b/pkg/common/identifiers.go @@ -35,3 +35,32 @@ func SanitizeReservationID(id, fallbackPrefix string) string { } return s } + +// idempotencyIDTokenLen is how many leading hex characters of the +// idempotency token are folded into a derived reservation ID. 40 hex chars = +// 160 bits, collision-free at any realistic purchase volume, and short enough +// to keep the prefixed result under every AWS reserved-instance/node ID length +// limit (RDS being the tightest). +const idempotencyIDTokenLen = 40 + +// IdempotentReservationID derives a deterministic, AWS-safe reservation ID from +// an idempotency token (issue #641). The same token always yields the same ID, +// so a re-driven purchase reuses the identical customer-supplied reservation ID +// and AWS rejects the duplicate server-side (RDS/ElastiCache/MemoryDB each +// return a *AlreadyExists* fault). Returns "" when token is empty so the caller +// keeps its prior non-idempotent (timestamp-based) ID behaviour for call sites +// that supply no token (e.g. the CLI path). +// +// prefix should be a short, lowercase, hyphen-terminated service tag (e.g. +// "rds-id-") so the reservation is identifiable in the console; the token is +// hex so the result needs no further sanitisation beyond SanitizeReservationID's +// invariants. +func IdempotentReservationID(prefix, token string) string { + if token == "" { + return "" + } + if len(token) > idempotencyIDTokenLen { + token = token[:idempotencyIDTokenLen] + } + return SanitizeReservationID(prefix+token, prefix) +} diff --git a/pkg/common/identifiers_test.go b/pkg/common/identifiers_test.go new file mode 100644 index 000000000..880a339e7 --- /dev/null +++ b/pkg/common/identifiers_test.go @@ -0,0 +1,33 @@ +package common + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestIdempotentReservationID_DeterministicAndSafe(t *testing.T) { + token := DeriveIdempotencyToken("exec-1", 0) + + a := IdempotentReservationID("rds-id-", token) + b := IdempotentReservationID("rds-id-", token) + + assert.Equal(t, a, b, "same token must yield the same reservation ID") + assert.True(t, strings.HasPrefix(a, "rds-id-"), "must carry the prefix for console identifiability") + assert.NotContains(t, a, "--", "must not contain consecutive hyphens") + assert.False(t, strings.HasSuffix(a, "-"), "must not end with a hyphen") + // prefix (7) + 40 hex chars = 47, well under RDS's tightest ID length cap. + assert.LessOrEqual(t, len(a), 60, "must stay under the tightest AWS reservation-ID length cap") +} + +func TestIdempotentReservationID_DistinctTokensDistinctIDs(t *testing.T) { + id0 := IdempotentReservationID("rds-id-", DeriveIdempotencyToken("exec-1", 0)) + id1 := IdempotentReservationID("rds-id-", DeriveIdempotencyToken("exec-1", 1)) + assert.NotEqual(t, id0, id1, "different recs in an execution must get different IDs") +} + +func TestIdempotentReservationID_EmptyTokenReturnsEmpty(t *testing.T) { + assert.Equal(t, "", IdempotentReservationID("rds-id-", ""), + "empty token must yield empty so the caller keeps its non-idempotent fallback") +} diff --git a/pkg/common/tokens.go b/pkg/common/tokens.go index 2e52eaf61..b1a299187 100644 --- a/pkg/common/tokens.go +++ b/pkg/common/tokens.go @@ -41,3 +41,20 @@ func DeriveIdempotencyToken(executionID string, recIndex int) string { sum := sha256.Sum256([]byte(fmt.Sprintf("%s:%d", executionID, recIndex))) return hex.EncodeToString(sum[:]) } + +// MaskToken returns a log-safe representation of an idempotency/approval token: +// the first 8 characters followed by an ellipsis, never the full value. This +// keeps just enough of the prefix to correlate log lines for a single purchase +// while avoiding emitting the whole caller-supplied token into persistent logs +// (a stable per-execution identifier that should not leak verbatim). An empty +// token yields "(none)"; a token of 8 chars or fewer is returned unchanged +// since there is nothing left to redact. +func MaskToken(token string) string { + if token == "" { + return "(none)" + } + if len(token) <= 8 { + return token + } + return token[:8] + "..." +} diff --git a/pkg/common/tokens_test.go b/pkg/common/tokens_test.go index 743465e81..2d536a2c6 100644 --- a/pkg/common/tokens_test.go +++ b/pkg/common/tokens_test.go @@ -64,3 +64,24 @@ func TestDeriveIdempotencyToken_FitsClientTokenLimit(t *testing.T) { require.NoError(t, err) assert.Len(t, raw, 32) } + +func TestMaskToken_NeverEmitsFullToken(t *testing.T) { + // CodeRabbit PR #652: the "already exists" skip-purchase log lines must not + // emit the raw idempotency token (a stable per-execution identifier). The + // masked form keeps only an 8-char prefix for log correlation. + full := DeriveIdempotencyToken("exec-abc-123", 0) // 64-char hex digest + masked := MaskToken(full) + + assert.NotEqual(t, full, masked, "masked form must differ from the raw token") + assert.NotContains(t, masked, full, "masked output must not contain the full token") + assert.Less(t, len(masked), len(full), "masked output must be shorter than the raw token") + assert.Equal(t, full[:8]+"...", masked, "masked form is an 8-char prefix plus ellipsis") + assert.Len(t, masked, 11, "8 prefix chars + 3-char ellipsis") +} + +func TestMaskToken_EmptyAndShort(t *testing.T) { + assert.Equal(t, "(none)", MaskToken(""), "empty token must be reported as (none)") + assert.Equal(t, "abc", MaskToken("abc"), "tokens of <=8 chars have nothing to redact") + assert.Equal(t, "12345678", MaskToken("12345678"), "exactly 8 chars is returned unchanged") + assert.Equal(t, "12345678...", MaskToken("123456789"), "9 chars is truncated to 8 + ellipsis") +} diff --git a/providers/aws/services/elasticache/client.go b/providers/aws/services/elasticache/client.go index dd38c8044..6d3cf7266 100644 --- a/providers/aws/services/elasticache/client.go +++ b/providers/aws/services/elasticache/client.go @@ -3,7 +3,9 @@ package elasticache import ( "context" + "errors" "fmt" + "log" "sort" "time" @@ -124,7 +126,26 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, result.Error } - reservationID := common.SanitizeReservationID(fmt.Sprintf("elasticache-%s-%d", rec.ResourceType, time.Now().Unix()), "elasticache-reserved-") + // When an idempotency token is supplied (issue #641) the reservation ID is + // derived deterministically from it, so a re-drive sends the identical + // ReservedCacheNodeId and ElastiCache rejects the duplicate server-side + // (ReservedCacheNodeAlreadyExistsFault). Otherwise keep the prior + // timestamp-based ID (non-idempotent path). + reservationID := common.IdempotentReservationID("elasticache-id-", opts.IdempotencyToken) + if reservationID == "" { + reservationID = common.SanitizeReservationID(fmt.Sprintf("elasticache-%s-%d", rec.ResourceType, time.Now().Unix()), "elasticache-reserved-") + } + + // Idempotency dedupe guard (issue #641): short-circuit if a reservation + // already exists under the derived ID; fail loud on lookup error. + if existingID, shortCircuit, guardErr := c.idempotencyGuard(ctx, opts.IdempotencyToken, reservationID); guardErr != nil { + result.Error = guardErr + return result, result.Error + } else if shortCircuit { + result.Success = true + result.CommitmentID = existingID + return result, nil + } input := &elasticache.PurchaseReservedCacheNodesOfferingInput{ ReservedCacheNodesOfferingId: aws.String(offeringID), @@ -135,6 +156,11 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati response, err := c.client.PurchaseReservedCacheNodesOffering(ctx, input) if err != nil { + if existingID, recovered := c.recoverAlreadyExists(ctx, opts.IdempotencyToken, reservationID, err); recovered { + result.Success = true + result.CommitmentID = existingID + return result, nil + } result.Error = fmt.Errorf("failed to purchase Reserved Cache Node: %w", err) return result, result.Error } @@ -153,6 +179,75 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, nil } +// findReservationByID looks for an active or payment-pending reserved cache node +// with the given ReservedCacheNodeId (issue #641), so a re-driven purchase can +// short-circuit instead of buying a second node. Retired/expired nodes are +// excluded (same state filter as GetExistingCommitments). +func (c *Client) findReservationByID(ctx context.Context, reservationID string) (string, bool, error) { + response, err := c.client.DescribeReservedCacheNodes(ctx, &elasticache.DescribeReservedCacheNodesInput{ + ReservedCacheNodeId: aws.String(reservationID), + }) + if err != nil { + // ElastiCache returns ReservedCacheNodeNotFound for an unknown reservation + // ID; treat that as "not found" (a first-time purchase), not a lookup + // failure. Any other error is a genuine failure. + var notFound *types.ReservedCacheNodeNotFoundFault + if errors.As(err, ¬Found) { + return "", false, nil + } + return "", false, fmt.Errorf("failed to describe reserved cache nodes for idempotency check: %w", err) + } + for _, node := range response.ReservedCacheNodes { + state := aws.ToString(node.State) + if state != "active" && state != "payment-pending" { + continue + } + if node.ReservedCacheNodeId != nil { + return aws.ToString(node.ReservedCacheNodeId), true, nil + } + } + return "", false, nil +} + +// idempotencyGuard short-circuits a re-drive (issue #641): when token is set, it +// reports (existingID, true, nil) if a reservation already exists under +// reservationID, ("", false, nil) for a first-time purchase, or a fail-loud +// error on lookup failure. With an empty token it is a no-op. +func (c *Client) idempotencyGuard(ctx context.Context, token, reservationID string) (string, bool, error) { + if token == "" { + return "", false, nil + } + existingID, found, lookupErr := c.findReservationByID(ctx, reservationID) + if lookupErr != nil { + return "", false, fmt.Errorf("idempotency lookup failed before ElastiCache purchase (refusing to purchase to avoid a possible double-buy): %w", lookupErr) + } + if found { + log.Printf("ElastiCache reservation for idempotency token %s already exists (%s); skipping purchase (issue #641 re-drive)", common.MaskToken(token), existingID) + return existingID, true, nil + } + return "", false, nil +} + +// recoverAlreadyExists handles the native server-side dedupe backstop (issue +// #641): if the by-ID guard missed the existing reservation but AWS rejected the +// duplicate ID with ReservedCacheNodeAlreadyExistsFault, it re-Describes by ID +// and returns (existingID, true) so the re-drive recovers it instead of erroring. +func (c *Client) recoverAlreadyExists(ctx context.Context, token, reservationID string, purchaseErr error) (string, bool) { + if token == "" { + return "", false + } + var already *types.ReservedCacheNodeAlreadyExistsFault + if !errors.As(purchaseErr, &already) { + return "", false + } + existingID, found, lookupErr := c.findReservationByID(ctx, reservationID) + if lookupErr == nil && found { + log.Printf("ElastiCache reservation %s already existed at purchase time; treating as idempotent re-drive (issue #641)", existingID) + return existingID, true + } + return "", false +} + // findOfferingID finds the appropriate Reserved Cache Node offering ID func (c *Client) findOfferingID(ctx context.Context, rec common.Recommendation) (string, error) { details, ok := rec.Details.(*common.CacheDetails) diff --git a/providers/aws/services/elasticache/client_test.go b/providers/aws/services/elasticache/client_test.go index d974afb83..58b8c2ec4 100644 --- a/providers/aws/services/elasticache/client_test.go +++ b/providers/aws/services/elasticache/client_test.go @@ -411,6 +411,112 @@ func TestClient_ConvertPaymentOption(t *testing.T) { } } +func idemRec() common.Recommendation { + return common.Recommendation{ + Service: common.ServiceCache, + ResourceType: "cache.m6g.large", + Count: 1, + PaymentOption: "all-upfront", + Term: "1yr", + Details: &common.CacheDetails{Engine: "redis", NodeType: "cache.m6g.large"}, + } +} + +func expectECOffering(m *MockElastiCacheClient) { + m.On("DescribeReservedCacheNodesOfferings", mock.Anything, mock.Anything). + Return(&elasticache.DescribeReservedCacheNodesOfferingsOutput{ + ReservedCacheNodesOfferings: []types.ReservedCacheNodesOffering{ + { + ReservedCacheNodesOfferingId: aws.String("offering-1"), + CacheNodeType: aws.String("cache.m6g.large"), + ProductDescription: aws.String("redis"), + OfferingType: aws.String("All Upfront"), + Duration: aws.Int32(31536000), + }, + }, + }, nil) +} + +func TestClient_PurchaseCommitment_Idempotent_GuardShortCircuits(t *testing.T) { + mockEC := &MockElastiCacheClient{} + client := &Client{client: mockEC, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-1", 0) + derivedID := common.IdempotentReservationID("elasticache-id-", token) + + expectECOffering(mockEC) + mockEC.On("DescribeReservedCacheNodes", mock.Anything, mock.MatchedBy(func(in *elasticache.DescribeReservedCacheNodesInput) bool { + return aws.ToString(in.ReservedCacheNodeId) == derivedID + })).Return(&elasticache.DescribeReservedCacheNodesOutput{ + ReservedCacheNodes: []types.ReservedCacheNode{{ReservedCacheNodeId: aws.String(derivedID), State: aws.String("active")}}, + }, nil) + + result, err := client.PurchaseCommitment(context.Background(), idemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, derivedID, result.CommitmentID) + mockEC.AssertNotCalled(t, "PurchaseReservedCacheNodesOffering", mock.Anything, mock.Anything) +} + +func TestClient_PurchaseCommitment_Idempotent_NotFoundProceeds(t *testing.T) { + mockEC := &MockElastiCacheClient{} + client := &Client{client: mockEC, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-2", 0) + derivedID := common.IdempotentReservationID("elasticache-id-", token) + + expectECOffering(mockEC) + mockEC.On("DescribeReservedCacheNodes", mock.Anything, mock.Anything). + Return((*elasticache.DescribeReservedCacheNodesOutput)(nil), &types.ReservedCacheNodeNotFoundFault{}) + mockEC.On("PurchaseReservedCacheNodesOffering", mock.Anything, mock.MatchedBy(func(in *elasticache.PurchaseReservedCacheNodesOfferingInput) bool { + return aws.ToString(in.ReservedCacheNodeId) == derivedID + })).Return(&elasticache.PurchaseReservedCacheNodesOfferingOutput{ + ReservedCacheNode: &types.ReservedCacheNode{ReservedCacheNodeId: aws.String(derivedID)}, + }, nil) + + result, err := client.PurchaseCommitment(context.Background(), idemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, derivedID, result.CommitmentID) + mockEC.AssertExpectations(t) +} + +func TestClient_PurchaseCommitment_Idempotent_AlreadyExistsRecovers(t *testing.T) { + mockEC := &MockElastiCacheClient{} + client := &Client{client: mockEC, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-3", 0) + derivedID := common.IdempotentReservationID("elasticache-id-", token) + + expectECOffering(mockEC) + mockEC.On("DescribeReservedCacheNodes", mock.Anything, mock.Anything). + Return((*elasticache.DescribeReservedCacheNodesOutput)(nil), &types.ReservedCacheNodeNotFoundFault{}).Once() + mockEC.On("PurchaseReservedCacheNodesOffering", mock.Anything, mock.Anything). + Return((*elasticache.PurchaseReservedCacheNodesOfferingOutput)(nil), &types.ReservedCacheNodeAlreadyExistsFault{}) + mockEC.On("DescribeReservedCacheNodes", mock.Anything, mock.Anything). + Return(&elasticache.DescribeReservedCacheNodesOutput{ + ReservedCacheNodes: []types.ReservedCacheNode{{ReservedCacheNodeId: aws.String(derivedID), State: aws.String("active")}}, + }, nil).Once() + + result, err := client.PurchaseCommitment(context.Background(), idemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, derivedID, result.CommitmentID) +} + +func TestClient_PurchaseCommitment_Idempotent_FailLoudOnLookupError(t *testing.T) { + mockEC := &MockElastiCacheClient{} + client := &Client{client: mockEC, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-4", 0) + + expectECOffering(mockEC) + mockEC.On("DescribeReservedCacheNodes", mock.Anything, mock.Anything). + Return((*elasticache.DescribeReservedCacheNodesOutput)(nil), fmt.Errorf("access denied")) + + result, err := client.PurchaseCommitment(context.Background(), idemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.Error(t, err) + assert.False(t, result.Success) + assert.Contains(t, err.Error(), "refusing to purchase") + mockEC.AssertNotCalled(t, "PurchaseReservedCacheNodesOffering", mock.Anything, mock.Anything) +} + func TestCreatePurchaseTags_IncludesPurchaseAutomation(t *testing.T) { c := &Client{} rec := common.Recommendation{ResourceType: "cache.m5.large", Region: "us-east-1"} diff --git a/providers/aws/services/memorydb/client.go b/providers/aws/services/memorydb/client.go index 73e32abff..0d7710756 100644 --- a/providers/aws/services/memorydb/client.go +++ b/providers/aws/services/memorydb/client.go @@ -3,7 +3,9 @@ package memorydb import ( "context" + "errors" "fmt" + "log" "sort" "time" @@ -120,7 +122,26 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, result.Error } - reservationID := common.SanitizeReservationID(fmt.Sprintf("memorydb-%s-%d", rec.ResourceType, time.Now().Unix()), "memorydb-reserved-") + // When an idempotency token is supplied (issue #641) the reservation ID is + // derived deterministically from it, so a re-drive sends the identical + // ReservationId and MemoryDB rejects the duplicate server-side + // (ReservedNodeAlreadyExistsFault). Otherwise keep the prior timestamp-based + // ID (non-idempotent path). + reservationID := common.IdempotentReservationID("memorydb-id-", opts.IdempotencyToken) + if reservationID == "" { + reservationID = common.SanitizeReservationID(fmt.Sprintf("memorydb-%s-%d", rec.ResourceType, time.Now().Unix()), "memorydb-reserved-") + } + + // Idempotency dedupe guard (issue #641): short-circuit if a reserved node + // already exists under the derived ID; fail loud on lookup error. + if existingID, shortCircuit, guardErr := c.idempotencyGuard(ctx, opts.IdempotencyToken, reservationID); guardErr != nil { + result.Error = guardErr + return result, result.Error + } else if shortCircuit { + result.Success = true + result.CommitmentID = existingID + return result, nil + } input := &memorydb.PurchaseReservedNodesOfferingInput{ ReservedNodesOfferingId: aws.String(offeringID), @@ -131,6 +152,11 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati response, err := c.client.PurchaseReservedNodesOffering(ctx, input) if err != nil { + if existingID, recovered := c.recoverAlreadyExists(ctx, opts.IdempotencyToken, reservationID, err); recovered { + result.Success = true + result.CommitmentID = existingID + return result, nil + } result.Error = fmt.Errorf("failed to purchase MemoryDB Reserved Nodes: %w", err) return result, result.Error } @@ -147,6 +173,75 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, nil } +// findReservationByID looks for an active or payment-pending MemoryDB reserved +// node with the given ReservationId (issue #641), so a re-driven purchase can +// short-circuit instead of buying a second node. Retired/expired nodes are +// excluded (same state filter as GetExistingCommitments). +func (c *Client) findReservationByID(ctx context.Context, reservationID string) (string, bool, error) { + response, err := c.client.DescribeReservedNodes(ctx, &memorydb.DescribeReservedNodesInput{ + ReservationId: aws.String(reservationID), + }) + if err != nil { + // MemoryDB returns ReservedNodeNotFoundFault for an unknown reservation + // ID; treat that as "not found" (no existing reservation), not a lookup + // failure, so a first-time purchase is not blocked. + var notFound *types.ReservedNodeNotFoundFault + if errors.As(err, ¬Found) { + return "", false, nil + } + return "", false, fmt.Errorf("failed to describe reserved nodes for idempotency check: %w", err) + } + for _, node := range response.ReservedNodes { + state := aws.ToString(node.State) + if state != "active" && state != "payment-pending" { + continue + } + if node.ReservationId != nil { + return aws.ToString(node.ReservationId), true, nil + } + } + return "", false, nil +} + +// idempotencyGuard short-circuits a re-drive (issue #641): when token is set, it +// reports (existingID, true, nil) if a reserved node already exists under +// reservationID, ("", false, nil) for a first-time purchase, or a fail-loud +// error on lookup failure. With an empty token it is a no-op. +func (c *Client) idempotencyGuard(ctx context.Context, token, reservationID string) (string, bool, error) { + if token == "" { + return "", false, nil + } + existingID, found, lookupErr := c.findReservationByID(ctx, reservationID) + if lookupErr != nil { + return "", false, fmt.Errorf("idempotency lookup failed before MemoryDB purchase (refusing to purchase to avoid a possible double-buy): %w", lookupErr) + } + if found { + log.Printf("MemoryDB reservation for idempotency token %s already exists (%s); skipping purchase (issue #641 re-drive)", common.MaskToken(token), existingID) + return existingID, true, nil + } + return "", false, nil +} + +// recoverAlreadyExists handles the native server-side dedupe backstop (issue +// #641): if the by-ID guard missed the existing reservation but AWS rejected the +// duplicate ID with ReservedNodeAlreadyExistsFault, it re-Describes by ID and +// returns (existingID, true) so the re-drive recovers it instead of erroring. +func (c *Client) recoverAlreadyExists(ctx context.Context, token, reservationID string, purchaseErr error) (string, bool) { + if token == "" { + return "", false + } + var already *types.ReservedNodeAlreadyExistsFault + if !errors.As(purchaseErr, &already) { + return "", false + } + existingID, found, lookupErr := c.findReservationByID(ctx, reservationID) + if lookupErr == nil && found { + log.Printf("MemoryDB reservation %s already existed at purchase time; treating as idempotent re-drive (issue #641)", existingID) + return existingID, true + } + return "", false +} + // findOfferingID finds the appropriate Reserved Node offering ID func (c *Client) findOfferingID(ctx context.Context, rec common.Recommendation) (string, error) { requiredMonths := c.getTermMonthsFromString(rec.Term) diff --git a/providers/aws/services/memorydb/client_test.go b/providers/aws/services/memorydb/client_test.go index 06724577b..e56471b56 100644 --- a/providers/aws/services/memorydb/client_test.go +++ b/providers/aws/services/memorydb/client_test.go @@ -679,6 +679,111 @@ func TestClient_GetTermMonthsFromDuration(t *testing.T) { } } +func mdbIdemRec() common.Recommendation { + return common.Recommendation{ + Service: common.ServiceCache, + ResourceType: "db.r6gd.large", + Count: 1, + PaymentOption: "all-upfront", + Term: "1yr", + Details: common.CacheDetails{Engine: "redis", NodeType: "db.r6gd.large"}, + } +} + +func expectMDBOffering(m *MockMemoryDBClient) { + m.On("DescribeReservedNodesOfferings", mock.Anything, mock.Anything). + Return(&memorydb.DescribeReservedNodesOfferingsOutput{ + ReservedNodesOfferings: []types.ReservedNodesOffering{ + { + ReservedNodesOfferingId: aws.String("offering-1"), + NodeType: aws.String("db.r6gd.large"), + Duration: 31536000, + OfferingType: aws.String("All Upfront"), + }, + }, + }, nil) +} + +func TestClient_PurchaseCommitment_Idempotent_GuardShortCircuits(t *testing.T) { + mockMDB := &MockMemoryDBClient{} + client := &Client{client: mockMDB, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-1", 0) + derivedID := common.IdempotentReservationID("memorydb-id-", token) + + expectMDBOffering(mockMDB) + mockMDB.On("DescribeReservedNodes", mock.Anything, mock.MatchedBy(func(in *memorydb.DescribeReservedNodesInput) bool { + return aws.ToString(in.ReservationId) == derivedID + })).Return(&memorydb.DescribeReservedNodesOutput{ + ReservedNodes: []types.ReservedNode{{ReservationId: aws.String(derivedID), State: aws.String("active")}}, + }, nil) + + result, err := client.PurchaseCommitment(context.Background(), mdbIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, derivedID, result.CommitmentID) + mockMDB.AssertNotCalled(t, "PurchaseReservedNodesOffering", mock.Anything, mock.Anything) +} + +func TestClient_PurchaseCommitment_Idempotent_NotFoundProceeds(t *testing.T) { + mockMDB := &MockMemoryDBClient{} + client := &Client{client: mockMDB, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-2", 0) + derivedID := common.IdempotentReservationID("memorydb-id-", token) + + expectMDBOffering(mockMDB) + mockMDB.On("DescribeReservedNodes", mock.Anything, mock.Anything). + Return((*memorydb.DescribeReservedNodesOutput)(nil), &types.ReservedNodeNotFoundFault{}) + mockMDB.On("PurchaseReservedNodesOffering", mock.Anything, mock.MatchedBy(func(in *memorydb.PurchaseReservedNodesOfferingInput) bool { + return aws.ToString(in.ReservationId) == derivedID + })).Return(&memorydb.PurchaseReservedNodesOfferingOutput{ + ReservedNode: &types.ReservedNode{ReservationId: aws.String(derivedID)}, + }, nil) + + result, err := client.PurchaseCommitment(context.Background(), mdbIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, derivedID, result.CommitmentID) + mockMDB.AssertExpectations(t) +} + +func TestClient_PurchaseCommitment_Idempotent_AlreadyExistsRecovers(t *testing.T) { + mockMDB := &MockMemoryDBClient{} + client := &Client{client: mockMDB, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-3", 0) + derivedID := common.IdempotentReservationID("memorydb-id-", token) + + expectMDBOffering(mockMDB) + mockMDB.On("DescribeReservedNodes", mock.Anything, mock.Anything). + Return((*memorydb.DescribeReservedNodesOutput)(nil), &types.ReservedNodeNotFoundFault{}).Once() + mockMDB.On("PurchaseReservedNodesOffering", mock.Anything, mock.Anything). + Return((*memorydb.PurchaseReservedNodesOfferingOutput)(nil), &types.ReservedNodeAlreadyExistsFault{}) + mockMDB.On("DescribeReservedNodes", mock.Anything, mock.Anything). + Return(&memorydb.DescribeReservedNodesOutput{ + ReservedNodes: []types.ReservedNode{{ReservationId: aws.String(derivedID), State: aws.String("active")}}, + }, nil).Once() + + result, err := client.PurchaseCommitment(context.Background(), mdbIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, derivedID, result.CommitmentID) +} + +func TestClient_PurchaseCommitment_Idempotent_FailLoudOnLookupError(t *testing.T) { + mockMDB := &MockMemoryDBClient{} + client := &Client{client: mockMDB, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-4", 0) + + expectMDBOffering(mockMDB) + mockMDB.On("DescribeReservedNodes", mock.Anything, mock.Anything). + Return((*memorydb.DescribeReservedNodesOutput)(nil), fmt.Errorf("access denied")) + + result, err := client.PurchaseCommitment(context.Background(), mdbIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.Error(t, err) + assert.False(t, result.Success) + assert.Contains(t, err.Error(), "refusing to purchase") + mockMDB.AssertNotCalled(t, "PurchaseReservedNodesOffering", mock.Anything, mock.Anything) +} + func TestCreatePurchaseTags_IncludesPurchaseAutomation(t *testing.T) { c := &Client{} rec := common.Recommendation{ResourceType: "db.r6g.large", Region: "us-east-1"} diff --git a/providers/aws/services/opensearch/client.go b/providers/aws/services/opensearch/client.go index b1b0d24fc..389060b1e 100644 --- a/providers/aws/services/opensearch/client.go +++ b/providers/aws/services/opensearch/client.go @@ -3,6 +3,7 @@ package opensearch import ( "context" + "errors" "fmt" "log" "sync" @@ -149,16 +150,41 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, result.Error } - reservationID := common.SanitizeReservationID(fmt.Sprintf("opensearch-%s-%d", rec.ResourceType, time.Now().Unix()), "opensearch-reserved-") + // When an idempotency token is supplied (issue #641) the ReservationName is + // derived deterministically from it. ReservationName is unique per + // account+region, so a re-drive sends the identical name and OpenSearch + // rejects the duplicate server-side (ResourceAlreadyExistsException) — it + // cannot create a second reservation. Otherwise keep the prior timestamp-based + // name (non-idempotent path). + reservationName := common.IdempotentReservationID("opensearch-id-", opts.IdempotencyToken) + if reservationName == "" { + reservationName = common.SanitizeReservationID(fmt.Sprintf("opensearch-%s-%d", rec.ResourceType, time.Now().Unix()), "opensearch-reserved-") + } + + // Idempotency dedupe guard (issue #641): short-circuit if a reservation with + // the derived name already exists; fail loud on lookup error. + if existingID, shortCircuit, guardErr := c.idempotencyGuard(ctx, opts.IdempotencyToken, reservationName); guardErr != nil { + result.Error = guardErr + return result, result.Error + } else if shortCircuit { + result.Success = true + result.CommitmentID = existingID + return result, nil + } input := &opensearch.PurchaseReservedInstanceOfferingInput{ ReservedInstanceOfferingId: aws.String(offeringID), - ReservationName: aws.String(reservationID), + ReservationName: aws.String(reservationName), InstanceCount: aws.Int32(int32(rec.Count)), } response, err := c.client.PurchaseReservedInstanceOffering(ctx, input) if err != nil { + if existingID, recovered := c.recoverAlreadyExists(ctx, opts.IdempotencyToken, reservationName, err); recovered { + result.Success = true + result.CommitmentID = existingID + return result, nil + } result.Error = fmt.Errorf("failed to purchase OpenSearch RI: %w", err) return result, result.Error } @@ -178,6 +204,81 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, nil } +// findReservationByName looks for an active or payment-pending OpenSearch +// reserved instance whose ReservationName matches the given name (issue #641), +// so a re-driven purchase can short-circuit. DescribeReservedInstances has no +// name filter, so it pages through all reservations and matches client-side. +// Retired/expired reservations are excluded (same state filter as +// GetExistingCommitments). +func (c *Client) findReservationByName(ctx context.Context, name string) (string, bool, error) { + var nextToken *string + for { + response, err := c.client.DescribeReservedInstances(ctx, &opensearch.DescribeReservedInstancesInput{ + NextToken: nextToken, + MaxResults: 100, + }) + if err != nil { + return "", false, fmt.Errorf("failed to describe reserved instances for idempotency check: %w", err) + } + for _, ri := range response.ReservedInstances { + if aws.ToString(ri.ReservationName) != name { + continue + } + state := aws.ToString(ri.State) + if state != "active" && state != "payment-pending" { + continue + } + if ri.ReservedInstanceId != nil { + return aws.ToString(ri.ReservedInstanceId), true, nil + } + } + if response.NextToken == nil || aws.ToString(response.NextToken) == "" { + break + } + nextToken = response.NextToken + } + return "", false, nil +} + +// idempotencyGuard short-circuits a re-drive (issue #641): when token is set, it +// reports (existingID, true, nil) if a reservation with reservationName already +// exists, ("", false, nil) for a first-time purchase, or a fail-loud error on +// lookup failure. With an empty token it is a no-op. +func (c *Client) idempotencyGuard(ctx context.Context, token, reservationName string) (string, bool, error) { + if token == "" { + return "", false, nil + } + existingID, found, lookupErr := c.findReservationByName(ctx, reservationName) + if lookupErr != nil { + return "", false, fmt.Errorf("idempotency lookup failed before OpenSearch RI purchase (refusing to purchase to avoid a possible double-buy): %w", lookupErr) + } + if found { + log.Printf("OpenSearch RI for idempotency token %s already exists (%s); skipping purchase (issue #641 re-drive)", common.MaskToken(token), existingID) + return existingID, true, nil + } + return "", false, nil +} + +// recoverAlreadyExists handles the native server-side dedupe backstop (issue +// #641): if the by-name guard missed the existing reservation but OpenSearch +// rejected the duplicate name with ResourceAlreadyExistsException, it re-Describes +// by name and returns (existingID, true) so the re-drive recovers it. +func (c *Client) recoverAlreadyExists(ctx context.Context, token, reservationName string, purchaseErr error) (string, bool) { + if token == "" { + return "", false + } + var already *types.ResourceAlreadyExistsException + if !errors.As(purchaseErr, &already) { + return "", false + } + existingID, found, lookupErr := c.findReservationByName(ctx, reservationName) + if lookupErr == nil && found { + log.Printf("OpenSearch RI %s already existed at purchase time; treating as idempotent re-drive (issue #641)", existingID) + return existingID, true + } + return "", false +} + // resolveAccountID fetches the caller's AWS account ID via STS and caches it. func (c *Client) resolveAccountID(ctx context.Context) (string, error) { c.accountOnce.Do(func() { diff --git a/providers/aws/services/opensearch/client_test.go b/providers/aws/services/opensearch/client_test.go index b49b3df73..929db1b6e 100644 --- a/providers/aws/services/opensearch/client_test.go +++ b/providers/aws/services/opensearch/client_test.go @@ -706,3 +706,110 @@ func TestClient_GetTermMonthsFromDuration(t *testing.T) { }) } } + +func osIdemRec() common.Recommendation { + return common.Recommendation{ + Service: common.ServiceSearch, + ResourceType: "m5.xlarge.search", + Count: 1, + PaymentOption: "all-upfront", + Term: "3yr", + Details: common.SearchDetails{InstanceType: "m5.xlarge.search"}, + } +} + +func expectOSOffering(m *MockOpenSearchClient) { + m.On("DescribeReservedInstanceOfferings", mock.Anything, mock.Anything). + Return(&opensearch.DescribeReservedInstanceOfferingsOutput{ + ReservedInstanceOfferings: []types.ReservedInstanceOffering{ + { + ReservedInstanceOfferingId: aws.String("offering-1"), + InstanceType: types.OpenSearchPartitionInstanceTypeM5XlargeSearch, + Duration: 94608000, + PaymentOption: types.ReservedInstancePaymentOptionAllUpfront, + }, + }, + }, nil) +} + +func TestClient_PurchaseCommitment_Idempotent_GuardShortCircuits(t *testing.T) { + mockOS := &MockOpenSearchClient{} + client := &Client{client: mockOS, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-1", 0) + derivedName := common.IdempotentReservationID("opensearch-id-", token) + + expectOSOffering(mockOS) + mockOS.On("DescribeReservedInstances", mock.Anything, mock.Anything). + Return(&opensearch.DescribeReservedInstancesOutput{ + ReservedInstances: []types.ReservedInstance{ + {ReservedInstanceId: aws.String("os-existing"), ReservationName: aws.String(derivedName), State: aws.String("active")}, + }, + }, nil) + + result, err := client.PurchaseCommitment(context.Background(), osIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, "os-existing", result.CommitmentID) + mockOS.AssertNotCalled(t, "PurchaseReservedInstanceOffering", mock.Anything, mock.Anything) +} + +func TestClient_PurchaseCommitment_Idempotent_NotFoundProceeds(t *testing.T) { + mockOS := &MockOpenSearchClient{} + client := &Client{client: mockOS, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-2", 0) + derivedName := common.IdempotentReservationID("opensearch-id-", token) + + expectOSOffering(mockOS) + mockOS.On("DescribeReservedInstances", mock.Anything, mock.Anything). + Return(&opensearch.DescribeReservedInstancesOutput{}, nil) + mockOS.On("PurchaseReservedInstanceOffering", mock.Anything, mock.MatchedBy(func(in *opensearch.PurchaseReservedInstanceOfferingInput) bool { + return aws.ToString(in.ReservationName) == derivedName + })).Return(&opensearch.PurchaseReservedInstanceOfferingOutput{ReservedInstanceId: aws.String("os-new")}, nil) + + result, err := client.PurchaseCommitment(context.Background(), osIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, "os-new", result.CommitmentID) + mockOS.AssertExpectations(t) +} + +func TestClient_PurchaseCommitment_Idempotent_AlreadyExistsRecovers(t *testing.T) { + mockOS := &MockOpenSearchClient{} + client := &Client{client: mockOS, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-3", 0) + derivedName := common.IdempotentReservationID("opensearch-id-", token) + + expectOSOffering(mockOS) + // Guard misses, purchase rejected, recovery Describe finds it by name. + mockOS.On("DescribeReservedInstances", mock.Anything, mock.Anything). + Return(&opensearch.DescribeReservedInstancesOutput{}, nil).Once() + mockOS.On("PurchaseReservedInstanceOffering", mock.Anything, mock.Anything). + Return((*opensearch.PurchaseReservedInstanceOfferingOutput)(nil), &types.ResourceAlreadyExistsException{}) + mockOS.On("DescribeReservedInstances", mock.Anything, mock.Anything). + Return(&opensearch.DescribeReservedInstancesOutput{ + ReservedInstances: []types.ReservedInstance{ + {ReservedInstanceId: aws.String("os-recovered"), ReservationName: aws.String(derivedName), State: aws.String("payment-pending")}, + }, + }, nil).Once() + + result, err := client.PurchaseCommitment(context.Background(), osIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, "os-recovered", result.CommitmentID) +} + +func TestClient_PurchaseCommitment_Idempotent_FailLoudOnLookupError(t *testing.T) { + mockOS := &MockOpenSearchClient{} + client := &Client{client: mockOS, region: "eu-west-1"} + token := common.DeriveIdempotencyToken("exec-4", 0) + + expectOSOffering(mockOS) + mockOS.On("DescribeReservedInstances", mock.Anything, mock.Anything). + Return((*opensearch.DescribeReservedInstancesOutput)(nil), fmt.Errorf("access denied")) + + result, err := client.PurchaseCommitment(context.Background(), osIdemRec(), common.PurchaseOptions{IdempotencyToken: token}) + assert.Error(t, err) + assert.False(t, result.Success) + assert.Contains(t, err.Error(), "refusing to purchase") + mockOS.AssertNotCalled(t, "PurchaseReservedInstanceOffering", mock.Anything, mock.Anything) +} diff --git a/providers/aws/services/rds/client.go b/providers/aws/services/rds/client.go index feeefa1cf..ca02f0b15 100644 --- a/providers/aws/services/rds/client.go +++ b/providers/aws/services/rds/client.go @@ -3,6 +3,7 @@ package rds import ( "context" + "errors" "fmt" "log" "sort" @@ -139,15 +140,20 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, result.Error } - // Reservation ID (letters, digits, hyphens only; no leading/trailing/double - // hyphen). Prefer the caller-supplied descriptive ID (account/engine/region/ - // size aware) so the reservation is identifiable in the console and billing; - // fall back to a generic one when the caller didn't supply it. - rawID := opts.ReservationID - if rawID == "" { - rawID = fmt.Sprintf("rds-%s-%d", rec.ResourceType, time.Now().Unix()) + reservationID := c.deriveReservationID(rec, opts) + + // Idempotency dedupe guard (issue #641). When a token is supplied, look for a + // reservation already created under the derived ID before buying: if one + // exists this is a re-drive that already succeeded, so short-circuit. A + // lookup error must NOT fall through to a purchase — fail loud. + if existingID, shortCircuit, guardErr := c.idempotencyGuard(ctx, opts.IdempotencyToken, reservationID); guardErr != nil { + result.Error = guardErr + return result, result.Error + } else if shortCircuit { + result.Success = true + result.CommitmentID = existingID + return result, nil } - reservationID := common.SanitizeReservationID(rawID, "rds-reserved-") // Create the purchase request input := &rds.PurchaseReservedDBInstancesOfferingInput{ @@ -159,6 +165,11 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati response, err := c.client.PurchaseReservedDBInstancesOffering(ctx, input) if err != nil { + if existingID, recovered := c.recoverAlreadyExists(ctx, opts.IdempotencyToken, reservationID, err); recovered { + result.Success = true + result.CommitmentID = existingID + return result, nil + } result.Error = fmt.Errorf("failed to purchase RDS RI: %w", err) return result, result.Error } @@ -177,6 +188,95 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, nil } +// deriveReservationID returns the ReservedDBInstanceId to use. When an +// idempotency token is supplied (issue #641) the ID is derived deterministically +// from it, so a re-drive sends the identical ID and RDS rejects the duplicate +// server-side (ReservedDBInstanceAlreadyExistsFault). Otherwise it prefers the +// caller-supplied descriptive ID and falls back to a generic timestamped one +// (prior non-idempotent behaviour). +func (c *Client) deriveReservationID(rec common.Recommendation, opts common.PurchaseOptions) string { + if id := common.IdempotentReservationID("rds-id-", opts.IdempotencyToken); id != "" { + return id + } + rawID := opts.ReservationID + if rawID == "" { + rawID = fmt.Sprintf("rds-%s-%d", rec.ResourceType, time.Now().Unix()) + } + return common.SanitizeReservationID(rawID, "rds-reserved-") +} + +// idempotencyGuard short-circuits a re-drive (issue #641): when token is set, it +// reports (existingID, true, nil) if a reservation already exists under +// reservationID, ("", false, nil) for a first-time purchase, or a fail-loud +// error on lookup failure. With an empty token it is a no-op. +func (c *Client) idempotencyGuard(ctx context.Context, token, reservationID string) (string, bool, error) { + if token == "" { + return "", false, nil + } + existingID, found, lookupErr := c.findReservationByID(ctx, reservationID) + if lookupErr != nil { + return "", false, fmt.Errorf("idempotency lookup failed before RDS RI purchase (refusing to purchase to avoid a possible double-buy): %w", lookupErr) + } + if found { + log.Printf("RDS RI for idempotency token %s already exists (%s); skipping purchase (issue #641 re-drive)", common.MaskToken(token), existingID) + return existingID, true, nil + } + return "", false, nil +} + +// recoverAlreadyExists handles the native server-side dedupe backstop (issue +// #641): if the by-ID guard missed the existing reservation but AWS still +// rejected the duplicate ID with ReservedDBInstanceAlreadyExistsFault, it +// re-Describes by ID and returns (existingID, true) so the re-drive recovers the +// original reservation instead of erroring. +func (c *Client) recoverAlreadyExists(ctx context.Context, token, reservationID string, purchaseErr error) (string, bool) { + if token == "" { + return "", false + } + var already *types.ReservedDBInstanceAlreadyExistsFault + if !errors.As(purchaseErr, &already) { + return "", false + } + existingID, found, lookupErr := c.findReservationByID(ctx, reservationID) + if lookupErr == nil && found { + log.Printf("RDS RI %s already existed at purchase time; treating as idempotent re-drive (issue #641)", existingID) + return existingID, true + } + return "", false +} + +// findReservationByID looks for an active or payment-pending RDS reserved DB +// instance with the given ReservedDBInstanceId (issue #641). It returns the +// reservation ID and true when such a reservation exists, so a re-driven +// purchase can short-circuit. Retired/expired reservations are excluded (same +// state filter as GetExistingCommitments) so a returned reservation does not +// suppress a legitimate fresh purchase. +func (c *Client) findReservationByID(ctx context.Context, reservationID string) (string, bool, error) { + response, err := c.client.DescribeReservedDBInstances(ctx, &rds.DescribeReservedDBInstancesInput{ + ReservedDBInstanceId: aws.String(reservationID), + }) + if err != nil { + // RDS returns ReservedDBInstanceNotFound for an unknown reservation ID; + // treat that as "not found" (a first-time purchase), not a lookup + // failure, so it is not blocked. Any other error is a genuine failure. + var notFound *types.ReservedDBInstanceNotFoundFault + if errors.As(err, ¬Found) { + return "", false, nil + } + return "", false, fmt.Errorf("failed to describe reserved DB instances for idempotency check: %w", err) + } + for _, ri := range response.ReservedDBInstances { + state := aws.ToString(ri.State) + if state != "active" && state != "payment-pending" { + continue + } + if ri.ReservedDBInstanceId != nil { + return aws.ToString(ri.ReservedDBInstanceId), true, nil + } + } + return "", false, nil +} + // findOfferingID finds the appropriate RDS Reserved Instance offering ID func (c *Client) findOfferingID(ctx context.Context, rec common.Recommendation) (string, error) { details, ok := rec.Details.(*common.DatabaseDetails) diff --git a/providers/aws/services/rds/client_test.go b/providers/aws/services/rds/client_test.go index ffc815a95..69f96d32d 100644 --- a/providers/aws/services/rds/client_test.go +++ b/providers/aws/services/rds/client_test.go @@ -577,3 +577,140 @@ func TestCreatePurchaseTags_OmitsPurchaseAutomationWhenSourceEmpty(t *testing.T) assert.NotEqual(t, common.PurchaseTagKey, aws.ToString(tag.Key), "tag must be skipped when source is empty") } } + +// idempotencyTestRec is a minimal RDS recommendation whose offering resolves to +// "offering-1" via the mock below. +func idempotencyTestRec() common.Recommendation { + return common.Recommendation{ + Service: common.ServiceRelationalDB, + ResourceType: "db.r6g.large", + Count: 1, + PaymentOption: "all-upfront", + Term: "1yr", + Details: &common.DatabaseDetails{ + Engine: "mysql", + AZConfig: "single-az", + }, + } +} + +func expectOffering(m *MockRDSClient) { + m.On("DescribeReservedDBInstancesOfferings", mock.Anything, mock.Anything). + Return(&rds.DescribeReservedDBInstancesOfferingsOutput{ + ReservedDBInstancesOfferings: []types.ReservedDBInstancesOffering{ + { + ReservedDBInstancesOfferingId: aws.String("offering-1"), + DBInstanceClass: aws.String("db.r6g.large"), + ProductDescription: aws.String("mysql"), + MultiAZ: aws.Bool(false), + OfferingType: aws.String("All Upfront"), + Duration: aws.Int32(31536000), + }, + }, + }, nil) +} + +// TestClient_PurchaseCommitment_Idempotent_GuardShortCircuits asserts that when a +// reservation already exists under the token-derived ID, a re-drive returns it +// WITHOUT calling PurchaseReservedDBInstancesOffering a second time (issue #641). +func TestClient_PurchaseCommitment_Idempotent_GuardShortCircuits(t *testing.T) { + mockRDS := &MockRDSClient{} + client := &Client{client: mockRDS, region: "eu-west-1"} + + token := common.DeriveIdempotencyToken("exec-1", 0) + derivedID := common.IdempotentReservationID("rds-id-", token) + + expectOffering(mockRDS) + // The by-ID guard finds an existing active reservation under the derived ID. + mockRDS.On("DescribeReservedDBInstances", mock.Anything, mock.MatchedBy(func(in *rds.DescribeReservedDBInstancesInput) bool { + return aws.ToString(in.ReservedDBInstanceId) == derivedID + })).Return(&rds.DescribeReservedDBInstancesOutput{ + ReservedDBInstances: []types.ReservedDBInstance{ + {ReservedDBInstanceId: aws.String(derivedID), State: aws.String("active")}, + }, + }, nil) + + result, err := client.PurchaseCommitment(context.Background(), idempotencyTestRec(), common.PurchaseOptions{IdempotencyToken: token}) + + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, derivedID, result.CommitmentID) + mockRDS.AssertNotCalled(t, "PurchaseReservedDBInstancesOffering", mock.Anything, mock.Anything) +} + +// TestClient_PurchaseCommitment_Idempotent_NotFoundProceeds asserts a first-time +// purchase proceeds: the by-ID guard reports not-found (NotFound fault), the +// purchase runs, and the derived ID is used. +func TestClient_PurchaseCommitment_Idempotent_NotFoundProceeds(t *testing.T) { + mockRDS := &MockRDSClient{} + client := &Client{client: mockRDS, region: "eu-west-1"} + + token := common.DeriveIdempotencyToken("exec-2", 0) + derivedID := common.IdempotentReservationID("rds-id-", token) + + expectOffering(mockRDS) + mockRDS.On("DescribeReservedDBInstances", mock.Anything, mock.Anything). + Return((*rds.DescribeReservedDBInstancesOutput)(nil), &types.ReservedDBInstanceNotFoundFault{}) + mockRDS.On("PurchaseReservedDBInstancesOffering", mock.Anything, mock.MatchedBy(func(in *rds.PurchaseReservedDBInstancesOfferingInput) bool { + return aws.ToString(in.ReservedDBInstanceId) == derivedID + })).Return(&rds.PurchaseReservedDBInstancesOfferingOutput{ + ReservedDBInstance: &types.ReservedDBInstance{ReservedDBInstanceId: aws.String(derivedID)}, + }, nil) + + result, err := client.PurchaseCommitment(context.Background(), idempotencyTestRec(), common.PurchaseOptions{IdempotencyToken: token}) + + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, derivedID, result.CommitmentID) + mockRDS.AssertExpectations(t) +} + +// TestClient_PurchaseCommitment_Idempotent_AlreadyExistsRecovers asserts that if +// the guard missed but AWS rejects the duplicate ID with the AlreadyExists fault, +// the re-drive recovers the existing reservation instead of erroring. +func TestClient_PurchaseCommitment_Idempotent_AlreadyExistsRecovers(t *testing.T) { + mockRDS := &MockRDSClient{} + client := &Client{client: mockRDS, region: "eu-west-1"} + + token := common.DeriveIdempotencyToken("exec-3", 0) + derivedID := common.IdempotentReservationID("rds-id-", token) + + expectOffering(mockRDS) + // First Describe (guard): not found. Second Describe (recovery): found. + mockRDS.On("DescribeReservedDBInstances", mock.Anything, mock.Anything). + Return((*rds.DescribeReservedDBInstancesOutput)(nil), &types.ReservedDBInstanceNotFoundFault{}).Once() + mockRDS.On("PurchaseReservedDBInstancesOffering", mock.Anything, mock.Anything). + Return((*rds.PurchaseReservedDBInstancesOfferingOutput)(nil), &types.ReservedDBInstanceAlreadyExistsFault{}) + mockRDS.On("DescribeReservedDBInstances", mock.Anything, mock.Anything). + Return(&rds.DescribeReservedDBInstancesOutput{ + ReservedDBInstances: []types.ReservedDBInstance{ + {ReservedDBInstanceId: aws.String(derivedID), State: aws.String("active")}, + }, + }, nil).Once() + + result, err := client.PurchaseCommitment(context.Background(), idempotencyTestRec(), common.PurchaseOptions{IdempotencyToken: token}) + + assert.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, derivedID, result.CommitmentID) +} + +// TestClient_PurchaseCommitment_Idempotent_FailLoudOnLookupError asserts a lookup +// error fails loud and does NOT fall through to a purchase (no double-buy). +func TestClient_PurchaseCommitment_Idempotent_FailLoudOnLookupError(t *testing.T) { + mockRDS := &MockRDSClient{} + client := &Client{client: mockRDS, region: "eu-west-1"} + + token := common.DeriveIdempotencyToken("exec-4", 0) + + expectOffering(mockRDS) + mockRDS.On("DescribeReservedDBInstances", mock.Anything, mock.Anything). + Return((*rds.DescribeReservedDBInstancesOutput)(nil), fmt.Errorf("access denied")) + + result, err := client.PurchaseCommitment(context.Background(), idempotencyTestRec(), common.PurchaseOptions{IdempotencyToken: token}) + + assert.Error(t, err) + assert.False(t, result.Success) + assert.Contains(t, err.Error(), "refusing to purchase") + mockRDS.AssertNotCalled(t, "PurchaseReservedDBInstancesOffering", mock.Anything, mock.Anything) +} diff --git a/providers/aws/services/redshift/client.go b/providers/aws/services/redshift/client.go index f61e7176d..3663dfbda 100644 --- a/providers/aws/services/redshift/client.go +++ b/providers/aws/services/redshift/client.go @@ -24,6 +24,7 @@ type RedshiftAPI interface { 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 @@ -151,6 +152,37 @@ 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 recognise 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) + 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.CommitmentID = existingID + return result, nil + } + } + input := &redshift.PurchaseReservedNodeOfferingInput{ ReservedNodeOfferingId: aws.String(offeringID), NodeCount: aws.Int32(int32(rec.Count)), @@ -173,13 +205,98 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, result.Error } - if err := c.tagReservedNode(ctx, result.CommitmentID, rec, opts.Source); err != nil { - log.Printf("WARNING: failed to tag Redshift reserved node %s after purchase (node is bought; tag missing): %v", result.CommitmentID, err) + 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 } +// findNodeByIdempotencyToken looks for an active or payment-pending 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 active 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) (string, bool, 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 { + 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) == "" { + break + } + marker = response.Marker + } + return "", false, nil +} + +// scanNodesForToken checks each active/payment-pending 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) (string, bool, error) { + for _, node := range nodes { + state := aws.ToString(node.State) + if state != "active" && state != "payment-pending" { + 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 @@ -204,10 +321,14 @@ func (c *Client) resolveAccountID(ctx context.Context) (string, error) { // 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 -// source is empty (opt-out) OR when the account ID can't be resolved — both -// mean "don't tag", logged by the caller. -func (c *Client) tagReservedNode(ctx context.Context, nodeID string, rec common.Recommendation, source string) error { - if source == "" { +// 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 +// recognise 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) @@ -225,7 +346,12 @@ func (c *Client) tagReservedNode(ctx context.Context, nodeID string, rec common. {Key: aws.String("Region"), Value: aws.String(rec.Region)}, {Key: aws.String("PurchaseDate"), Value: aws.String(time.Now().Format("2006-01-02"))}, {Key: aws.String("Tool"), Value: aws.String("CUDly")}, - {Key: aws.String(common.PurchaseTagKey), Value: aws.String(source)}, + } + 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} diff --git a/providers/aws/services/redshift/client_test.go b/providers/aws/services/redshift/client_test.go index 93835ba49..79a8b0b73 100644 --- a/providers/aws/services/redshift/client_test.go +++ b/providers/aws/services/redshift/client_test.go @@ -52,6 +52,14 @@ func (m *MockRedshiftClient) CreateTags(ctx context.Context, params *redshift.Cr 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 @@ -935,3 +943,116 @@ func TestClient_FindOfferingID_UnknownOfferingType(t *testing.T) { assert.Contains(t, err.Error(), "no offerings found") mockRS.AssertExpectations(t) } + +func rsIdemRec() common.Recommendation { + return 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}, + } +} + +func expectRSOffering(m *MockRedshiftClient) { + m.On("DescribeReservedNodeOfferings", mock.Anything, mock.Anything). + Return(&redshift.DescribeReservedNodeOfferingsOutput{ + ReservedNodeOfferings: []types.ReservedNodeOffering{ + { + ReservedNodeOfferingId: aws.String("offering-1"), + NodeType: aws.String("ra3.xlplus"), + Duration: aws.Int32(31536000), + ReservedNodeOfferingType: types.ReservedNodeOfferingType("Regular"), + }, + }, + }, 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.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.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.Contains(t, err.Error(), "refusing to purchase") + mockRS.AssertNotCalled(t, "PurchaseReservedNodeOffering", mock.Anything, mock.Anything) +}