diff --git a/providers/aws/internal/purchaseguard/guard.go b/providers/aws/internal/purchaseguard/guard.go new file mode 100644 index 00000000..29d2f659 --- /dev/null +++ b/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 +} diff --git a/providers/aws/internal/purchaseguard/guard_test.go b/providers/aws/internal/purchaseguard/guard_test.go new file mode 100644 index 00000000..843ae989 --- /dev/null +++ b/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")) + }) + } +} diff --git a/providers/aws/services/elasticache/client.go b/providers/aws/services/elasticache/client.go index b9b31403..ffa941aa 100644 --- a/providers/aws/services/elasticache/client.go +++ b/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