feat(redshift): journal purchase attempts around the reserved-node purchase

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

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=

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 {

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
+}