refactor(aws): share the purchase idempotency guard, starting with ElastiCache

providers/aws/internal/purchaseguard/guard.go

@@ -0,0 +1,57 @@
+// Package purchaseguard holds the purchase idempotency guard shared by the AWS
+// services whose reservations are keyed by a caller-supplied ID (issue #641):
+// look the reservation up before buying, and recover when AWS rejects the
+// duplicate ID.
+package purchaseguard
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+)
+
+// Finder looks up a nonterminal reservation by the ID or name being purchased.
+type Finder func(ctx context.Context) (id string, found bool, err error)
+
+// Guard short-circuits a re-drive (issue #641): with a token set it reports
+// (existingID, true, nil) if find locates the reservation, ("", false, nil) for
+// a first-time purchase, or a fail-loud error on lookup failure. With an empty
+// token it is a no-op and never calls find.
+func Guard(ctx context.Context, token, label string, find Finder) (existingID string, found bool, err error) {
+	if token == "" {
+		return "", false, nil
+	}
+	existingID, found, lookupErr := find(ctx)
+	if lookupErr != nil {
+		return "", false, fmt.Errorf("idempotency lookup failed before %s purchase (refusing to purchase to avoid a possible double-buy): %w", label, lookupErr)
+	}
+	if found {
+		log.Printf("%s for idempotency token %s already exists (%s); skipping purchase (issue #641 re-drive)", label, common.MaskToken(token), existingID)
+		return existingID, true, nil
+	}
+	return "", false, nil
+}
+
+// Recover handles the native server-side dedupe backstop (issue #641): if the
+// guard missed the existing reservation but purchaseErr is the service's
+// AlreadyExists fault F (a pointer type, matched with errors.As), it re-looks
+// it up and returns (existingID, true). A lookup error or a miss returns
+// ("", false) so the caller reports the original purchase error.
+func Recover[F error](ctx context.Context, token, label string, purchaseErr error, find Finder) (string, bool) {
+	if token == "" {
+		return "", false
+	}
+	var already F
+	if !errors.As(purchaseErr, &already) {
+		return "", false
+	}
+	existingID, found, lookupErr := find(ctx)
+	if lookupErr == nil && found {
+		log.Printf("%s %s already existed at purchase time; treating as idempotent re-drive (issue #641)", label, existingID)
+		return existingID, true
+	}
+	return "", false
+}

providers/aws/internal/purchaseguard/guard_test.go

@@ -0,0 +1,142 @@
+package purchaseguard
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type fakeFault struct{}
+
+func (*fakeFault) Error() string { return "fake already exists" }
+
+type otherFault struct{}
+
+func (*otherFault) Error() string { return "other fault" }
+
+const (
+	token  = "tok-0123456789abcdef"
+	masked = "tok-0123..."
+)
+
+func captureLog(t *testing.T) *bytes.Buffer {
+	t.Helper()
+	var buf bytes.Buffer
+	prev := log.Writer()
+	log.SetOutput(&buf)
+	t.Cleanup(func() { log.SetOutput(prev) })
+	return &buf
+}
+
+type finder struct {
+	calls int
+	id    string
+	found bool
+	err   error
+}
+
+func (f *finder) find(context.Context) (string, bool, error) {
+	f.calls++
+	return f.id, f.found, f.err
+}
+
+func TestGuard_EmptyTokenSkipsLookup(t *testing.T) {
+	f := &finder{id: "x", found: true}
+	id, found, err := Guard(context.Background(), "", "RDS RI", f.find)
+	require.NoError(t, err)
+	assert.False(t, found)
+	assert.Empty(t, id)
+	assert.Zero(t, f.calls)
+}
+
+func TestGuard_LookupErrorFailsLoud(t *testing.T) {
+	cause := errors.New("describe boom")
+	f := &finder{err: cause}
+	id, found, err := Guard(context.Background(), token, "ElastiCache reservation", f.find)
+	require.Error(t, err)
+	assert.False(t, found)
+	assert.Empty(t, id)
+	assert.ErrorIs(t, err, cause)
+	assert.Equal(t, "idempotency lookup failed before ElastiCache reservation purchase (refusing to purchase to avoid a possible double-buy): describe boom", err.Error())
+}
+
+func TestGuard_Found(t *testing.T) {
+	buf := captureLog(t)
+	f := &finder{id: "res-1", found: true}
+	id, found, err := Guard(context.Background(), token, "RDS RI", f.find)
+	require.NoError(t, err)
+	assert.True(t, found)
+	assert.Equal(t, "res-1", id)
+	assert.Equal(t, 1, f.calls)
+	assert.Contains(t, buf.String(), "RDS RI for idempotency token "+masked+" already exists (res-1); skipping purchase (issue #641 re-drive)\n")
+	assert.NotContains(t, buf.String(), token)
+}
+
+func TestGuard_NotFound(t *testing.T) {
+	buf := captureLog(t)
+	f := &finder{}
+	id, found, err := Guard(context.Background(), token, "RDS RI", f.find)
+	require.NoError(t, err)
+	assert.False(t, found)
+	assert.Empty(t, id)
+	assert.Equal(t, 1, f.calls)
+	assert.Empty(t, buf.String())
+}
+
+func TestRecover_EmptyTokenSkipsLookup(t *testing.T) {
+	f := &finder{id: "x", found: true}
+	id, ok := Recover[*fakeFault](context.Background(), "", "RDS RI", &fakeFault{}, f.find)
+	assert.False(t, ok)
+	assert.Empty(t, id)
+	assert.Zero(t, f.calls)
+}
+
+func TestRecover_NonMatchingErrorSkipsLookup(t *testing.T) {
+	f := &finder{id: "x", found: true}
+	for _, e := range []error{&otherFault{}, errors.New("plain"), nil} {
+		id, ok := Recover[*fakeFault](context.Background(), token, "RDS RI", e, f.find)
+		assert.False(t, ok)
+		assert.Empty(t, id)
+	}
+	assert.Zero(t, f.calls)
+}
+
+func TestRecover_AdoptsOnAlreadyExists(t *testing.T) {
+	for name, perr := range map[string]error{
+		"direct":  &fakeFault{},
+		"wrapped": fmt.Errorf("purchase failed: %w", &fakeFault{}),
+	} {
+		t.Run(name, func(t *testing.T) {
+			buf := captureLog(t)
+			f := &finder{id: "res-2", found: true}
+			id, ok := Recover[*fakeFault](context.Background(), token, "MemoryDB reservation", perr, f.find)
+			assert.True(t, ok)
+			assert.Equal(t, "res-2", id)
+			assert.Equal(t, 1, f.calls)
+			assert.Contains(t, buf.String(), "MemoryDB reservation res-2 already existed at purchase time; treating as idempotent re-drive (issue #641)\n")
+		})
+	}
+}
+
+func TestRecover_LookupErrorOrMissReturnsNotRecovered(t *testing.T) {
+	for name, f := range map[string]*finder{
+		"lookup error": {id: "res-3", found: true, err: errors.New("describe boom")},
+		"not found":    {},
+	} {
+		t.Run(name, func(t *testing.T) {
+			buf := captureLog(t)
+			id, ok := Recover[*fakeFault](context.Background(), token, "RDS RI", &fakeFault{}, f.find)
+			assert.False(t, ok)
+			assert.Empty(t, id)
+			assert.Equal(t, 1, f.calls)
+			assert.False(t, strings.Contains(buf.String(), "already existed"))
+		})
+	}
+}

providers/aws/services/elasticache/client.go

@@ -16,6 +16,7 @@ import (
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
 	"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"
 	"github.com/LeanerCloud/cloud-commitments-go/providers/aws/internal/tagging"
@@ -234,43 +235,19 @@ func (c *Client) findReservationByID(ctx context.Context, reservationID string)
 	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.
+// idempotencyGuard short-circuits a re-drive (issue #641); see purchaseguard.Guard.
 func (c *Client) idempotencyGuard(ctx context.Context, token, reservationID string) (existingID string, found bool, err 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
+	return purchaseguard.Guard(ctx, token, "ElastiCache reservation", func(ctx context.Context) (string, bool, error) {
+		return c.findReservationByID(ctx, reservationID)
+	})
 }
 
-// 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.
+// recoverAlreadyExists adopts the existing reservation when AWS rejects a
+// duplicate ID (issue #641); see purchaseguard.Recover.
 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
+	return purchaseguard.Recover[*types.ReservedCacheNodeAlreadyExistsFault](ctx, token, "ElastiCache reservation", purchaseErr, func(ctx context.Context) (string, bool, error) {
+		return c.findReservationByID(ctx, reservationID)
+	})
 }
 
 // maxOfferingPages is the maximum number of DescribeReservedCacheNodesOfferings