diff --git a/providers/aws/go.mod b/providers/aws/go.mod index 9b096a83..4e090755 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-20261104063122-3efa339480ac + github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261106140354-6b6d481e8298 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 bc8ce553..a45abbbf 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-20261104063122-3efa339480ac h1:EvEUCt1uH5oBtPcJJPt1mX6XYQWAKtL9QBSo9CfeaOw= -github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261104063122-3efa339480ac/go.mod h1:pseCFRgm1rzBKomktjw6E0ayxmVE7XfHHiOmew4ka+g= +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/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/services/redshift/client.go b/providers/aws/services/redshift/client.go index 45c9a8a9..556c81a1 100644 --- a/providers/aws/services/redshift/client.go +++ b/providers/aws/services/redshift/client.go @@ -125,6 +125,12 @@ func (c *Client) GetExistingCommitments(ctx context.Context) ([]common.Commitmen // The tokenized path relies on the caller contract documented on // findMatchingNodes: no concurrent same-spec purchase for the account and region. // +// With an IdempotencyToken and PurchaseOptions.AttemptJournal (go#107) the +// attempt is also recorded: sending before the call, then purchased or +// outcome_unknown. A node recorded as purchased by an earlier attempt is adopted +// without the window, and an unresolved earlier attempt that the window cannot +// resolve refuses. The window dedupe alone applies when no journal is set. +// // Reserved nodes cannot be tagged, so the purchased node is not tagged. func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendation, opts common.PurchaseOptions) (common.PurchaseResult, error) { result := common.PurchaseResult{ @@ -148,16 +154,13 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, result.Error } - if opts.IdempotencyToken != "" { - if done := c.guardAgainstRepurchase(ctx, &result, offeringID, rec.Count, opts); done { - return result, result.Error - } + if done := c.guardBeforePurchase(ctx, &result, offeringID, rec.Count, opts); done { + return result, result.Error } // After the dedupe: a re-drive that already bought must not be refused for a // later price rise (ec2/client.go does the same). - if err = c.checkApprovedUpfront(ctx, rec, offeringID); err != nil { - result.Error = fmt.Errorf("%w: %w", common.ErrRefusedBeforePurchase, err) + if !c.readyToSend(ctx, &result, rec, offeringID, opts) { return result, result.Error } @@ -169,6 +172,9 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati // A single attempt: the API has no ClientToken or caller ID, so an SDK retry // after a lost response would buy a second node (MON-02). response, err := c.client.PurchaseReservedNodeOffering(ctx, input, func(o *redshift.Options) { o.RetryMaxAttempts = 1 }) + if journalActive(opts) { + return c.finishJournaled(ctx, result, response, err, offeringID, rec.Count, opts) + } if err != nil { result.Error = fmt.Errorf("failed to purchase Redshift Reserved Node: %w", purchasecfg.ClassifyPurchaseError(err)) if c.adoptAfterUnknownOutcome(ctx, &result, offeringID, rec.Count, opts) { @@ -193,6 +199,17 @@ func (c *Client) PurchaseCommitment(ctx context.Context, rec common.Recommendati return result, nil } +// readyToSend runs the last pre-send checks: the approved-upfront guard, then +// the attempt record on the journal path. It reports false, with the refusal in +// result.Error, when the purchase must not be sent. +func (c *Client) readyToSend(ctx context.Context, result *common.PurchaseResult, rec common.Recommendation, offeringID string, opts common.PurchaseOptions) bool { + if err := c.checkApprovedUpfront(ctx, rec, offeringID); err != nil { + result.Error = fmt.Errorf("%w: %w", common.ErrRefusedBeforePurchase, err) + return false + } + return c.recordSending(ctx, result, offeringID, opts) +} + // reservedNodeTotal is the reservation's upfront total: FixedPrice is per node, // so it is FixedPrice x NodeCount, or nil when either is missing. func reservedNodeTotal(rn *redshifttypes.ReservedNode) *float64 { diff --git a/providers/aws/services/redshift/dedupe.go b/providers/aws/services/redshift/dedupe.go index 789cb1d9..de3a0008 100644 --- a/providers/aws/services/redshift/dedupe.go +++ b/providers/aws/services/redshift/dedupe.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log" + "slices" "strings" "time" @@ -13,6 +14,7 @@ import ( 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/purchasecfg" ) const ( @@ -40,7 +42,10 @@ var errRequestedAtRequired = errors.New("RequestedAt is required with an idempot // dispatch another Redshift purchase for the same account, region, offering ID // and node count while a re-drive of an earlier lineage is still possible (the // platform's execution claim provides this). Otherwise a re-drive may adopt the -// later node or be refused as ambiguous. +// 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. func (c *Client) findMatchingNodes(ctx context.Context, offeringID string, count int, requestedAt time.Time) ([]string, error) { since := requestedAt.Add(-redshiftClockSkew) var ids []string @@ -132,3 +137,183 @@ func adopt(result *common.PurchaseResult, nodeID string) { result.CommitmentID = nodeID result.Error = nil } + +// journalActive reports whether the attempt journal governs this purchase +// (go#107). Without a token or a journal the flow is the window dedupe alone. +func journalActive(opts common.PurchaseOptions) bool { + return opts.IdempotencyToken != "" && opts.AttemptJournal != nil +} + +// guardBeforePurchase runs the pre-send dedupe for a tokenized purchase and +// reports done when the purchase must not proceed (result carries the outcome). +func (c *Client) guardBeforePurchase(ctx context.Context, result *common.PurchaseResult, offeringID string, count int, opts common.PurchaseOptions) (done bool) { + switch { + case journalActive(opts): + return c.journalPreflight(ctx, result, offeringID, count, opts) + case opts.IdempotencyToken != "": + return c.guardAgainstRepurchase(ctx, result, offeringID, count, opts) + default: + return false + } +} + +func refuseBeforePurchase(result *common.PurchaseResult, format string, args ...any) { + result.Error = fmt.Errorf("%w: "+format, append([]any{common.ErrRefusedBeforePurchase}, args...)...) +} + +// journalPreflight is the journal path of the pre-send dedupe: the caller's +// recorded attempts first, then the start-time window. +func (c *Client) journalPreflight(ctx context.Context, result *common.PurchaseResult, offeringID string, count int, opts common.PurchaseOptions) (done bool) { + if opts.RequestedAt.IsZero() { + result.Error = errRequestedAtRequired + return true + } + if opts.ExecutionID == "" { + refuseBeforePurchase(result, "an execution ID is required with an attempt journal") + return true + } + priors, err := opts.AttemptJournal.PriorAttempts(ctx, opts.IdempotencyToken) + if err != nil { + refuseBeforePurchase(result, "attempt journal lookup failed for token %s: %w", common.MaskToken(opts.IdempotencyToken), err) + return true + } + if ids := recordedReservationIDs(priors); len(ids) > 0 { + c.adoptRecordedNode(ctx, result, ids, offeringID, count, opts) + return true + } + if c.guardAgainstRepurchase(ctx, result, offeringID, count, opts) { + if result.Success { + c.recordPurchased(ctx, result, offeringID, opts) + } + return true + } + return refuseUnresolvedPriors(result, priors, offeringID, opts) +} + +// recordedReservationIDs returns the distinct reservation IDs of Purchased +// records, in first-seen order. +func recordedReservationIDs(priors []common.AttemptRecord) []string { + var ids []string + for _, p := range priors { + if p.State == common.AttemptPurchased && !slices.Contains(ids, p.ReservationID) { + ids = append(ids, p.ReservationID) + } + } + return ids +} + +// adoptRecordedNode adopts the node an earlier attempt recorded as bought, +// whatever else the window matches, once the account still lists it. It writes +// nothing: the existing Purchased record already answers later attempts. +func (c *Client) adoptRecordedNode(ctx context.Context, result *common.PurchaseResult, ids []string, offeringID string, count int, opts common.PurchaseOptions) { + if len(ids) > 1 { + refuseBeforePurchase(result, "attempt journal is inconsistent for token %s: %d distinct reservation IDs recorded as purchased", common.MaskToken(opts.IdempotencyToken), len(ids)) + return + } + matches, err := c.findMatchingNodes(ctx, offeringID, count, opts.RequestedAt) + switch { + case err != nil: + refuseBeforePurchase(result, "lookup of recorded reserved node %s failed: %w", ids[0], err) + case !slices.Contains(matches, ids[0]): + refuseBeforePurchase(result, "reserved node %s recorded for token %s is not among the matching owned nodes; check the account before buying again", ids[0], common.MaskToken(opts.IdempotencyToken)) + default: + log.Printf("Redshift reserved node %s is recorded for token %s; skipping purchase", ids[0], common.MaskToken(opts.IdempotencyToken)) + adopt(result, ids[0]) + } +} + +// refuseUnresolvedPriors refuses when an earlier attempt may have sent a buy +// that this attempt could not recognize: the caller's own sending or +// outcome_unknown record, or another execution's record taken under an offering +// that differs from the current one (or has none). +func refuseUnresolvedPriors(result *common.PurchaseResult, priors []common.AttemptRecord, offeringID string, opts common.PurchaseOptions) (done bool) { + for _, p := range priors { + if p.State != common.AttemptSending && p.State != common.AttemptOutcomeUnknown { + continue + } + switch { + case p.ExecutionID == opts.ExecutionID: + result.Error = fmt.Errorf("an earlier attempt of this execution for token %s is %s and no matching node was found: %w; check the account before buying again", + common.MaskToken(opts.IdempotencyToken), p.State, common.ErrOutcomeUnknown) + return true + case p.OfferingID != offeringID: + result.Error = fmt.Errorf("an earlier attempt for token %s under offering %q may have bought a node that the current offering %q cannot match: %w; check the account before buying again", + common.MaskToken(opts.IdempotencyToken), p.OfferingID, offeringID, common.ErrOutcomeUnknown) + return true + } + } + return false +} + +// recordSending writes the pre-send record on the journal path. It reports +// false, with the refusal in result.Error, when the record is not durable: no +// record, no buy. Without a journal it does nothing and reports true. +func (c *Client) recordSending(ctx context.Context, result *common.PurchaseResult, offeringID string, opts common.PurchaseOptions) bool { + if !journalActive(opts) { + return true + } + err := opts.AttemptJournal.RecordAttempt(ctx, common.AttemptRecord{ + IdempotencyToken: opts.IdempotencyToken, ExecutionID: opts.ExecutionID, + State: common.AttemptSending, OfferingID: offeringID, + }) + if err != nil { + refuseBeforePurchase(result, "could not record the purchase attempt for token %s: %w", common.MaskToken(opts.IdempotencyToken), err) + return false + } + return true +} + +// writeAttempt writes a record on a context detached from the caller's +// cancellation: the buy may already have happened. +func writeAttempt(ctx context.Context, opts common.PurchaseOptions, state common.AttemptState, reservationID, offeringID string) error { + writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), recoverLookupTimeout) + defer cancel() + return opts.AttemptJournal.RecordAttempt(writeCtx, common.AttemptRecord{ + IdempotencyToken: opts.IdempotencyToken, ExecutionID: opts.ExecutionID, + State: state, ReservationID: reservationID, OfferingID: offeringID, + }) +} + +// recordPurchased records result.CommitmentID as bought. A node that exists but +// is not recorded is never a Success: the error names the ID and wraps +// ErrOutcomeUnknown, and CommitmentID stays set. +func (c *Client) recordPurchased(ctx context.Context, result *common.PurchaseResult, offeringID string, opts common.PurchaseOptions) { + err := writeAttempt(ctx, opts, common.AttemptPurchased, result.CommitmentID, offeringID) + if err == nil { + return + } + log.Printf("Redshift reserved node %s was bought for token %s but the attempt record failed: %v", result.CommitmentID, common.MaskToken(opts.IdempotencyToken), err) + result.Success = false + result.ExistingCommitment = false + result.Error = fmt.Errorf("redshift reserved node %s was bought but not recorded: %w: %w", result.CommitmentID, err, common.ErrOutcomeUnknown) +} + +// recordUnknown records a send whose outcome could not be resolved. The +// original error is returned either way. +func recordUnknown(ctx context.Context, result *common.PurchaseResult, offeringID string, opts common.PurchaseOptions) { + if err := writeAttempt(ctx, opts, common.AttemptOutcomeUnknown, "", offeringID); err != nil { + result.Error = fmt.Errorf("%w; additionally the outcome_unknown attempt record could not be written: %w", result.Error, err) + } +} + +// finishJournaled turns the buy response into a result and records the outcome. +func (c *Client) finishJournaled(ctx context.Context, result common.PurchaseResult, response *redshift.PurchaseReservedNodeOfferingOutput, sendErr error, offeringID string, count int, opts common.PurchaseOptions) (common.PurchaseResult, error) { + switch { + case sendErr != nil: + result.Error = fmt.Errorf("failed to purchase Redshift Reserved Node: %w", purchasecfg.ClassifyPurchaseError(sendErr)) + case response == nil || response.ReservedNode == nil: + result.Error = fmt.Errorf("purchase response was empty: %w", common.ErrOutcomeUnknown) + default: + result.Success = true + result.CommitmentID = aws.ToString(response.ReservedNode.ReservedNodeId) + result.Cost = reservedNodeTotal(response.ReservedNode) + c.recordPurchased(ctx, &result, offeringID, opts) + return result, result.Error + } + if c.adoptAfterUnknownOutcome(ctx, &result, offeringID, count, opts) { + c.recordPurchased(ctx, &result, offeringID, opts) + return result, result.Error + } + recordUnknown(ctx, &result, offeringID, opts) + return result, result.Error +} diff --git a/providers/aws/services/redshift/journal_test.go b/providers/aws/services/redshift/journal_test.go new file mode 100644 index 00000000..78ed2fb5 --- /dev/null +++ b/providers/aws/services/redshift/journal_test.go @@ -0,0 +1,539 @@ +package redshift + +import ( + "context" + "errors" + "slices" + "sync" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/redshift" + "github.com/aws/aws-sdk-go-v2/service/redshift/types" + "github.com/aws/smithy-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" +) + +// All tests here are fixture-based: mocks and an in-test journal, no AWS calls. + +const ( + jExecNew = "exec-new" + jExecOld = "exec-old" + jOffer = "offering-1" +) + +var jToken = common.DeriveIdempotencyToken("journal", 0) + +type journalCall struct { + Method string + Rec common.AttemptRecord +} + +// fakeJournal enforces the transition table and keeps the first OfferingID of +// each (token, execution) row, as a store must. +type fakeJournal struct { + mu sync.Mutex + rows []common.AttemptRecord + calls []journalCall + priorErr error + failOn map[common.AttemptState]error +} + +func newFakeJournal() *fakeJournal { + return &fakeJournal{failOn: map[common.AttemptState]error{}} +} + +func (f *fakeJournal) find(token, exec string) int { + return slices.IndexFunc(f.rows, func(r common.AttemptRecord) bool { return r.IdempotencyToken == token && r.ExecutionID == exec }) +} + +func (f *fakeJournal) seed(exec string, state common.AttemptState, id, offering string) { + f.rows = append(f.rows, common.AttemptRecord{IdempotencyToken: jToken, ExecutionID: exec, State: state, ReservationID: id, OfferingID: offering}) +} + +func (f *fakeJournal) PriorAttempts(_ context.Context, token string) ([]common.AttemptRecord, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, journalCall{Method: "PriorAttempts"}) + if f.priorErr != nil { + return nil, f.priorErr + } + var out []common.AttemptRecord + for _, r := range f.rows { + if r.IdempotencyToken == token { + out = append(out, r) + } + } + return out, nil +} + +func (f *fakeJournal) RecordAttempt(ctx context.Context, rec common.AttemptRecord) error { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, journalCall{Method: "RecordAttempt", Rec: rec}) + if err := ctx.Err(); err != nil { + return err + } + if err := f.failOn[rec.State]; err != nil { + return err + } + if err := rec.Validate(); err != nil { + return err + } + i := f.find(rec.IdempotencyToken, rec.ExecutionID) + var from *common.AttemptRecord + if i >= 0 { + from = &f.rows[i] + } + if !validTransition(from, rec) { + return errors.New("fake journal: invalid transition") + } + if i < 0 { + f.rows = append(f.rows, rec) + return nil + } + rec.OfferingID = f.rows[i].OfferingID + f.rows[i] = rec + return nil +} + +func (f *fakeJournal) recordCalls() []common.AttemptRecord { + var out []common.AttemptRecord + for _, c := range f.calls { + if c.Method == "RecordAttempt" { + out = append(out, c.Rec) + } + } + return out +} + +func (f *fakeJournal) row(t *testing.T, exec string) common.AttemptRecord { + t.Helper() + i := f.find(jToken, exec) + require.GreaterOrEqual(t, i, 0, "no record for %s", exec) + return f.rows[i] +} + +// validTransition mirrors the table the AttemptJournal contract documents; +// a store implements the same table. +func validTransition(from *common.AttemptRecord, to common.AttemptRecord) bool { + if from == nil { + return to.State == common.AttemptSending || to.State == common.AttemptPurchased + } + switch from.State { + case common.AttemptSending: + return to.State == common.AttemptPurchased || to.State == common.AttemptOutcomeUnknown + case common.AttemptOutcomeUnknown: + return to.State == common.AttemptPurchased + case common.AttemptPurchased: + return to.State == common.AttemptPurchased && to.ReservationID == from.ReservationID + } + return false +} + +func TestValidTransition(t *testing.T) { + t.Parallel() + S, U, P := common.AttemptSending, common.AttemptOutcomeUnknown, common.AttemptPurchased + rec := func(s common.AttemptState, id string) common.AttemptRecord { + return common.AttemptRecord{State: s, ReservationID: id} + } + none, sending, purchased, unknown := (*common.AttemptRecord)(nil), ptr(rec(S, "")), ptr(rec(P, "rn-1")), ptr(rec(U, "")) + for _, tc := range []struct { + name string + from *common.AttemptRecord + to common.AttemptRecord + want bool + }{ + {"none to sending", none, rec(S, ""), true}, + {"none to unknown", none, rec(U, ""), false}, + {"none to purchased", none, rec(P, "rn-1"), true}, + {"none to purchased other id", none, rec(P, "rn-2"), true}, + {"sending to sending", sending, rec(S, ""), false}, + {"sending to unknown", sending, rec(U, ""), true}, + {"sending to purchased", sending, rec(P, "rn-1"), true}, + {"sending to purchased other id", sending, rec(P, "rn-2"), true}, + {"purchased to sending", purchased, rec(S, ""), false}, + {"purchased to unknown", purchased, rec(U, ""), false}, + {"purchased to purchased same id", purchased, rec(P, "rn-1"), true}, + {"purchased to purchased other id", purchased, rec(P, "rn-2"), false}, + {"unknown to sending", unknown, rec(S, ""), false}, + {"unknown to unknown", unknown, rec(U, ""), false}, + {"unknown to purchased", unknown, rec(P, "rn-1"), true}, + {"unknown to purchased other id", unknown, rec(P, "rn-2"), true}, + } { + assert.Equal(t, tc.want, validTransition(tc.from, tc.to), tc.name) + } +} + +func ptr[T any](v T) *T { return &v } + +func jOpts(j *fakeJournal, exec string) common.PurchaseOptions { + return common.PurchaseOptions{IdempotencyToken: jToken, RequestedAt: dedupeReqAt, ExecutionID: exec, AttemptJournal: j} +} + +func jWindow(ids ...string) []types.ReservedNode { + nodes := make([]types.ReservedNode, 0, len(ids)) + for _, id := range ids { + nodes = append(nodes, dnode(id, "active", dedupeReqAt.Add(time.Minute))) + } + return nodes +} + +func jNoBuy(t *testing.T, m *MockRedshiftClient) { + t.Helper() + m.AssertNotCalled(t, "PurchaseReservedNodeOffering", mock.Anything, mock.Anything) +} + +func transportErr() error { + return &smithy.OperationError{ServiceID: "Redshift", OperationName: "PurchaseReservedNodeOffering", Err: errors.New("read: connection reset")} +} + +func TestJournal_PriorErrorRefusesNoSend(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.priorErr = errors.New("db down") + c, m := newDedupe(nil) + 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 TestJournal_SendingWriteErrorRefusesNoSend(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.failOn[common.AttemptSending] = errors.New("db down") + c, m := newDedupe(nil) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrRefusedBeforePurchase) + assert.False(t, result.Success) + jNoBuy(t, m) +} + +func TestJournal_PurchasedWriteErrorIsOutcomeUnknownWithID(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.failOn[common.AttemptPurchased] = errors.New("db down") + c, m := newDedupe(nil) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.Error(t, err) + assert.False(t, result.Success) + assert.Equal(t, "rn-new", result.CommitmentID) + assert.Contains(t, err.Error(), "rn-new") + assert.ErrorIs(t, err, common.ErrOutcomeUnknown) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) +} + +func TestJournal_BuyRecordsSendingThenPurchased(t *testing.T) { + t.Parallel() + j := newFakeJournal() + c, m := newDedupe(nil) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.NoError(t, err) + assert.True(t, result.Success) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) + assert.Equal(t, []common.AttemptRecord{ + {IdempotencyToken: jToken, ExecutionID: jExecNew, State: common.AttemptSending, OfferingID: jOffer}, + {IdempotencyToken: jToken, ExecutionID: jExecNew, State: common.AttemptPurchased, ReservationID: "rn-new", OfferingID: jOffer}, + }, j.recordCalls()) +} + +func TestJournal_RecordedIDAdoptsDespiteAmbiguousWindow(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seed(jExecOld, common.AttemptPurchased, "rn-1", 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.Success) + assert.True(t, result.ExistingCommitment) + assert.Equal(t, "rn-1", result.CommitmentID) + assert.Empty(t, j.recordCalls()) + jNoBuy(t, m) +} + +func TestJournal_RecordedIDNotMatchedRefuses(t *testing.T) { + t.Parallel() + for name, window := range map[string][]types.ReservedNode{"other node only": jWindow("rn-1"), "empty window": nil} { + t.Run(name, func(t *testing.T) { + j := newFakeJournal() + j.seed(jExecOld, common.AttemptPurchased, "rn-9", jOffer) + c, m := newDedupe(window) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrRefusedBeforePurchase) + assert.False(t, result.Success) + assert.NotEqual(t, "rn-1", result.CommitmentID) + assert.Empty(t, j.recordCalls()) + jNoBuy(t, m) + }) + } +} + +func TestJournal_ConflictingRecordedIDsRefuses(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seed("exec-a", common.AttemptPurchased, "rn-1", jOffer) + j.seed("exec-b", common.AttemptPurchased, "rn-2", jOffer) + c, m := newDedupe(jWindow("rn-1", "rn-2")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrRefusedBeforePurchase) + assert.Empty(t, result.CommitmentID) + assert.Empty(t, j.recordCalls()) + jNoBuy(t, m) +} + +func TestJournal_SameExecUnresolvedPriorRefusesNoSend(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seed("exec-1", common.AttemptSending, "", jOffer) + c, m := newDedupe(nil) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, "exec-1")) + require.ErrorIs(t, err, common.ErrOutcomeUnknown) + assert.NotErrorIs(t, err, common.ErrRefusedBeforePurchase) + assert.False(t, result.Success) + assert.Empty(t, j.recordCalls()) + jNoBuy(t, m) +} + +func TestJournal_OtherExecUnresolvedPriorBuys(t *testing.T) { + t.Parallel() + for _, state := range []common.AttemptState{common.AttemptSending, common.AttemptOutcomeUnknown} { + t.Run(string(state), func(t *testing.T) { + j := newFakeJournal() + j.seed(jExecOld, state, "", jOffer) + c, m := newDedupe(nil) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.NoError(t, err) + assert.True(t, result.Success) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) + assert.Equal(t, common.AttemptPurchased, j.row(t, jExecNew).State) + assert.Equal(t, "rn-new", j.row(t, jExecNew).ReservationID) + }) + } +} + +func TestJournal_OfferingChangedUnresolvedPriorRefuses(t *testing.T) { + t.Parallel() + for name, prior := range map[string]string{"different prior offering": "offering-0", "empty prior offering": ""} { + t.Run(name, func(t *testing.T) { + j := newFakeJournal() + j.seed(jExecOld, common.AttemptSending, "", prior) + c, m := newDedupe(nil) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrOutcomeUnknown) + assert.NotErrorIs(t, err, common.ErrRefusedBeforePurchase) + assert.False(t, result.Success) + assert.Empty(t, j.recordCalls()) + jNoBuy(t, m) + }) + } +} + +func TestJournal_CrashAfterSendResolvedByWindow(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seed(jExecOld, common.AttemptSending, "", jOffer) + 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, common.AttemptPurchased, j.row(t, jExecNew).State) + assert.Equal(t, "rn-1", j.row(t, jExecNew).ReservationID) +} + +func TestJournal_UnknownThenAdoptRecordsPurchased(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seed("exec-1", common.AttemptOutcomeUnknown, "", jOffer) + c, m := newDedupe(jWindow("rn-1")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, "exec-1")) + require.NoError(t, err) + assert.Equal(t, "rn-1", result.CommitmentID) + jNoBuy(t, m) + assert.Equal(t, common.AttemptPurchased, j.row(t, "exec-1").State) + assert.Equal(t, "rn-1", j.row(t, "exec-1").ReservationID) +} + +func TestJournal_AdoptWriteErrorNotSuccess(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.failOn[common.AttemptPurchased] = errors.New("db down") + c, m := newDedupe(jWindow("rn-1")) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.Error(t, err) + assert.False(t, result.Success) + assert.Equal(t, "rn-1", result.CommitmentID) + assert.False(t, result.ExistingCommitment) + assert.Contains(t, err.Error(), "rn-1") + assert.ErrorIs(t, err, common.ErrOutcomeUnknown) + jNoBuy(t, m) +} + +func TestJournal_PostSendAdoptWriteErrorNotSuccess(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.failOn[common.AttemptPurchased] = errors.New("db down") + 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.False(t, result.ExistingCommitment) + assert.Equal(t, "rn-1", result.CommitmentID) + assert.Contains(t, err.Error(), "rn-1") +} + +func TestJournal_AdoptKeepsFirstOfferingID(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seed("exec-1", common.AttemptOutcomeUnknown, "", "offering-0") + c, _ := newDedupe(jWindow("rn-1")) + _, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, "exec-1")) + require.NoError(t, err) + row := j.row(t, "exec-1") + assert.Equal(t, common.AttemptPurchased, row.State) + assert.Equal(t, "offering-0", row.OfferingID) +} + +func TestFakeJournalEnforcesTable(t *testing.T) { + t.Parallel() + ctx := context.Background() + rec := func(s common.AttemptState, id string) common.AttemptRecord { + return common.AttemptRecord{IdempotencyToken: jToken, ExecutionID: "e", State: s, ReservationID: id} + } + j := newFakeJournal() + require.NoError(t, j.RecordAttempt(ctx, rec(common.AttemptSending, ""))) + require.Error(t, j.RecordAttempt(ctx, rec(common.AttemptSending, ""))) + j = newFakeJournal() + require.NoError(t, j.RecordAttempt(ctx, rec(common.AttemptPurchased, "rn-1"))) + require.Error(t, j.RecordAttempt(ctx, rec(common.AttemptPurchased, "rn-2"))) +} + +func TestJournal_CancelledCtxAfterSendStillRecords(t *testing.T) { + t.Parallel() + j := newFakeJournal() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + m := &MockRedshiftClient{} + expectRSOffering(m) + expectNodePages(m, nil) + m.On("PurchaseReservedNodeOffering", mock.Anything, mock.Anything).Run(func(mock.Arguments) { cancel() }). + Return(&redshift.PurchaseReservedNodeOfferingOutput{ReservedNode: &types.ReservedNode{ReservedNodeId: aws.String("rn-new")}}, nil) + c := &Client{client: m, region: "eu-west-1"} + result, err := c.PurchaseCommitment(ctx, rsIdemRec(), jOpts(j, jExecNew)) + require.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, common.AttemptPurchased, j.row(t, jExecNew).State) +} + +func TestJournal_UnknownOutcomeRecorded(t *testing.T) { + t.Parallel() + j := newFakeJournal() + c, m := recoverMock(buyFails(transportErr())) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrOutcomeUnknown) + assert.False(t, result.Success) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) + assert.Equal(t, common.AttemptOutcomeUnknown, j.row(t, jExecNew).State) +} + +func TestJournal_PostSendAdoptRecordsPurchased(t *testing.T) { + t.Parallel() + j := newFakeJournal() + 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.NoError(t, err) + assert.True(t, result.Success) + assert.Equal(t, "rn-1", result.CommitmentID) + m.AssertNumberOfCalls(t, "DescribeReservedNodes", 2) + recs := j.recordCalls() + require.Len(t, recs, 2) + assert.Equal(t, common.AttemptSending, recs[0].State) + assert.Equal(t, common.AttemptPurchased, recs[1].State) + assert.Equal(t, "rn-1", recs[1].ReservationID) +} + +func TestJournal_AmbiguousResponseRecordedUnknown(t *testing.T) { + t.Parallel() + t.Run("empty 200", func(t *testing.T) { + j := newFakeJournal() + c, _ := recoverMock(buyEmpty) + _, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrOutcomeUnknown) + assert.Equal(t, common.AttemptOutcomeUnknown, j.row(t, jExecNew).State) + }) + t.Run("definite 4xx", func(t *testing.T) { + j := newFakeJournal() + c, _ := recoverMock(buyFails(clean4xx())) + _, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.Error(t, err) + assert.NotErrorIs(t, err, common.ErrOutcomeUnknown) + assert.Equal(t, common.AttemptOutcomeUnknown, j.row(t, jExecNew).State) + }) +} + +func TestJournal_UnknownWriteErrorStillOutcomeUnknown(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.failOn[common.AttemptOutcomeUnknown] = errors.New("db down") + c, _ := recoverMock(buyFails(transportErr())) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrOutcomeUnknown) + assert.Contains(t, err.Error(), "db down") + assert.False(t, result.Success) +} + +func TestJournal_UpfrontRefusalWritesNothing(t *testing.T) { + t.Parallel() + j := newFakeJournal() + m := guardMock(1000, "") + expectNodePages(m, nil) + c := &Client{client: m, region: "eu-west-1"} + rec := rsIdemRec() + rec.CommitmentCost = 10 + _, err := c.PurchaseCommitment(context.Background(), rec, jOpts(j, jExecNew)) + require.ErrorIs(t, err, common.ErrRefusedBeforePurchase) + assert.Empty(t, j.recordCalls()) + jNoBuy(t, m) +} + +func TestJournal_ZeroRequestedAtRefusesBeforeJournal(t *testing.T) { + t.Parallel() + j := newFakeJournal() + j.seed(jExecOld, common.AttemptPurchased, "rn-1", jOffer) + c, m := newDedupe(jWindow("rn-1")) + opts := jOpts(j, jExecNew) + opts.RequestedAt = time.Time{} + _, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), opts) + require.ErrorIs(t, err, errRequestedAtRequired) + assert.Empty(t, j.calls) + jNoBuy(t, m) +} + +func TestJournal_EmptyTokenSkipsJournal(t *testing.T) { + t.Parallel() + j := newFakeJournal() + c, m := newDedupe(nil) + result, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), common.PurchaseOptions{AttemptJournal: j}) + require.NoError(t, err) + assert.True(t, result.Success) + assert.Empty(t, j.calls) + m.AssertNumberOfCalls(t, "PurchaseReservedNodeOffering", 1) +} + +func TestJournal_RequiresExecutionID(t *testing.T) { + t.Parallel() + j := newFakeJournal() + c, m := newDedupe(nil) + _, err := c.PurchaseCommitment(context.Background(), rsIdemRec(), jOpts(j, "")) + require.ErrorIs(t, err, common.ErrRefusedBeforePurchase) + assert.Empty(t, j.calls) + jNoBuy(t, m) +}