From 274914568be4c353d89dd8282bb7b00429e24906 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 23:24:24 +0100 Subject: [PATCH 1/5] test(redshift): add claim reads and purchased uniqueness to the fake journal The fake journal now implements common.ClaimReader, logs each call, and rejects a purchased write when any other row already holds the reservation ID as purchased, as the platform's unique index will. Part of #700 Co-Authored-By: Claude Opus 5.5 --- .../aws/services/redshift/journal_test.go | 74 ++++++++++++++++++- 1 file changed, 72 insertions(+), 2 deletions(-) diff --git a/providers/aws/services/redshift/journal_test.go b/providers/aws/services/redshift/journal_test.go index 78ed2fb5..db6fedb8 100644 --- a/providers/aws/services/redshift/journal_test.go +++ b/providers/aws/services/redshift/journal_test.go @@ -32,20 +32,31 @@ 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. +// claims maps a reservation ID to the token that recorded it as purchased, for +// ClaimedByOthers; 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 + claims map[string]string + claimErr error + onPrior func(*fakeJournal) } +var _ common.ClaimReader = (*fakeJournal)(nil) + func newFakeJournal() *fakeJournal { - return &fakeJournal{failOn: map[common.AttemptState]error{}} + return &fakeJournal{failOn: map[common.AttemptState]error{}, claims: map[string]string{}} } func (f *fakeJournal) find(token, exec string) int { @@ -69,9 +80,49 @@ 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 owner, ok := f.claims[id]; ok && owner != 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 +144,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 +470,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() From 1dcab0073911541281a79f516437222d462577e2 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 23:24:45 +0100 Subject: [PATCH 2/5] fix(redshift): skip nodes another purchase recorded as bought When the attempt journal also implements common.ClaimReader, the window dedupe drops reserved nodes another token recorded as purchased, both before the send and when recovering from an unknown outcome. A claim read error refuses before the send and leaves the outcome unknown after it. If the skip reduces several matches to one, the node is adopted only when the token's own records show a send; otherwise the purchase is refused as ambiguous and the refusal names the excluded IDs. A journal without the reader logs one warning per purchase and behaves as before. Fixes #700 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + .../aws/services/redshift/claims_test.go | 242 ++++++++++++++++++ providers/aws/services/redshift/dedupe.go | 93 ++++++- 3 files changed, 324 insertions(+), 12 deletions(-) create mode 100644 providers/aws/services/redshift/claims_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b84ea95..a1700860 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,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/services/redshift/claims_test.go b/providers/aws/services/redshift/claims_test.go new file mode 100644 index 00000000..a8bc4e08 --- /dev/null +++ b/providers/aws/services/redshift/claims_test.go @@ -0,0 +1,242 @@ +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" + +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.claims["rn-1"] = jOtherToken + 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.claims["rn-1"] = jOtherToken + 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") + assert.Contains(t, err.Error(), "rn-1") + 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.claims["rn-1"] = jOtherToken + 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.claims["rn-1"] = jOtherToken + 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") + 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.claims["rn-1"] = jOtherToken + 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.claims["rn-1"] = jOtherToken + 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.claims["rn-1"] = jToken + c, m := newDedupe(jWindow("rn-1")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.NoError(t, err) + assert.Equal(t, "rn-1", result.CommitmentID) + assert.True(t, result.ExistingCommitment) + jNoBuy(t, m) + 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() + j.claims["rn-1"] = jOtherToken + 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 de3a0008..229ffe24 100644 --- a/providers/aws/services/redshift/dedupe.go +++ b/providers/aws/services/redshift/dedupe.go @@ -45,7 +45,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) { since := requestedAt.Add(-redshiftClockSkew) var ids []string @@ -89,32 +92,86 @@ 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, offeringID, count, opts) + case len(ids) == 1 && len(raw) > 1 && !hasSendEvidence(priors): + result.Error = ambiguousNodes(ids, excluded, 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. +func ambiguousNodes(ids, excluded []string, 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, and no earlier attempt of this token shows a send", strings.Join(excluded, ", ")) + } + 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: a +// sending or outcome_unknown record. +func hasSendEvidence(priors []common.AttemptRecord) bool { + return slices.ContainsFunc(priors, func(p common.AttemptRecord) bool { + return p.State == common.AttemptSending || p.State == common.AttemptOutcomeUnknown + }) +} + +// 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 @@ -122,6 +179,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 @@ -151,7 +211,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 } @@ -181,7 +241,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) } @@ -190,6 +251,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 { From 049ce03dd3c01c6aef9e4f86722f7019ebaff3ea Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 23:37:00 +0100 Subject: [PATCH 3/5] fix(redshift): require send evidence under the current offering Adopting the single node left after a claim exclusion now needs a sending or outcome_unknown record under the current offering, as in refuseUnresolvedPriors. The no-send note appears only on that refusal. The fake journal derives claims from its purchased rows, like the platform query. Part of #700 Co-Authored-By: Claude Opus 5.5 --- .../aws/services/redshift/claims_test.go | 45 +++++++++++++------ providers/aws/services/redshift/dedupe.go | 24 +++++----- .../aws/services/redshift/journal_test.go | 11 ++--- 3 files changed, 52 insertions(+), 28 deletions(-) diff --git a/providers/aws/services/redshift/claims_test.go b/providers/aws/services/redshift/claims_test.go index a8bc4e08..edb5fac9 100644 --- a/providers/aws/services/redshift/claims_test.go +++ b/providers/aws/services/redshift/claims_test.go @@ -20,6 +20,11 @@ import ( 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)) @@ -32,7 +37,7 @@ func purchasedStates(j *fakeJournal) []common.AttemptState { func TestClaims_WindowSkipsOtherClaim(t *testing.T) { t.Parallel() j := newFakeJournal() - j.claims["rn-1"] = jOtherToken + j.seedOther() c, m := newDedupe(jWindow("rn-1")) result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) require.NoError(t, err) @@ -60,7 +65,7 @@ func TestClaims_ExclusionToOneNeedsOwnSendEvidence(t *testing.T) { t.Run("no priors refuses as ambiguous", func(t *testing.T) { t.Parallel() j := newFakeJournal() - j.claims["rn-1"] = jOtherToken + j.seedOther() c, m := newDedupe(jWindow("rn-1", "rn-2")) result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) require.Error(t, err) @@ -68,16 +73,31 @@ func TestClaims_ExclusionToOneNeedsOwnSendEvidence(t *testing.T) { 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") + 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.claims["rn-1"] = jOtherToken + 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)) @@ -93,7 +113,7 @@ func TestClaims_ExclusionToOneNeedsOwnSendEvidence(t *testing.T) { func TestClaims_ThreeNodesOneClaimedStillAmbiguous(t *testing.T) { t.Parallel() j := newFakeJournal() - j.claims["rn-1"] = jOtherToken + 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) @@ -101,7 +121,8 @@ func TestClaims_ThreeNodesOneClaimedStillAmbiguous(t *testing.T) { 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") + 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) @@ -134,7 +155,7 @@ func TestClaims_ZeroIDsSkipsRead(t *testing.T) { func TestClaims_PostSendSkipsOtherClaim(t *testing.T) { t.Parallel() j := newFakeJournal() - j.claims["rn-1"] = jOtherToken + 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) @@ -149,7 +170,7 @@ func TestClaims_PostSendSkipsOtherClaim(t *testing.T) { func TestClaims_PostSendSingleUnclaimedAdopts(t *testing.T) { t.Parallel() j := newFakeJournal() - j.claims["rn-1"] = jOtherToken + j.seedOther() c, _ := recoverMock(buyFails(transportErr()), dnode("rn-1", "payment-pending", dedupeReqAt.Add(time.Second)), dnode("rn-2", "payment-pending", dedupeReqAt.Add(time.Second))) @@ -175,13 +196,12 @@ func TestClaims_PostSendClaimErrorNoAdopt(t *testing.T) { func TestClaims_SameTokenClaimPassesToken(t *testing.T) { t.Parallel() j := newFakeJournal() - j.claims["rn-1"] = jToken + 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-1", result.CommitmentID) - assert.True(t, result.ExistingCommitment) - jNoBuy(t, m) + 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) @@ -211,7 +231,6 @@ func TestClaims_ReaderOptional(t *testing.T) { log.SetOutput(&buf) t.Cleanup(func() { log.SetOutput(os.Stderr) }) j := newFakeJournal() - j.claims["rn-1"] = jOtherToken c, m := newDedupe(jWindow("rn-1")) opts := jOpts(j, jExecNew) opts.AttemptJournal = hiddenReader{j} diff --git a/providers/aws/services/redshift/dedupe.go b/providers/aws/services/redshift/dedupe.go index 229ffe24..d56f21bf 100644 --- a/providers/aws/services/redshift/dedupe.go +++ b/providers/aws/services/redshift/dedupe.go @@ -111,9 +111,9 @@ func (c *Client) guardAgainstRepurchase(ctx context.Context, result *common.Purc } switch { case len(ids) > 1: - result.Error = ambiguousNodes(ids, excluded, offeringID, count, opts) - case len(ids) == 1 && len(raw) > 1 && !hasSendEvidence(priors): - result.Error = ambiguousNodes(ids, excluded, offeringID, count, opts) + 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]) @@ -125,21 +125,25 @@ func (c *Client) guardAgainstRepurchase(ctx context.Context, result *common.Purc // 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. -func ambiguousNodes(ids, excluded []string, offeringID string, count int, opts common.PurchaseOptions) error { +// 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, and no earlier attempt of this token shows a send", strings.Join(excluded, ", ")) + 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: a -// sending or outcome_unknown record. -func hasSendEvidence(priors []common.AttemptRecord) bool { +// 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 + return (p.State == common.AttemptSending || p.State == common.AttemptOutcomeUnknown) && p.OfferingID == offeringID }) } diff --git a/providers/aws/services/redshift/journal_test.go b/providers/aws/services/redshift/journal_test.go index db6fedb8..07491c69 100644 --- a/providers/aws/services/redshift/journal_test.go +++ b/providers/aws/services/redshift/journal_test.go @@ -40,15 +40,14 @@ type journalCall struct { // fakeJournal enforces the transition table and keeps the first OfferingID of // 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. -// claims maps a reservation ID to the token that recorded it as purchased, for -// ClaimedByOthers; onPrior runs after PriorAttempts reads, to model a race. +// 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 - claims map[string]string claimErr error onPrior func(*fakeJournal) } @@ -56,7 +55,7 @@ type fakeJournal struct { var _ common.ClaimReader = (*fakeJournal)(nil) func newFakeJournal() *fakeJournal { - return &fakeJournal{failOn: map[common.AttemptState]error{}, claims: map[string]string{}} + return &fakeJournal{failOn: map[common.AttemptState]error{}} } func (f *fakeJournal) find(token, exec string) int { @@ -95,7 +94,9 @@ func (f *fakeJournal) ClaimedByOthers(_ context.Context, token string, ids []str } var out []string for _, id := range ids { - if owner, ok := f.claims[id]; ok && owner != token { + 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) } } From 13f0b4d857fa49282af2b78f6a9297ff1427bee4 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Sat, 7 Nov 2026 00:54:04 +0100 Subject: [PATCH 4/5] chore(aws): bump pkg to the claim reader contract Part of #700 Co-Authored-By: Claude Opus 5.5 --- providers/aws/go.mod | 2 +- providers/aws/go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) 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= From b6641e4378fa9ca9af9b0b7fce85acf970eaf4a8 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Sat, 7 Nov 2026 03:09:14 +0100 Subject: [PATCH 5/5] fix(redshift): treat a node another purchase claimed as uncertain when reconciling When the attempt journal can read claims, a matching node that another token recorded as purchased makes the reconcile verdict Unknown instead of Found. A claim read error is Unknown with its own reason. Fixes #711 Co-Authored-By: Claude Opus 5.5 --- .../aws/internal/purchasecfg/reconcile.go | 1 + .../internal/purchasecfg/reconcile_test.go | 3 +- providers/aws/services/redshift/reconcile.go | 32 ++++++- .../redshift/reconcile_claims_test.go | 92 +++++++++++++++++++ 4 files changed, 123 insertions(+), 5 deletions(-) create mode 100644 providers/aws/services/redshift/reconcile_claims_test.go 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/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) +}