feat(api): gate Redshift retries and copy the first dispatch time

internal/api/handler_purchases.go

@@ -2556,6 +2556,9 @@ func (h *Handler) loadAndValidateRetryRequest(ctx context.Context, req *events.L
 	if err := checkRetryEligibilityGates(failedExec, req); err != nil {
 		return nil, nil, err
 	}
+	if err := h.checkRedshiftRetryGate(ctx, failedExec); err != nil {
+		return nil, nil, err
+	}
 
 	return failedExec, session, nil
 }
@@ -2784,6 +2787,7 @@ func (h *Handler) persistRetryExecution(ctx context.Context, failedExec *config.
 		CapacityPercent:        failedExec.CapacityPercent,
 		CreatedByUserID:        resolveCreatorUserID(session),
 		RetryAttemptN:          failedExec.RetryAttemptN + 1,
+		FirstDispatchAt:        retrySuccessorFirstDispatchAt(failedExec),
 	}
 
 	graceCfg, err := h.gracePeriodConfig(ctx)

internal/api/handler_purchases_redshift_retry.go

@@ -0,0 +1,45 @@
+package api
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/purchase"
+)
+
+// checkRedshiftRetryGate refuses a Retry that could still double-buy: permanently
+// (unknown or past-horizon dispatch) or until AWS settles. Force never applies.
+func (h *Handler) checkRedshiftRetryGate(ctx context.Context, failedExec *config.PurchaseExecution) error {
+	if !purchase.IsRedshiftExecution(failedExec) {
+		return nil
+	}
+	clock, err := h.config.GetRedshiftRetryClock(ctx, failedExec.ExecutionID)
+	if err != nil {
+		return fmt.Errorf("read redshift retry clock: %w", err)
+	}
+	if reason := purchase.RedshiftRetryPermanentRefusal(failedExec, clock.SinceStamp); reason != "" {
+		return NewClientErrorWithDetails(409,
+			"this purchase cannot be retried safely: "+reason,
+			map[string]any{"ops_hint": reason, "redrive_unsafe": true})
+	}
+	if wait := purchase.RedshiftRetryWait(failedExec, clock.SinceUpdate); wait > 0 {
+		secs := int((wait + time.Second - 1) / time.Second)
+		return NewClientErrorWithHeaders(409,
+			"this Redshift purchase failed too recently to retry safely; retry after the wait",
+			map[string]any{"retry_after_seconds": secs},
+			map[string]string{"Retry-After": strconv.Itoa(secs)})
+	}
+	return nil
+}
+
+// retrySuccessorFirstDispatchAt carries the lineage stamp to a retry successor;
+// a never-submitted attempt 0 has none and the successor mints its own.
+func retrySuccessorFirstDispatchAt(failedExec *config.PurchaseExecution) *time.Time {
+	if purchase.NeverSubmitted(failedExec) {
+		return nil
+	}
+	return clonePtr(failedExec.FirstDispatchAt)
+}

internal/api/handler_purchases_redshift_retry_test.go

@@ -0,0 +1,158 @@
+package api
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+const redshiftRetryExecID = "22222222-3333-4444-5555-666666666701"
+
+func redshiftFailedRow(attempt int, noSubmission bool) *config.PurchaseExecution {
+	failed := redriveFailedRow(redshiftRetryExecID, "aws", "redshift")
+	failed.RetryAttemptN = attempt
+	failed.NoSubmission = noSubmission
+	return failed
+}
+
+func gateWithClock(t *testing.T, failed *config.PurchaseExecution, clock config.RedshiftRetryClock) *Handler {
+	t.Helper()
+	store := new(MockConfigStore)
+	store.On("GetRedshiftRetryClock", mock.Anything, failed.ExecutionID).Return(clock, nil)
+	return &Handler{config: store}
+}
+
+func ageOf(d time.Duration) *time.Duration { return &d }
+
+func TestCheckRedshiftRetryGate_LegacyStamplessPermanent(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	h := gateWithClock(t, failed, config.RedshiftRetryClock{SinceUpdate: time.Hour})
+
+	err := h.checkRedshiftRetryGate(context.Background(), failed)
+
+	assertRedriveRefused(t, err, "no first dispatch time recorded")
+	ce, _ := IsClientError(err)
+	assert.NotContains(t, ce.Details(), "retry_after_seconds")
+}
+
+func TestCheckRedshiftRetryGate_PastHorizonPermanent(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	h := gateWithClock(t, failed, config.RedshiftRetryClock{SinceUpdate: time.Hour, SinceStamp: ageOf(24 * time.Hour)})
+
+	assertRedriveRefused(t, h.checkRedshiftRetryGate(context.Background(), failed), "at least 24h ago")
+}
+
+func TestCheckRedshiftRetryGate_InsideSettleRetryAfter(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	// 9m59.5s since update leaves 500ms to wait, which must round up to 1s.
+	h := gateWithClock(t, failed, config.RedshiftRetryClock{SinceUpdate: 10*time.Minute - 500*time.Millisecond, SinceStamp: ageOf(time.Hour)})
+
+	err := h.checkRedshiftRetryGate(context.Background(), failed)
+
+	ce, ok := IsClientError(err)
+	require.True(t, ok, "got %v", err)
+	assert.Equal(t, 409, ce.code)
+	assert.Equal(t, 1, ce.Details()["retry_after_seconds"])
+	assert.NotContains(t, ce.Details(), "redrive_unsafe", "a settle wait is temporary")
+	assert.Equal(t, "1", ce.headers["Retry-After"])
+}
+
+func TestCheckRedshiftRetryGate_AfterSettleAllowed(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	h := gateWithClock(t, failed, config.RedshiftRetryClock{SinceUpdate: 10 * time.Minute, SinceStamp: ageOf(time.Hour)})
+
+	assert.NoError(t, h.checkRedshiftRetryGate(context.Background(), failed))
+}
+
+func TestCheckRedshiftRetryGate_NoSubmissionAttempt0Skipped(t *testing.T) {
+	failed := redshiftFailedRow(0, true)
+	h := gateWithClock(t, failed, config.RedshiftRetryClock{SinceUpdate: 0, SinceStamp: nil})
+
+	assert.NoError(t, h.checkRedshiftRetryGate(context.Background(), failed))
+}
+
+func TestCheckRedshiftRetryGate_NonRedshiftNeverReadsClock(t *testing.T) {
+	for _, svc := range []string{"ec2", "rds"} {
+		failed := redriveFailedRow(redshiftRetryExecID, "aws", svc)
+		store := new(MockConfigStore)
+		h := &Handler{config: store}
+
+		assert.NoError(t, h.checkRedshiftRetryGate(context.Background(), failed), svc)
+		store.AssertNotCalled(t, "GetRedshiftRetryClock", mock.Anything, mock.Anything)
+	}
+}
+
+func TestCheckRedshiftRetryGate_ClockErrorFailsClosed(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	store := new(MockConfigStore)
+	store.On("GetRedshiftRetryClock", mock.Anything, failed.ExecutionID).Return(config.RedshiftRetryClock{}, assert.AnError)
+	h := &Handler{config: store}
+
+	err := h.checkRedshiftRetryGate(context.Background(), failed)
+
+	require.ErrorIs(t, err, assert.AnError)
+	_, isClient := IsClientError(err)
+	assert.False(t, isClient, "a store failure is a 500, not a refusal the UI would render as permanent")
+}
+
+// Until the old guard is lifted it refuses every Redshift Retry first, so the
+// new gate must not even read the clock.
+func TestLoadAndValidateRetry_OldRedriveRefusalStillWinsForRedshift(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	session := &Session{UserID: retryCallerID, Email: "operator@example.com"}
+	h, store, _ := buildSessionRetryHandler(t, failed, session, false, true)
+
+	_, _, err := h.loadAndValidateRetryRequest(context.Background(), sessionRetryReqWithForce(), failed.ExecutionID)
+
+	assertRedriveRefused(t, err, "Check EC2 Reserved Instances or Redshift Reserved Nodes")
+	store.AssertNotCalled(t, "GetRedshiftRetryClock", mock.Anything, mock.Anything)
+}
+
+func persistedSuccessor(t *testing.T, failed *config.PurchaseExecution) *config.PurchaseExecution {
+	t.Helper()
+	session := &Session{UserID: retryCallerID, Email: "operator@example.com"}
+	store := new(MockConfigStore)
+	t.Cleanup(func() { store.AssertExpectations(t) })
+	store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{}, nil).Maybe()
+	h := &Handler{config: store}
+	var saved *config.PurchaseExecution
+	store.On("SavePurchaseExecution", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution")).
+		Run(func(args mock.Arguments) { snap := *args.Get(1).(*config.PurchaseExecution); saved = &snap }).
+		Return(nil)
+	store.On("LinkRetryExecutionAtomic", mock.Anything, mock.Anything, failed.ExecutionID, mock.AnythingOfType("string")).Return(true, nil)
+
+	_, _, err := h.persistRetryExecution(context.Background(), failed, session, 0, 0)
+
+	require.NoError(t, err)
+	require.NotNil(t, saved)
+	return saved
+}
+
+func TestPersistRetryExecution_CopiesFirstDispatchAt(t *testing.T) {
+	stamp := time.Now().Add(-time.Hour).UTC().Truncate(time.Microsecond)
+	failed := redshiftFailedRow(1, false)
+	failed.FirstDispatchAt = &stamp
+
+	successor := persistedSuccessor(t, failed)
+
+	require.NotNil(t, successor.FirstDispatchAt)
+	assert.True(t, stamp.Equal(*successor.FirstDispatchAt), "the successor keeps the lineage's original stamp, not a new one")
+	assert.NotSame(t, failed.FirstDispatchAt, successor.FirstDispatchAt, "the successor must not alias the predecessor's pointer")
+}
+
+func TestPersistRetryExecution_NoSubmissionAttempt0LeavesStampNil(t *testing.T) {
+	stamp := time.Now().Add(-time.Hour)
+	failed := redshiftFailedRow(0, true)
+	failed.FirstDispatchAt = &stamp
+
+	assert.Nil(t, persistedSuccessor(t, failed).FirstDispatchAt)
+}
+
+func TestPersistRetryExecution_StamplessPredecessorStaysStampless(t *testing.T) {
+	assert.Nil(t, persistedSuccessor(t, redshiftFailedRow(1, false)).FirstDispatchAt)
+}

internal/config/interfaces.go

@@ -827,6 +827,10 @@ type StoreInterface interface {
 	// not race (see the implementation for the predicate).
 	FindOpenRedshiftLineageConflictsTx(ctx context.Context, tx pgx.Tx, selfID, family string, tuples []RedshiftTuple, accountID *string, horizon time.Duration, stamp time.Time, window time.Duration) ([]RedshiftLineageConflict, error)
 
+	// GetRedshiftRetryClock returns how long ago an execution was last written
+	// and first dispatched, both measured on the database clock.
+	GetRedshiftRetryClock(ctx context.Context, executionID string) (RedshiftRetryClock, error)
+
 	// ListRecentSubmitsTx takes a per-creator transaction-scoped lock and
 	// returns the creator's web-submitted executions scheduled at or after
 	// since that are pending, in flight or bought. executePurchase's

internal/config/store_postgres_redshift_retry.go

@@ -0,0 +1,46 @@
+package config
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// RedshiftRetryClock is a row's age measured on the database clock.
+// SinceStamp is nil when the row has no first_dispatch_at, so a missing stamp
+// never reads as fresh.
+type RedshiftRetryClock struct {
+	SinceUpdate time.Duration
+	SinceStamp  *time.Duration
+}
+
+// GetRedshiftRetryClock returns how long ago the row was last written and
+// first dispatched, both computed from now() in SQL so app clock skew cannot
+// shorten the settle wait or the horizon.
+func (s *PostgresStore) GetRedshiftRetryClock(ctx context.Context, executionID string) (RedshiftRetryClock, error) {
+	var sinceUpdate float64
+	var sinceStamp *float64
+	err := s.db.QueryRow(ctx, `
+		SELECT EXTRACT(EPOCH FROM now() - updated_at)::float8,
+		       EXTRACT(EPOCH FROM now() - first_dispatch_at)::float8
+		FROM purchase_executions WHERE execution_id = $1`, executionID).Scan(&sinceUpdate, &sinceStamp)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return RedshiftRetryClock{}, fmt.Errorf("%w: execution %s", ErrNotFound, executionID)
+	}
+	if err != nil {
+		return RedshiftRetryClock{}, fmt.Errorf("read redshift retry clock of execution %s: %w", executionID, err)
+	}
+	clock := RedshiftRetryClock{SinceUpdate: secondsToDuration(sinceUpdate)}
+	if sinceStamp != nil {
+		d := secondsToDuration(*sinceStamp)
+		clock.SinceStamp = &d
+	}
+	return clock, nil
+}
+
+func secondsToDuration(secs float64) time.Duration {
+	return time.Duration(secs * float64(time.Second))
+}

internal/config/store_postgres_redshift_retry_integration_test.go

@@ -0,0 +1,51 @@
+//go:build integration
+// +build integration
+
+package config
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// The ages come from the database clock. The 000128 trigger overwrites updated_at
+// on UPDATE, so the aged row is INSERTed with explicit timestamps.
+func TestGetRedshiftRetryClock_UsesDBNow(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+
+	var id string
+	require.NoError(t, store.db.QueryRow(ctx, `
+		INSERT INTO purchase_executions (status, step_number, scheduled_date, first_dispatch_at, updated_at)
+		VALUES ('failed', 1, now(), now() - interval '3 hours', now() - interval '7 minutes')
+		RETURNING execution_id::text`).Scan(&id))
+
+	clock, err := store.GetRedshiftRetryClock(ctx, id)
+	require.NoError(t, err)
+	assert.InDelta(t, (7 * time.Minute).Seconds(), clock.SinceUpdate.Seconds(), 30)
+	require.NotNil(t, clock.SinceStamp)
+	assert.InDelta(t, (3 * time.Hour).Seconds(), clock.SinceStamp.Seconds(), 30)
+}
+
+func TestGetRedshiftRetryClock_NullStampIsNil(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+	exec := newNoSubmissionExec(ctx, t, store, "failed")
+
+	clock, err := store.GetRedshiftRetryClock(ctx, exec.ExecutionID)
+	require.NoError(t, err)
+	assert.Nil(t, clock.SinceStamp, "a NULL stamp must not read as fresh")
+	assert.GreaterOrEqual(t, clock.SinceUpdate, time.Duration(0))
+}
+
+func TestGetRedshiftRetryClock_UnknownExecutionIsNotFound(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+
+	_, err := store.GetRedshiftRetryClock(ctx, "00000000-0000-0000-0000-000000000000")
+	assert.ErrorIs(t, err, ErrNotFound)
+}

internal/mocks/stores.go

@@ -589,6 +589,17 @@ func (m *MockConfigStore) FindOpenRedshiftLineageConflictsTx(ctx context.Context
 	return res, args.Error(1)
 }
 
+// GetRedshiftRetryClock mocks the Redshift retry clock read (no default: testify panics if unexpected).
+func (m *MockConfigStore) GetRedshiftRetryClock(ctx context.Context, executionID string) (config.RedshiftRetryClock, error) {
+	m.record("GetRedshiftRetryClock", ctx, executionID)
+	args := m.Called(ctx, executionID)
+	clock, ok := args.Get(0).(config.RedshiftRetryClock)
+	if !ok {
+		m.reporter.wrongReturnType("GetRedshiftRetryClock", "config.RedshiftRetryClock", args.Get(0))
+	}
+	return clock, args.Error(1)
+}
+
 // TransitionExecutionStatus mocks the TransitionExecutionStatus operation.
 func (m *MockConfigStore) TransitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string) (*config.PurchaseExecution, error) {
 	m.record("TransitionExecutionStatus", ctx, executionID, fromStatuses, toStatus, actor)