diff --git a/CHANGELOG.md b/CHANGELOG.md index bee8afbf..866981b9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/). ### Added +- **AWS/Redshift:** when the `AttemptJournal` also implements `common.ClaimReader`, the reserved-node dedupe skips nodes another purchase recorded as bought, before and after the send. A claim read error refuses before the send or leaves the outcome unknown after it; if the skip leaves one of several matches and the token has no sending or outcome_unknown record, the purchase is refused as ambiguous (#700). - **`pkg`:** optional `common.ClaimReader` (`ClaimedByOthers(ctx, token, ids)`), which an `AttemptJournal` may implement so a provider can skip commitments another purchase recorded as bought (#700). - **`pkg`:** `common.Commitment.CostKnown` reports whether `Cost` holds a price the provider reported. The zero value means unknown, so a `Cost` of 0 without `CostKnown` must not be read as free. Producers set it in a follow-up (#74). diff --git a/providers/aws/go.mod b/providers/aws/go.mod index 4e090755..e56a301b 100644 --- a/providers/aws/go.mod +++ b/providers/aws/go.mod @@ -3,7 +3,7 @@ module github.com/LeanerCloud/cloud-commitments-go/providers/aws go 1.26.10 require ( - github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261106140354-6b6d481e8298 + github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261106235222-20c9dacb335f github.com/aws/aws-sdk-go-v2 v1.47.5 github.com/aws/aws-sdk-go-v2/config v1.33.13 github.com/aws/aws-sdk-go-v2/credentials v1.20.13 diff --git a/providers/aws/go.sum b/providers/aws/go.sum index a45abbbf..8809cfad 100644 --- a/providers/aws/go.sum +++ b/providers/aws/go.sum @@ -1,5 +1,5 @@ -github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261106140354-6b6d481e8298 h1:BsbLH7s1R9FXHHdih54yw3IDsWcsoaZNEKj9yxtTuf0= -github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261106140354-6b6d481e8298/go.mod h1:l9V+ecJCfAigYr6XiK9kpIUemF9bnXOKHJN1s4TPsM0= +github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261106235222-20c9dacb335f h1:+GzNJalL4vfbk1Lwcu2jIGvJxc7DKScfR2N94+uGbXk= +github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261106235222-20c9dacb335f/go.mod h1:l9V+ecJCfAigYr6XiK9kpIUemF9bnXOKHJN1s4TPsM0= github.com/aws/aws-sdk-go-v2 v1.47.5 h1:5yA+A+BiXbDq1ZLbYwPViwhEGDMdR0MEhL8Jh2LnLS4= github.com/aws/aws-sdk-go-v2 v1.47.5/go.mod h1:iNIkogrRvFwm+1OQIUOP2lPBtlXZboOIUWiCa+pHEMw= github.com/aws/aws-sdk-go-v2/config v1.33.13 h1:u52x6ISQOx172tpwM0VRxhJzEsTJYpMcEYUGM2A1wrw= diff --git a/providers/aws/internal/purchasecfg/reconcile.go b/providers/aws/internal/purchasecfg/reconcile.go index 1c0706c6..5db2f3a4 100644 --- a/providers/aws/internal/purchasecfg/reconcile.go +++ b/providers/aws/internal/purchasecfg/reconcile.go @@ -20,6 +20,7 @@ const ( ReasonNoCandidate = "no-candidate" ReasonAmbiguous = "ambiguous" ReasonUncertainCandidate = "uncertain-candidate" + ReasonClaimLookupFailed = "claim-lookup-failed" ) // ReconcileWindow returns the padded interval a candidate's start time must fall in. diff --git a/providers/aws/internal/purchasecfg/reconcile_test.go b/providers/aws/internal/purchasecfg/reconcile_test.go index 449b7496..eb5213f1 100644 --- a/providers/aws/internal/purchasecfg/reconcile_test.go +++ b/providers/aws/internal/purchasecfg/reconcile_test.go @@ -131,6 +131,7 @@ func TestReasonPhrasesAreDistinct(t *testing.T) { phrases := []string{ ReasonInvalidWindow, ReasonInvalidRequest, ReasonLookupFailed, ReasonLookupIncomplete, ReasonOfferingUnresolved, ReasonNoCandidate, ReasonAmbiguous, ReasonUncertainCandidate, + ReasonClaimLookupFailed, } seen := map[string]bool{} for _, p := range phrases { @@ -138,7 +139,7 @@ func TestReasonPhrasesAreDistinct(t *testing.T) { assert.False(t, seen[p], "duplicate reason phrase %q", p) seen[p] = true } - assert.Len(t, seen, 8) + assert.Len(t, seen, 9) } func TestDecideCopiesCriteria(t *testing.T) { diff --git a/providers/aws/services/redshift/claims_test.go b/providers/aws/services/redshift/claims_test.go new file mode 100644 index 00000000..edb5fac9 --- /dev/null +++ b/providers/aws/services/redshift/claims_test.go @@ -0,0 +1,261 @@ +package redshift + +import ( + "bytes" + "context" + "log" + "os" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/service/redshift/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" +) + +// go#700: the window dedupe skips nodes another token recorded as purchased. +// All tests are fixture-based: mocks and an in-test journal, no AWS calls. + +const jOtherToken = "other-token" + +// seedOther records rn-1 as purchased under a different token. +func (j *fakeJournal) seedOther() { + j.rows = append(j.rows, common.AttemptRecord{IdempotencyToken: jOtherToken, ExecutionID: "exec-other", State: common.AttemptPurchased, ReservationID: "rn-1", OfferingID: jOffer}) +} + +func purchasedStates(j *fakeJournal) []common.AttemptState { + recs := j.recordCalls() + out := make([]common.AttemptState, 0, len(recs)) + for _, r := range recs { + out = append(out, r.State) + } + return out +} + +func TestClaims_WindowSkipsOtherClaim(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seedOther() + c, m := newDedupe(jWindow("rn-1")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.NoError(t, err) + assert.True(t, result.Success) + assert.False(t, result.ExistingCommitment) + assert.Equal(t, "rn-new", result.CommitmentID) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) + assert.Equal(t, []common.AttemptState{common.AttemptSending, common.AttemptPurchased}, purchasedStates(j)) +} + +func TestClaims_SingleRawMatchNoClaimNoPriorsAdopts(t *testing.T) { + t.Parallel() + j := newFakeJournal() + c, m := newDedupe(jWindow("rn-1")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.NoError(t, err) + assert.True(t, result.ExistingCommitment) + assert.Equal(t, "rn-1", result.CommitmentID) + jNoBuy(t, m) + assert.Equal(t, "rn-1", j.row(t, jExecNew).ReservationID) +} + +func TestClaims_ExclusionToOneNeedsOwnSendEvidence(t *testing.T) { + t.Parallel() + t.Run("no priors refuses as ambiguous", func(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seedOther() + c, m := newDedupe(jWindow("rn-1", "rn-2")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.Error(t, err) + assert.False(t, result.Success) + assert.Empty(t, result.CommitmentID) + assert.Contains(t, err.Error(), "ambiguous") + assert.Contains(t, err.Error(), "rn-2") + assert.Contains(t, err.Error(), "recorded by another purchase and excluded") + assert.Contains(t, err.Error(), "no earlier attempt of this token shows a send") + assert.Contains(t, err.Error(), "rn-1") + assert.Empty(t, j.recordCalls()) + jNoBuy(t, m) + }) + t.Run("own prior under another offering refuses as ambiguous", func(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seedOther() + j.seed(jExecOld, common.AttemptSending, "", "offering-0") + c, m := newDedupe(jWindow("rn-1", "rn-2")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.Error(t, err) + assert.False(t, result.Success) + assert.Contains(t, err.Error(), "ambiguous") + assert.Contains(t, err.Error(), "no earlier attempt of this token shows a send") + assert.Empty(t, j.recordCalls()) + jNoBuy(t, m) + }) + for _, state := range []common.AttemptState{common.AttemptSending, common.AttemptOutcomeUnknown} { + t.Run("own "+string(state)+" prior adopts", func(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seedOther() + j.seed(jExecOld, state, "", jOffer) + c, m := newDedupe(jWindow("rn-1", "rn-2")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.NoError(t, err) + assert.True(t, result.ExistingCommitment) + assert.Equal(t, "rn-2", result.CommitmentID) + jNoBuy(t, m) + assert.Equal(t, "rn-2", j.row(t, jExecNew).ReservationID) + }) + } +} + +func TestClaims_ThreeNodesOneClaimedStillAmbiguous(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seedOther() + c, m := newDedupe(jWindow("rn-1", "rn-2", "rn-3")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.Error(t, err) + assert.False(t, result.Success) + assert.Contains(t, err.Error(), "ambiguous") + assert.Contains(t, err.Error(), "rn-2") + assert.Contains(t, err.Error(), "rn-3") + assert.Contains(t, err.Error(), "recorded by another purchase and excluded") + assert.NotContains(t, err.Error(), "shows a send") + assert.Contains(t, err.Error(), "rn-1") + assert.Empty(t, j.recordCalls()) + jNoBuy(t, m) +} + +func TestClaims_ClaimErrorRefusesNoSend(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.claimErr = assert.AnError + c, m := newDedupe(jWindow("rn-1")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrRefusedBeforePurchase) + assert.False(t, result.Success) + assert.Empty(t, j.recordCalls()) + jNoBuy(t, m) +} + +func TestClaims_ZeroIDsSkipsRead(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.claimErr = assert.AnError + c, m := newDedupe(nil) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.NoError(t, err) + assert.True(t, result.Success) + assert.Empty(t, j.claimCalls()) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) +} + +func TestClaims_PostSendSkipsOtherClaim(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seedOther() + c, m := recoverMock(buyFails(transportErr()), dnode("rn-1", "payment-pending", dedupeReqAt.Add(time.Second))) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrOutcomeUnknown) + assert.False(t, result.Success) + assert.Empty(t, result.CommitmentID) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) + assert.Equal(t, common.AttemptOutcomeUnknown, j.row(t, jExecNew).State) + require.Len(t, j.claimCalls(), 1) + assert.Equal(t, jToken, j.claimCalls()[0].Token) +} + +func TestClaims_PostSendSingleUnclaimedAdopts(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seedOther() + c, _ := recoverMock(buyFails(transportErr()), + dnode("rn-1", "payment-pending", dedupeReqAt.Add(time.Second)), + dnode("rn-2", "payment-pending", dedupeReqAt.Add(time.Second))) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, "rn-2", result.CommitmentID) + assert.Equal(t, "rn-2", j.row(t, jExecNew).ReservationID) +} + +func TestClaims_PostSendClaimErrorNoAdopt(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.claimErr = assert.AnError + c, _ := recoverMock(buyFails(transportErr()), dnode("rn-1", "payment-pending", dedupeReqAt.Add(time.Second))) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrOutcomeUnknown) + assert.False(t, result.Success) + assert.Empty(t, result.CommitmentID) + assert.Equal(t, common.AttemptOutcomeUnknown, j.row(t, jExecNew).State) +} + +func TestClaims_SameTokenClaimPassesToken(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seedOther() + c, m := newDedupe(jWindow("rn-1")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.NoError(t, err) + assert.Equal(t, "rn-new", result.CommitmentID) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) + require.Len(t, j.claimCalls(), 1) + assert.Equal(t, jToken, j.claimCalls()[0].Token) + assert.Equal(t, []string{"rn-1"}, j.claimCalls()[0].IDs) +} + +// A purchased row for the node appears under another execution of this token +// after PriorAttempts was read; the unique-index rule must fail the write. +func TestClaims_SameTokenRaceFailsClosed(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.onPrior = func(f *fakeJournal) { + f.rows = append(f.rows, common.AttemptRecord{IdempotencyToken: jToken, ExecutionID: "exec-race", State: common.AttemptPurchased, ReservationID: "rn-1", OfferingID: jOffer}) + f.onPrior = nil + } + c, m := newDedupe(jWindow("rn-1")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrOutcomeUnknown) + assert.False(t, result.Success) + jNoBuy(t, m) +} + +// hiddenReader wraps a journal so only the AttemptJournal methods show. +type hiddenReader struct{ common.AttemptJournal } + +func TestClaims_ReaderOptional(t *testing.T) { + var buf bytes.Buffer + log.SetOutput(&buf) + t.Cleanup(func() { log.SetOutput(os.Stderr) }) + j := newFakeJournal() + c, m := newDedupe(jWindow("rn-1")) + opts := jOpts(j, jExecNew) + opts.AttemptJournal = hiddenReader{j} + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), opts) + require.NoError(t, err) + assert.True(t, result.ExistingCommitment) + assert.Equal(t, "rn-1", result.CommitmentID) + jNoBuy(t, m) + assert.Empty(t, j.claimCalls()) + assert.Equal(t, 1, bytes.Count(buf.Bytes(), []byte("does not implement common.ClaimReader"))) + assert.Contains(t, buf.String(), common.MaskToken(jToken)) + assert.NotContains(t, buf.String(), jToken) +} + +func TestClaims_TokenNoJournalUnchanged(t *testing.T) { + t.Parallel() + opts := dedupeOpts() + c, m := newDedupe([]types.ReservedNode{dnode("rn-1", "active", dedupeReqAt)}) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), opts) + require.NoError(t, err) + assert.Equal(t, "rn-1", result.CommitmentID) + jNoBuy(t, m) + c, m = newDedupe([]types.ReservedNode{dnode("rn-1", "active", dedupeReqAt), dnode("rn-2", "active", dedupeReqAt)}) + _, err = c.PurchaseCommitment(context.Background(), rsIdemRec(), opts) + require.Error(t, err) + assert.NotContains(t, err.Error(), "recorded by another purchase") + jNoBuy(t, m) +} diff --git a/providers/aws/services/redshift/dedupe.go b/providers/aws/services/redshift/dedupe.go index 07c72252..ebfd7c10 100644 --- a/providers/aws/services/redshift/dedupe.go +++ b/providers/aws/services/redshift/dedupe.go @@ -48,7 +48,10 @@ var errRequestedAtRequired = errors.New("RequestedAt is required with an idempot // later node or be refused as ambiguous. A caller that re-dispatches a token // whose earlier attempt may have been sent must first wait long enough for // DescribeReservedNodes to list a node that attempt bought; an AttemptJournal -// (go#107) records the attempts but cannot close that listing lag. +// (go#107) records the attempts but cannot close that listing lag. When the +// journal also implements common.ClaimReader (go#700), nodes another token +// recorded as purchased are skipped; a node whose purchase was never recorded +// stays invisible, so the dispatch rule above still applies. func (c *Client) findMatchingNodes(ctx context.Context, offeringID string, count int, requestedAt time.Time) ([]string, error) { nodes, err := c.describeAllNodes(ctx) if err != nil { @@ -104,32 +107,90 @@ func nodeMatches(node *redshifttypes.ReservedNode, offeringID string, count int, // guardAgainstRepurchase runs before a tokenized purchase. It reports done // when the purchase must not proceed: result then carries either the adopted -// node (Success) or the refusal in result.Error. -func (c *Client) guardAgainstRepurchase(ctx context.Context, result *common.PurchaseResult, offeringID string, count int, opts common.PurchaseOptions) (done bool) { +// node (Success) or the refusal in result.Error. priors are the token's +// journal records, nil without a journal. +func (c *Client) guardAgainstRepurchase(ctx context.Context, result *common.PurchaseResult, offeringID string, count int, opts common.PurchaseOptions, priors []common.AttemptRecord) (done bool) { if opts.RequestedAt.IsZero() { result.Error = errRequestedAtRequired return true } - ids, err := c.findMatchingNodes(ctx, offeringID, count, opts.RequestedAt) - switch { - case err != nil: + raw, err := c.findMatchingNodes(ctx, offeringID, count, opts.RequestedAt) + if err != nil { result.Error = fmt.Errorf("idempotency lookup failed before Redshift purchase (refusing to purchase to avoid a possible double-buy): %w", err) + return true + } + ids, excluded, err := c.unclaimed(ctx, raw, opts) + if err != nil { + refuseBeforePurchase(result, "claim lookup failed before Redshift purchase (refusing to purchase to avoid a possible double-buy): %w", err) + return true + } + switch { case len(ids) > 1: - result.Error = fmt.Errorf("ambiguous Redshift dedupe: %d reserved nodes (%s) match offering %s, node count %d, started since %s; refusing to purchase to avoid a possible double-buy; check the account and, if the commitment is missing, start a new purchase", - len(ids), strings.Join(ids, ", "), offeringID, count, opts.RequestedAt.Add(-redshiftClockSkew).UTC().Format(time.RFC3339)) + result.Error = ambiguousNodes(ids, excluded, false, offeringID, count, opts) + case len(ids) == 1 && len(raw) > 1 && !hasSendEvidence(priors, offeringID): + result.Error = ambiguousNodes(ids, excluded, true, offeringID, count, opts) case len(ids) == 1: log.Printf("Redshift reserved node %s already matches idempotency token %s; skipping purchase", ids[0], common.MaskToken(opts.IdempotencyToken)) adopt(result, ids[0]) - return true default: return false } return true } +// ambiguousNodes is the refusal for a window that cannot name one node. The +// excluded IDs (recorded by another purchase) are named so an operator can +// tell why a smaller set remains; noSend adds that the token shows no send. +func ambiguousNodes(ids, excluded []string, noSend bool, offeringID string, count int, opts common.PurchaseOptions) error { + msg := fmt.Sprintf("ambiguous Redshift dedupe: %d reserved nodes (%s) match offering %s, node count %d, started since %s", + len(ids), strings.Join(ids, ", "), offeringID, count, opts.RequestedAt.Add(-redshiftClockSkew).UTC().Format(time.RFC3339)) + if len(excluded) > 0 { + msg += fmt.Sprintf("; %s recorded by another purchase and excluded", strings.Join(excluded, ", ")) + } + if noSend { + msg += ", and no earlier attempt of this token shows a send" + } + return fmt.Errorf("%s; refusing to purchase to avoid a possible double-buy; check the account and, if the commitment is missing, start a new purchase", msg) +} + +// hasSendEvidence reports whether the token's own records show a send under +// this offering: a sending or outcome_unknown record. A send under another +// offering cannot have bought a node that matches this one. +func hasSendEvidence(priors []common.AttemptRecord, offeringID string) bool { + return slices.ContainsFunc(priors, func(p common.AttemptRecord) bool { + return (p.State == common.AttemptSending || p.State == common.AttemptOutcomeUnknown) && p.OfferingID == offeringID + }) +} + +// unclaimed splits ids into those no other token recorded as purchased and +// those another token did (go#700). It does nothing without a journal, when +// the journal cannot read claims, or when there are no ids. +func (c *Client) unclaimed(ctx context.Context, ids []string, opts common.PurchaseOptions) (kept, excluded []string, err error) { + if len(ids) == 0 { + return ids, nil, nil + } + reader, ok := opts.AttemptJournal.(common.ClaimReader) + if !ok { + return ids, nil, nil + } + claimed, err := reader.ClaimedByOthers(ctx, opts.IdempotencyToken, ids) + if err != nil { + return nil, nil, fmt.Errorf("claim lookup for token %s failed: %w", common.MaskToken(opts.IdempotencyToken), err) + } + for _, id := range ids { + if slices.Contains(claimed, id) { + excluded = append(excluded, id) + } else { + kept = append(kept, id) + } + } + return kept, excluded, nil +} + // adoptAfterUnknownOutcome runs after a purchase error. When the outcome is // unknown and exactly one node matches, the purchase did happen: adopt it. A -// missing or ambiguous match leaves the original error in place. +// missing or ambiguous match, or a failed claim read, leaves the original error +// in place; a node another token recorded as purchased does not match. func (c *Client) adoptAfterUnknownOutcome(ctx context.Context, result *common.PurchaseResult, offeringID string, count int, opts common.PurchaseOptions) bool { if !errors.Is(result.Error, common.ErrOutcomeUnknown) || opts.IdempotencyToken == "" || opts.RequestedAt.IsZero() { return false @@ -137,6 +198,9 @@ func (c *Client) adoptAfterUnknownOutcome(ctx context.Context, result *common.Pu lookupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), recoverLookupTimeout) defer cancel() ids, err := c.findMatchingNodes(lookupCtx, offeringID, count, opts.RequestedAt) + if err == nil { + ids, _, err = c.unclaimed(lookupCtx, ids, opts) + } if err != nil || len(ids) != 1 { log.Printf("Redshift purchase outcome unknown and not recoverable (matches=%d, lookup error=%v)", len(ids), err) return false @@ -166,7 +230,7 @@ func (c *Client) guardBeforePurchase(ctx context.Context, result *common.Purchas case journalActive(opts): return c.journalPreflight(ctx, result, offeringID, count, opts) case opts.IdempotencyToken != "": - return c.guardAgainstRepurchase(ctx, result, offeringID, count, opts) + return c.guardAgainstRepurchase(ctx, result, offeringID, count, opts, nil) default: return false } @@ -196,7 +260,8 @@ func (c *Client) journalPreflight(ctx context.Context, result *common.PurchaseRe c.adoptRecordedNode(ctx, result, ids, offeringID, count, opts) return true } - if c.guardAgainstRepurchase(ctx, result, offeringID, count, opts) { + warnWithoutClaimReader(opts) + if c.guardAgainstRepurchase(ctx, result, offeringID, count, opts, priors) { if result.Success { c.recordPurchased(ctx, result, offeringID, opts) } @@ -205,6 +270,14 @@ func (c *Client) journalPreflight(ctx context.Context, result *common.PurchaseRe return refuseUnresolvedPriors(result, priors, offeringID, opts) } +// warnWithoutClaimReader logs once per purchase that the journal cannot read +// other tokens' claims, so the dedupe runs as it did before go#700. +func warnWithoutClaimReader(opts common.PurchaseOptions) { + if _, ok := opts.AttemptJournal.(common.ClaimReader); !ok { + log.Printf("attempt journal for token %s does not implement common.ClaimReader; nodes recorded by other purchases are not excluded from the Redshift dedupe", common.MaskToken(opts.IdempotencyToken)) + } +} + // recordedReservationIDs returns the distinct reservation IDs of Purchased // records, in first-seen order. func recordedReservationIDs(priors []common.AttemptRecord) []string { diff --git a/providers/aws/services/redshift/journal_test.go b/providers/aws/services/redshift/journal_test.go index 78ed2fb5..07491c69 100644 --- a/providers/aws/services/redshift/journal_test.go +++ b/providers/aws/services/redshift/journal_test.go @@ -32,18 +32,28 @@ var jToken = common.DeriveIdempotencyToken("journal", 0) type journalCall struct { Method string Rec common.AttemptRecord + // Token and IDs are set on ClaimedByOthers calls. + Token string + IDs []string } // fakeJournal enforces the transition table and keeps the first OfferingID of -// each (token, execution) row, as a store must. +// each (token, execution) row, as a store must. It also enforces the platform's +// unique index: a reservation ID is purchased at most once across all rows. +// ClaimedByOthers reads the stored rows, as the platform query does; onPrior +// runs after PriorAttempts reads, to model a race. type fakeJournal struct { mu sync.Mutex rows []common.AttemptRecord calls []journalCall priorErr error failOn map[common.AttemptState]error + claimErr error + onPrior func(*fakeJournal) } +var _ common.ClaimReader = (*fakeJournal)(nil) + func newFakeJournal() *fakeJournal { return &fakeJournal{failOn: map[common.AttemptState]error{}} } @@ -69,9 +79,51 @@ func (f *fakeJournal) PriorAttempts(_ context.Context, token string) ([]common.A out = append(out, r) } } + if f.onPrior != nil { + f.onPrior(f) + } return out, nil } +func (f *fakeJournal) ClaimedByOthers(_ context.Context, token string, ids []string) ([]string, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, journalCall{Method: "ClaimedByOthers", Token: token, IDs: slices.Clone(ids)}) + if f.claimErr != nil { + return nil, f.claimErr + } + var out []string + for _, id := range ids { + if slices.ContainsFunc(f.rows, func(r common.AttemptRecord) bool { + return r.State == common.AttemptPurchased && r.ReservationID == id && r.IdempotencyToken != token + }) { + out = append(out, id) + } + } + return out, nil +} + +func (f *fakeJournal) claimCalls() []journalCall { + var out []journalCall + for _, c := range f.calls { + if c.Method == "ClaimedByOthers" { + out = append(out, c) + } + } + return out +} + +// purchasedElsewhere reports whether a row other than index self holds id as +// purchased, whatever its token or execution. +func (f *fakeJournal) purchasedElsewhere(self int, id string) bool { + for i, r := range f.rows { + if i != self && r.State == common.AttemptPurchased && r.ReservationID == id { + return true + } + } + return false +} + func (f *fakeJournal) RecordAttempt(ctx context.Context, rec common.AttemptRecord) error { f.mu.Lock() defer f.mu.Unlock() @@ -93,6 +145,9 @@ func (f *fakeJournal) RecordAttempt(ctx context.Context, rec common.AttemptRecor if !validTransition(from, rec) { return errors.New("fake journal: invalid transition") } + if rec.State == common.AttemptPurchased && f.purchasedElsewhere(i, rec.ReservationID) { + return errors.New("fake journal: reservation already recorded as purchased") + } if i < 0 { f.rows = append(f.rows, rec) return nil @@ -416,6 +471,22 @@ func TestFakeJournalEnforcesTable(t *testing.T) { require.Error(t, j.RecordAttempt(ctx, rec(common.AttemptPurchased, "rn-2"))) } +func TestFakeJournalPurchasedUnique(t *testing.T) { + t.Parallel() + ctx := context.Background() + rec := func(token, exec string, s common.AttemptState, id string) common.AttemptRecord { + return common.AttemptRecord{IdempotencyToken: token, ExecutionID: exec, State: s, ReservationID: id} + } + j := newFakeJournal() + require.NoError(t, j.RecordAttempt(ctx, rec(jToken, "e1", common.AttemptPurchased, "rn-1"))) + require.NoError(t, j.RecordAttempt(ctx, rec(jToken, "e1", common.AttemptPurchased, "rn-1")), "same row re-record") + require.Error(t, j.RecordAttempt(ctx, rec(jToken, "e2", common.AttemptPurchased, "rn-1")), "other execution") + require.Error(t, j.RecordAttempt(ctx, rec("other-token", "e3", common.AttemptPurchased, "rn-1")), "other token") + require.NoError(t, j.RecordAttempt(ctx, rec("other-token", "e3", common.AttemptPurchased, "rn-2")), "different ID") + require.NoError(t, j.RecordAttempt(ctx, rec(jToken, "e4", common.AttemptSending, "")), "sending holds no ID") + require.Error(t, j.RecordAttempt(ctx, rec(jToken, "e4", common.AttemptPurchased, "rn-2")), "sending row to taken ID") +} + func TestJournal_CancelledCtxAfterSendStillRecords(t *testing.T) { t.Parallel() j := newFakeJournal() diff --git a/providers/aws/services/redshift/reconcile.go b/providers/aws/services/redshift/reconcile.go index fca6c1fd..3ff0ea5c 100644 --- a/providers/aws/services/redshift/reconcile.go +++ b/providers/aws/services/redshift/reconcile.go @@ -6,6 +6,7 @@ import ( "time" "github.com/aws/aws-sdk-go-v2/aws" + redshifttypes "github.com/aws/aws-sdk-go-v2/service/redshift/types" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" "github.com/LeanerCloud/cloud-commitments-go/providers/aws/internal/offeringprice" @@ -59,10 +60,34 @@ func (c *Client) ReconcilePurchase(ctx context.Context, req common.ReconcileRequ return purchasecfg.UnknownResult(scanFailure(err)) } - var matches, uncertain []string + matches, uncertain := classifyNodes(nodes, offeringID, int(count), rec.ResourceType, months, lo, hi) + matches, uncertain, ok := c.reconcileClaims(ctx, req, matches, uncertain) + if !ok { + return purchasecfg.UnknownResult(purchasecfg.ReasonClaimLookupFailed) + } + // Tagged is EC2-only: Redshift reserved nodes cannot carry the idempotency tag, so it is always false. + return purchasecfg.Decide(matches, uncertain, reconcileCriteria, false) +} + +// reconcileClaims moves a match another token recorded as purchased into uncertain: Found there +// could hide a failed purchase (go#711). It does nothing without a token; ok is false when the +// claim read failed. +func (c *Client) reconcileClaims(ctx context.Context, req common.ReconcileRequest, matches, uncertain []string) (m, u []string, ok bool) { + if req.Options.IdempotencyToken == "" { + return matches, uncertain, true + } + kept, claimed, err := c.unclaimed(ctx, matches, req.Options) + if err != nil { + return nil, nil, false + } + return kept, append(uncertain, claimed...), true +} + +// classifyNodes splits the scanned nodes into full matches and uncertain candidates. +func classifyNodes(nodes []redshifttypes.ReservedNode, offeringID string, count int, nodeType string, months int, lo, hi time.Time) (matches, uncertain []string) { for i := range nodes { id := aws.ToString(nodes[i].ReservedNodeId) - switch nodeFit(&nodes[i], offeringID, int(count), rec.ResourceType, months, lo, hi) { + switch nodeFit(&nodes[i], offeringID, count, nodeType, months, lo, hi) { case fitMatch: matches = append(matches, id) case fitUncertain: @@ -70,6 +95,5 @@ func (c *Client) ReconcilePurchase(ctx context.Context, req common.ReconcileRequ case fitNo: } } - // Tagged is EC2-only: Redshift reserved nodes cannot carry the idempotency tag, so it is always false. - return purchasecfg.Decide(matches, uncertain, reconcileCriteria, false) + return matches, uncertain } diff --git a/providers/aws/services/redshift/reconcile_claims_test.go b/providers/aws/services/redshift/reconcile_claims_test.go new file mode 100644 index 00000000..9cc303e8 --- /dev/null +++ b/providers/aws/services/redshift/reconcile_claims_test.go @@ -0,0 +1,92 @@ +package redshift + +import ( + "context" + "testing" + + "github.com/aws/aws-sdk-go-v2/service/redshift/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-go/providers/aws/internal/purchasecfg" +) + +// go#711: reconcile never reports Found on a node another token recorded as purchased. + +func claimReq(j common.AttemptJournal) common.ReconcileRequest { + return reconcileReq(func(r *common.ReconcileRequest) { + r.Options.IdempotencyToken = jToken + r.Options.AttemptJournal = j + }) +} + +func reconcileWith(t *testing.T, j common.AttemptJournal, nodes ...types.ReservedNode) common.ReconcileResult { + t.Helper() + c, _ := reconcileFake{pages: [][]types.ReservedNode{nodes}}.client(t) + return c.ReconcilePurchase(context.Background(), claimReq(j)) +} + +func TestReconcileClaims_ClaimedWindowNodeIsUnknown(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seedOther() + got := reconcileWith(t, j, reconcileNode("rn-1")) + assert.Equal(t, common.ReconcileUnknown, got.Verdict) + assert.Equal(t, purchasecfg.ReasonUncertainCandidate, got.Reason) + assert.Equal(t, []string{"rn-1"}, got.CandidateIDs) + require.Len(t, j.claimCalls(), 1) +} + +func TestReconcileClaims_ClaimedPlusUnclaimedIsUnknown(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seedOther() + got := reconcileWith(t, j, reconcileNode("rn-1"), reconcileNode("rn-2")) + assert.Equal(t, common.ReconcileUnknown, got.Verdict) + assert.Equal(t, purchasecfg.ReasonUncertainCandidate, got.Reason) + assert.ElementsMatch(t, []string{"rn-1", "rn-2"}, got.CandidateIDs) +} + +func TestReconcileClaims_ReadErrorIsUnknown(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.claimErr = assert.AnError + got := reconcileWith(t, j, reconcileNode("rn-1")) + assert.Equal(t, common.ReconcileUnknown, got.Verdict) + assert.Equal(t, purchasecfg.ReasonClaimLookupFailed, got.Reason) +} + +func TestReconcileClaims_NoReaderFound(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seedOther() + got := reconcileWith(t, hiddenReader{j}, reconcileNode("rn-1")) + assert.Equal(t, common.ReconcileFound, got.Verdict) + assert.Equal(t, "rn-1", got.CommitmentID) + assert.Empty(t, j.claimCalls()) +} + +func TestReconcileClaims_NoJournalOrTokenFound(t *testing.T) { + t.Parallel() + c, _ := reconcileFake{pages: [][]types.ReservedNode{{reconcileNode("rn-1")}}}.client(t) + assert.Equal(t, common.ReconcileFound, c.ReconcilePurchase(context.Background(), reconcileReq()).Verdict) + j := newFakeJournal() + j.seedOther() + c, _ = reconcileFake{pages: [][]types.ReservedNode{{reconcileNode("rn-1")}}}.client(t) + req := claimReq(j) + req.Options.IdempotencyToken = "" + assert.Equal(t, common.ReconcileFound, c.ReconcilePurchase(context.Background(), req).Verdict) + assert.Empty(t, j.claimCalls()) +} + +func TestReconcileClaims_SameTokenClaimIsNotOther(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seed(jExecOld, common.AttemptPurchased, "rn-1", jOffer) + got := reconcileWith(t, j, reconcileNode("rn-1")) + assert.Equal(t, common.ReconcileFound, got.Verdict) + assert.Equal(t, "rn-1", got.CommitmentID) + require.Len(t, j.claimCalls(), 1) + assert.Equal(t, jToken, j.claimCalls()[0].Token) +}