refactor(aws): use the shared purchase guard in the OpenSearch purchase path

CHANGELOG.md

@@ -133,6 +133,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/).
 
 ### Changed
 
+- **`aws/opensearch`:** the OpenSearch purchase path uses the shared `purchaseguard` idempotency guard and AlreadyExists recovery, like ElastiCache. Behavior for valid purchases is unchanged (#10).
 - **`common.HoursPerMonth`:** exported 730-hour average month (`HoursPerYear / 12`) next to `HoursPerYear`; the remaining bare 730 and 8760 literals in the aws and gcp providers use the shared constants (#109).
 - **`gcp`:** region-lookup permission errors are now classified by type only (googleapi 403 or gRPC `PermissionDenied`) through the shared `IsPermissionError`. gRPC `PermissionDenied` is now skipped with a warning like a 403, and an untyped error whose text merely contains "403" and "permission" now propagates instead of being skipped (#6).
 - **BREAKING `gcp`: stored contexts removed (#6).** `computeengine.NewClient`

providers/aws/services/opensearch/client.go

@@ -3,7 +3,6 @@ package opensearch
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"log"
 	"math"
@@ -19,6 +18,7 @@ import (
 	"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/purchaseguard"
 	"github.com/LeanerCloud/cloud-commitments-go/providers/aws/internal/reservationexpiry"
 	"github.com/LeanerCloud/cloud-commitments-go/providers/aws/internal/reservationstate"
 )
@@ -291,43 +291,19 @@ func (c *Client) findReservationByName(ctx context.Context, name string) (reserv
 	return "", false, fmt.Errorf("pagination cap reached after %d pages for OpenSearch reservation idempotency lookup", maxReservationPages)
 }
 
-// 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.
+// idempotencyGuard short-circuits a re-drive (issue #641); see purchaseguard.Guard.
 func (c *Client) idempotencyGuard(ctx context.Context, token, reservationName string) (existingID string, found bool, err 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
+	return purchaseguard.Guard(ctx, token, "OpenSearch RI", func(ctx context.Context) (string, bool, error) {
+		return c.findReservationByName(ctx, reservationName)
+	})
 }
 
-// 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.
+// recoverAlreadyExists adopts the existing reservation when AWS rejects a
+// duplicate ID (issue #641); see purchaseguard.Recover.
 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
+	return purchaseguard.Recover[*types.ResourceAlreadyExistsException](ctx, token, "OpenSearch RI", purchaseErr, func(ctx context.Context) (string, bool, error) {
+		return c.findReservationByName(ctx, reservationName)
+	})
 }
 
 // resolveAccountID fetches the caller's AWS account ID via STS and caches it.

providers/aws/services/opensearch/purchase_guard_test.go

@@ -0,0 +1,137 @@
+package opensearch
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/opensearch"
+	"github.com/aws/aws-sdk-go-v2/service/opensearch/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+// The purchaseguard rules (issue #10), asserted at the opensearch purchase path. Each
+// mock is bounded with Test(t): an unexpected AWS call fails the test instead
+// of panicking.
+
+func newGuardMock(t *testing.T) *MockOpenSearchClient {
+	t.Helper()
+	m := &MockOpenSearchClient{}
+	m.Test(t)
+	expectOSOffering(m)
+	return m
+}
+
+func guardClient(m *MockOpenSearchClient) *Client { return &Client{client: m, region: "eu-west-1"} }
+
+// No token: the guard never looks the reservation up and the purchase runs.
+func TestPurchaseGuard_EmptyTokenSkipsLookup(t *testing.T) {
+	t.Parallel()
+	m := newGuardMock(t)
+	m.On("PurchaseReservedInstanceOffering", mock.Anything, mock.Anything).Return(&opensearch.PurchaseReservedInstanceOfferingOutput{ReservedInstanceId: aws.String("os-new")}, nil).Once()
+
+	result, err := guardClient(m).PurchaseCommitment(context.Background(), osIdemRec(), common.PurchaseOptions{})
+
+	require.NoError(t, err)
+	assert.True(t, result.Success)
+	assert.False(t, result.ExistingCommitment)
+	m.AssertNotCalled(t, "DescribeReservedInstances", mock.Anything, mock.Anything)
+	m.AssertNumberOfCalls(t, "PurchaseReservedInstanceOffering", 1)
+}
+
+// With a token, the guard looks up before buying, and the lookup runs strictly
+// before the purchase call.
+func TestPurchaseGuard_LookupPrecedesPurchase(t *testing.T) {
+	t.Parallel()
+	m := newGuardMock(t)
+	var order []string
+	m.On("DescribeReservedInstances", mock.Anything, mock.Anything).
+		Run(func(mock.Arguments) { order = append(order, "lookup") }).
+		Return(&opensearch.DescribeReservedInstancesOutput{}, nil).Once()
+	m.On("PurchaseReservedInstanceOffering", mock.Anything, mock.Anything).
+		Run(func(mock.Arguments) { order = append(order, "purchase") }).
+		Return(&opensearch.PurchaseReservedInstanceOfferingOutput{ReservedInstanceId: aws.String("os-new")}, nil).Once()
+
+	_, err := guardClient(m).PurchaseCommitment(context.Background(), osIdemRec(), common.PurchaseOptions{IdempotencyToken: common.DeriveIdempotencyToken("exec-order", 0)})
+
+	require.NoError(t, err)
+	assert.Equal(t, []string{"lookup", "purchase"}, order)
+}
+
+// A lookup failure refuses the purchase with the shared guard's error text.
+func TestPurchaseGuard_LookupErrorRefusesWithLabel(t *testing.T) {
+	t.Parallel()
+	m := newGuardMock(t)
+	m.On("DescribeReservedInstances", mock.Anything, mock.Anything).Return((*opensearch.DescribeReservedInstancesOutput)(nil), fmt.Errorf("access denied")).Once()
+
+	_, err := guardClient(m).PurchaseCommitment(context.Background(), osIdemRec(), common.PurchaseOptions{IdempotencyToken: common.DeriveIdempotencyToken("exec-label", 0)})
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "idempotency lookup failed before OpenSearch RI purchase (refusing to purchase to avoid a possible double-buy)")
+	m.AssertNotCalled(t, "PurchaseReservedInstanceOffering", mock.Anything, mock.Anything)
+}
+
+// AlreadyExists with a token but a failing re-lookup: the original purchase
+// error is reported, never a false success.
+func TestPurchaseGuard_AlreadyExistsRelookupErrorReportsPurchaseError(t *testing.T) {
+	t.Parallel()
+	m := newGuardMock(t)
+	m.On("DescribeReservedInstances", mock.Anything, mock.Anything).Return(&opensearch.DescribeReservedInstancesOutput{}, nil).Once()
+	m.On("PurchaseReservedInstanceOffering", mock.Anything, mock.Anything).Return((*opensearch.PurchaseReservedInstanceOfferingOutput)(nil), &types.ResourceAlreadyExistsException{}).Once()
+	m.On("DescribeReservedInstances", mock.Anything, mock.Anything).Return((*opensearch.DescribeReservedInstancesOutput)(nil), fmt.Errorf("throttled")).Once()
+
+	result, err := guardClient(m).PurchaseCommitment(context.Background(), osIdemRec(), common.PurchaseOptions{IdempotencyToken: common.DeriveIdempotencyToken("exec-relookup", 0)})
+
+	require.Error(t, err)
+	var already *types.ResourceAlreadyExistsException
+	assert.True(t, errors.As(err, &already), "the original AlreadyExists error must be reported")
+	assert.False(t, result.Success)
+	assert.False(t, result.ExistingCommitment)
+}
+
+// AlreadyExists with a token but a re-lookup that misses: same, no false success.
+func TestPurchaseGuard_AlreadyExistsRelookupMissReportsPurchaseError(t *testing.T) {
+	t.Parallel()
+	m := newGuardMock(t)
+	m.On("DescribeReservedInstances", mock.Anything, mock.Anything).Return(&opensearch.DescribeReservedInstancesOutput{}, nil).Twice()
+	m.On("PurchaseReservedInstanceOffering", mock.Anything, mock.Anything).Return((*opensearch.PurchaseReservedInstanceOfferingOutput)(nil), &types.ResourceAlreadyExistsException{}).Once()
+
+	result, err := guardClient(m).PurchaseCommitment(context.Background(), osIdemRec(), common.PurchaseOptions{IdempotencyToken: common.DeriveIdempotencyToken("exec-miss", 0)})
+
+	require.Error(t, err)
+	assert.False(t, result.Success)
+	assert.False(t, result.ExistingCommitment)
+}
+
+// No token: AlreadyExists is never recovered (and nothing is looked up).
+func TestPurchaseGuard_AlreadyExistsWithoutTokenNotRecovered(t *testing.T) {
+	t.Parallel()
+	m := newGuardMock(t)
+	m.On("PurchaseReservedInstanceOffering", mock.Anything, mock.Anything).Return((*opensearch.PurchaseReservedInstanceOfferingOutput)(nil), &types.ResourceAlreadyExistsException{}).Once()
+
+	result, err := guardClient(m).PurchaseCommitment(context.Background(), osIdemRec(), common.PurchaseOptions{})
+
+	require.Error(t, err)
+	assert.False(t, result.Success)
+	assert.False(t, result.ExistingCommitment)
+	m.AssertNotCalled(t, "DescribeReservedInstances", mock.Anything, mock.Anything)
+}
+
+// A purchase error that is not AlreadyExists triggers no recovery lookup.
+func TestPurchaseGuard_OtherPurchaseErrorNotRecovered(t *testing.T) {
+	t.Parallel()
+	m := newGuardMock(t)
+	m.On("DescribeReservedInstances", mock.Anything, mock.Anything).Return(&opensearch.DescribeReservedInstancesOutput{}, nil).Once()
+	m.On("PurchaseReservedInstanceOffering", mock.Anything, mock.Anything).Return((*opensearch.PurchaseReservedInstanceOfferingOutput)(nil), fmt.Errorf("boom")).Once()
+
+	result, err := guardClient(m).PurchaseCommitment(context.Background(), osIdemRec(), common.PurchaseOptions{IdempotencyToken: common.DeriveIdempotencyToken("exec-other", 0)})
+
+	require.Error(t, err)
+	assert.False(t, result.Success)
+	m.AssertNumberOfCalls(t, "DescribeReservedInstances", 1)
+}