fix(purchase): refuse a Redshift dispatch inside the adoption window

internal/config/interfaces.go

@@ -825,7 +825,7 @@ type StoreInterface interface {
 	// FindOpenRedshiftLineageConflictsTx returns the other open executions
 	// holding a Redshift rec for one of tuples that a dispatch of family must
 	// 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) ([]RedshiftLineageConflict, error)
+	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)
 
 	// ListRecentSubmitsTx takes a per-creator transaction-scoped lock and
 	// returns the creator's web-submitted executions scheduled at or after

internal/config/store_postgres_redshift_lineage.go

@@ -68,14 +68,12 @@ type RedshiftLineageConflict struct {
 	Recommendations []RecommendationRecord
 }
 
-// FindOpenRedshiftLineageConflictsTx returns the other open executions that
-// could buy or adopt a node for one of tuples, so a second lineage must not
-// dispatch. The row side account is the row's, else the rec's; the candidate
-// side is the tuple's, else accountID. A NULL on either side matches any; a rec account that is not a uuid counts as NULL.
-// A stamped running row is open at any age (the reaper bounds it); the other
-// open statuses only while their stamp is within horizon, judged on the
-// database clock. Same-family rows conflict only while running.
-func (s *PostgresStore) FindOpenRedshiftLineageConflictsTx(ctx context.Context, tx pgx.Tx, selfID, family string, tuples []RedshiftTuple, accountID *string, horizon time.Duration) ([]RedshiftLineageConflict, error) {
+// FindOpenRedshiftLineageConflictsTx returns the other executions whose node this lineage could
+// buy twice or adopt. Account: row's else rec's vs execution's else tuple's; NULL matches any.
+// Running rows conflict at any age; other open rows while their stamp is within horizon.
+// Other-family finished rows conflict while updated_at is within window of stamp, since go's
+// dedupe adopts their node. Same-family rows conflict only while running.
+func (s *PostgresStore) 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) {
 	if family == "" {
 		return nil, fmt.Errorf("find open redshift lineage conflicts: empty family")
 	}
@@ -97,6 +95,9 @@ func (s *PostgresStore) FindOpenRedshiftLineageConflictsTx(ctx context.Context,
 		        OR (split_part(COALESCE(NULLIF(e.idempotency_key, ''), e.execution_id::text), ':', 1) <> $2
 		            AND e.status IN ('approved', 'scheduled', 'pending', 'notified', 'paused')
 		            AND e.first_dispatch_at > now() - make_interval(secs => $3))
+		        OR (split_part(COALESCE(NULLIF(e.idempotency_key, ''), e.execution_id::text), ':', 1) <> $2
+		            AND e.status IN ('completed', 'partially_completed', 'revocation_requested')
+		            AND e.updated_at >= $6::timestamptz - make_interval(secs => $7))
 		      )
 		  AND EXISTS (
 		        SELECT 1
@@ -105,13 +106,13 @@ func (s *PostgresStore) FindOpenRedshiftLineageConflictsTx(ctx context.Context,
 		          ON r->>'region' = t.region AND r->>'resource_type' = t.node_type
 		         AND r->>'term' = t.term::text AND r->>'payment' = t.payment AND r->>'count' = t.count::text
 		         AND (COALESCE(e.cloud_account_id, CASE WHEN r->>'cloud_account_id' ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' THEN (r->>'cloud_account_id')::uuid END) IS NULL
-		              OR COALESCE(t.cloud_account_id, $4::uuid) IS NULL
-		              OR COALESCE(e.cloud_account_id, CASE WHEN r->>'cloud_account_id' ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' THEN (r->>'cloud_account_id')::uuid END) = COALESCE(t.cloud_account_id, $4::uuid))
+		              OR COALESCE($4::uuid, t.cloud_account_id) IS NULL
+		              OR COALESCE(e.cloud_account_id, CASE WHEN r->>'cloud_account_id' ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' THEN (r->>'cloud_account_id')::uuid END) = COALESCE($4::uuid, t.cloud_account_id))
 		        WHERE r->>'selected' = 'true'
 		          AND COALESCE(r->>'provider', '') IN ('', 'aws')
 		          AND r->>'service' IN ('redshift', 'data-warehouse'))
 		ORDER BY e.created_at, e.execution_id`,
-		selfID, family, horizon.Seconds(), accountID, tuplesJSON)
+		selfID, family, horizon.Seconds(), accountID, tuplesJSON, stamp, window.Seconds())
 	if err != nil {
 		return nil, fmt.Errorf("find open redshift lineage conflicts: %w", err)
 	}

internal/config/store_postgres_redshift_lineage_integration_test.go

@@ -13,7 +13,10 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-const lineageTestHorizon = 24 * time.Hour
+const (
+	lineageTestHorizon = 24 * time.Hour
+	lineageTestWindow  = 10 * time.Minute
+)
 
 var lineageTuple = RedshiftTuple{Region: "us-east-1", NodeType: "ra3.xlplus", Term: 1, Payment: "all-upfront", Count: 2}
 
@@ -43,7 +46,7 @@ func findLineageConflicts(ctx context.Context, t *testing.T, store *PostgresStor
 	var got []RedshiftLineageConflict
 	require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error {
 		var err error
-		got, err = store.FindOpenRedshiftLineageConflictsTx(ctx, tx, self, family, []RedshiftTuple{lineageTuple}, nil, lineageTestHorizon)
+		got, err = store.FindOpenRedshiftLineageConflictsTx(ctx, tx, self, family, []RedshiftTuple{lineageTuple}, nil, lineageTestHorizon, time.Now(), lineageTestWindow)
 		return err
 	}))
 	return got
@@ -156,7 +159,7 @@ func TestFindOpenRedshiftLineageConflicts(t *testing.T) {
 			var got []RedshiftLineageConflict
 			require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error {
 				var e error
-				got, e = store.FindOpenRedshiftLineageConflictsTx(ctx, tx, self.ExecutionID, "fam-k", []RedshiftTuple{tp}, nil, lineageTestHorizon)
+				got, e = store.FindOpenRedshiftLineageConflictsTx(ctx, tx, self.ExecutionID, "fam-k", []RedshiftTuple{tp}, nil, lineageTestHorizon, time.Now(), lineageTestWindow)
 				return e
 			}))
 			return got
@@ -177,7 +180,7 @@ func TestFindOpenRedshiftLineageConflicts(t *testing.T) {
 		var got []RedshiftLineageConflict
 		require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error {
 			var e error
-			got, e = store.FindOpenRedshiftLineageConflictsTx(ctx, tx, self, family, []RedshiftTuple{tp}, arg, lineageTestHorizon)
+			got, e = store.FindOpenRedshiftLineageConflictsTx(ctx, tx, self, family, []RedshiftTuple{tp}, arg, lineageTestHorizon, time.Now(), lineageTestWindow)
 			return e
 		}))
 		return got
@@ -269,7 +272,7 @@ func TestFindOpenRedshiftLineageConflicts(t *testing.T) {
 						}
 						tp := lineageTuple
 						tp.CloudAccountID = tupA
-						l, r := eff(rowA, recA), eff(tupA, argA)
+						l, r := eff(rowA, recA), eff(argA, tupA)
 						want := l == nil || r == nil || *l == *r
 						got := candidate(self.ExecutionID, "fam-q", tp, argA)
 						assert.Equal(t, want, len(got) == 1, "row=%v rec=%v tuple=%v arg=%v", rowA, recA, tupA, argA)
@@ -288,10 +291,29 @@ func TestFindOpenRedshiftLineageConflicts(t *testing.T) {
 		tp.CloudAccountID = &a1
 		assert.Equal(t, []string{other.ExecutionID}, conflictIDs(candidate(self.ExecutionID, "fam-r", tp, nil)))
 	})
+	t.Run("candidate_exec_account_beats_rec_account", func(t *testing.T) {
+		a1, a2 := "11111111-1111-1111-1111-111111111111", "22222222-2222-2222-2222-222222222222"
+		for _, a := range []string{a1, a2} {
+			_, err := store.db.Exec(ctx, `INSERT INTO cloud_accounts (id, name, provider, external_id) VALUES ($1::uuid, $1::text, 'aws', $1::text) ON CONFLICT DO NOTHING`, a)
+			require.NoError(t, err)
+		}
+		reset(t)
+		self := seedLineageRow(ctx, t, store, "fam-s", "running", recent)
+		other := seedLineageRow(ctx, t, store, "fam-s-other", "running", recent)
+		_, err := store.db.Exec(ctx, `UPDATE purchase_executions SET cloud_account_id = $2 WHERE execution_id = $1`, other.ExecutionID, a1)
+		require.NoError(t, err)
+		// The tuple (rec) account differs from the row's, the execution account is the row's.
+		tp := lineageTuple
+		tp.CloudAccountID = &a2
+		assert.Equal(t, []string{other.ExecutionID}, conflictIDs(candidate(self.ExecutionID, "fam-s", tp, &a1)))
+		// Reverse: the tuple account is the row's, the execution account differs.
+		tp.CloudAccountID = &a1
+		assert.Empty(t, candidate(self.ExecutionID, "fam-s", tp, &a2))
+	})
 	t.Run("empty_family_is_rejected", func(t *testing.T) {
 		reset(t)
 		require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error {
-			_, err := store.FindOpenRedshiftLineageConflictsTx(ctx, tx, "x", "", []RedshiftTuple{lineageTuple}, nil, lineageTestHorizon)
+			_, err := store.FindOpenRedshiftLineageConflictsTx(ctx, tx, "x", "", []RedshiftTuple{lineageTuple}, nil, lineageTestHorizon, time.Now(), lineageTestWindow)
 			assert.Error(t, err)
 			return nil
 		}))
@@ -327,3 +349,131 @@ func TestLockRedshiftTuplesTx_BlocksSameTupleAndSetsReadCommitted(t *testing.T)
 	third.Count = 11
 	assert.NoError(t, store.LockRedshiftTuplesTx(ctx, tx3, []RedshiftTuple{third}), "a different tuple does not block")
 }
+
+// Stamps come from each row's stored updated_at: the trigger overwrites any updated_at an
+// UPDATE sets, so tests cannot place it directly.
+func TestFindOpenRedshiftLineageConflicts_AdoptionWindow(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+	recent := ptrDur(time.Hour)
+	reset := func(t *testing.T) {
+		t.Helper()
+		_, err := store.db.Exec(ctx, `DELETE FROM purchase_executions`)
+		require.NoError(t, err)
+	}
+	updatedAt := func(t *testing.T, id string) time.Time {
+		t.Helper()
+		var ts time.Time
+		require.NoError(t, store.db.QueryRow(ctx, `SELECT updated_at FROM purchase_executions WHERE execution_id = $1`, id).Scan(&ts))
+		return ts
+	}
+	find := func(t *testing.T, self, family string, tp RedshiftTuple, arg *string, stamp time.Time) []RedshiftLineageConflict {
+		t.Helper()
+		var got []RedshiftLineageConflict
+		require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error {
+			var e error
+			got, e = store.FindOpenRedshiftLineageConflictsTx(ctx, tx, self, family, []RedshiftTuple{tp}, arg, lineageTestHorizon, stamp, lineageTestWindow)
+			return e
+		}))
+		return got
+	}
+
+	t.Run("recent_completed_other_family_conflicts", func(t *testing.T) {
+		reset(t)
+		self := seedLineageRow(ctx, t, store, "fam-w1", "running", recent)
+		other := seedLineageRow(ctx, t, store, "fam-w1-other", "completed", recent)
+		got := find(t, self.ExecutionID, "fam-w1", lineageTuple, nil, updatedAt(t, other.ExecutionID))
+		require.Len(t, got, 1)
+		assert.Equal(t, other.ExecutionID, got[0].ExecutionID)
+		assert.Equal(t, "completed", got[0].Status)
+		assert.False(t, got[0].SameFamily)
+	})
+	t.Run("partially_completed_conflicts", func(t *testing.T) {
+		reset(t)
+		self := seedLineageRow(ctx, t, store, "fam-w2", "running", recent)
+		other := seedLineageRow(ctx, t, store, "fam-w2-other", "partially_completed", recent)
+		got := find(t, self.ExecutionID, "fam-w2", lineageTuple, nil, updatedAt(t, other.ExecutionID))
+		assert.Equal(t, []string{other.ExecutionID}, conflictIDs(got))
+		assert.Equal(t, "partially_completed", got[0].Status)
+	})
+	t.Run("revocation_requested_conflicts", func(t *testing.T) {
+		reset(t)
+		self := seedLineageRow(ctx, t, store, "fam-w3", "running", recent)
+		other := seedLineageRow(ctx, t, store, "fam-w3-other", "revocation_requested", recent)
+		got := find(t, self.ExecutionID, "fam-w3", lineageTuple, nil, updatedAt(t, other.ExecutionID))
+		assert.Equal(t, []string{other.ExecutionID}, conflictIDs(got))
+		assert.Equal(t, "revocation_requested", got[0].Status)
+	})
+	t.Run("completed_older_than_window_passes", func(t *testing.T) {
+		reset(t)
+		self := seedLineageRow(ctx, t, store, "fam-w4", "running", recent)
+		other := seedLineageRow(ctx, t, store, "fam-w4-other", "completed", recent)
+		at := updatedAt(t, other.ExecutionID)
+		// updated_at == stamp - window is inside the window; one microsecond later is outside.
+		assert.Len(t, find(t, self.ExecutionID, "fam-w4", lineageTuple, nil, at.Add(lineageTestWindow)), 1)
+		assert.Empty(t, find(t, self.ExecutionID, "fam-w4", lineageTuple, nil, at.Add(lineageTestWindow+time.Microsecond)))
+	})
+	t.Run("same_family_completed_not_conflict", func(t *testing.T) {
+		reset(t)
+		self := seedLineageRow(ctx, t, store, "fam-w5", "running", recent)
+		other := seedLineageRow(ctx, t, store, "fam-w5:acct", "completed", recent)
+		assert.Empty(t, find(t, self.ExecutionID, "fam-w5", lineageTuple, nil, updatedAt(t, other.ExecutionID)))
+	})
+	t.Run("different_count_passes", func(t *testing.T) {
+		reset(t)
+		self := seedLineageRow(ctx, t, store, "fam-w6", "running", recent)
+		other := seedLineageRow(ctx, t, store, "fam-w6-other", "completed", recent)
+		tp := lineageTuple
+		tp.Count = 3
+		assert.Empty(t, find(t, self.ExecutionID, "fam-w6", tp, nil, updatedAt(t, other.ExecutionID)))
+	})
+	t.Run("different_account_passes", func(t *testing.T) {
+		reset(t)
+		a1, a2 := "11111111-1111-1111-1111-111111111111", "22222222-2222-2222-2222-222222222222"
+		for _, a := range []string{a1, a2} {
+			_, err := store.db.Exec(ctx, `INSERT INTO cloud_accounts (id, name, provider, external_id) VALUES ($1::uuid, $1::text, 'aws', $1::text) ON CONFLICT DO NOTHING`, a)
+			require.NoError(t, err)
+		}
+		self := seedLineageRow(ctx, t, store, "fam-w7", "running", recent)
+		other := seedLineageRow(ctx, t, store, "fam-w7-other", "completed", recent)
+		_, err := store.db.Exec(ctx, `UPDATE purchase_executions SET cloud_account_id = $2 WHERE execution_id = $1`, other.ExecutionID, a1)
+		require.NoError(t, err)
+		stamp := updatedAt(t, other.ExecutionID)
+		assert.Empty(t, find(t, self.ExecutionID, "fam-w7", lineageTuple, &a2, stamp))
+		assert.Len(t, find(t, self.ExecutionID, "fam-w7", lineageTuple, &a1, stamp), 1)
+	})
+	t.Run("null_account_wildcard_conflicts", func(t *testing.T) {
+		reset(t)
+		a1 := "11111111-1111-1111-1111-111111111111"
+		_, err := store.db.Exec(ctx, `INSERT INTO cloud_accounts (id, name, provider, external_id) VALUES ($1::uuid, $1::text, 'aws', $1::text) ON CONFLICT DO NOTHING`, a1)
+		require.NoError(t, err)
+		self := seedLineageRow(ctx, t, store, "fam-w8", "running", recent)
+		other := seedLineageRow(ctx, t, store, "fam-w8-other", "completed", recent)
+		_, err = store.db.Exec(ctx, `UPDATE purchase_executions SET cloud_account_id = $2 WHERE execution_id = $1`, other.ExecutionID, a1)
+		require.NoError(t, err)
+		// The candidate has no account at all, so it matches any row account.
+		assert.Equal(t, []string{other.ExecutionID}, conflictIDs(find(t, self.ExecutionID, "fam-w8", lineageTuple, nil, updatedAt(t, other.ExecutionID))))
+	})
+	t.Run("completed_at_earlier_than_updated_at_still_conflicts", func(t *testing.T) {
+		reset(t)
+		self := seedLineageRow(ctx, t, store, "fam-w9", "running", recent)
+		other := seedLineageRow(ctx, t, store, "fam-w9-other", "completed", recent)
+		_, err := store.db.Exec(ctx, `UPDATE purchase_executions SET completed_at = now() - interval '2 hours' WHERE execution_id = $1`, other.ExecutionID)
+		require.NoError(t, err)
+		assert.Equal(t, []string{other.ExecutionID}, conflictIDs(find(t, self.ExecutionID, "fam-w9", lineageTuple, nil, updatedAt(t, other.ExecutionID))))
+	})
+	t.Run("completed_after_old_stamp_conflicts", func(t *testing.T) {
+		reset(t)
+		self := seedLineageRow(ctx, t, store, "fam-w10", "running", ptrDur(48*time.Hour))
+		// INSERT keeps the old updated_at (the trigger fires on UPDATE only): a buy 3h ago,
+		// inside this old stamp's window but outside a now()-anchored one.
+		stamp := time.Now().Add(-3*time.Hour - 5*time.Minute)
+		var id string
+		require.NoError(t, store.db.QueryRow(ctx, `
+			INSERT INTO purchase_executions (status, step_number, scheduled_date, idempotency_key, first_dispatch_at, updated_at, recommendations)
+			VALUES ('completed', 1, NOW(), 'fam-w10-other', $1, $2, $3::jsonb)
+			RETURNING execution_id::text`, stamp, stamp.Add(10*time.Minute),
+			`[{"provider":"aws","service":"redshift","region":"us-east-1","resource_type":"ra3.xlplus","term":1,"payment":"all-upfront","count":2,"selected":true}]`).Scan(&id))
+		assert.Equal(t, []string{id}, conflictIDs(find(t, self.ExecutionID, "fam-w10", lineageTuple, nil, stamp)))
+	})
+}

internal/mocks/stores.go

@@ -576,9 +576,9 @@ func (m *MockConfigStore) LockRedshiftTuplesTx(ctx context.Context, tx pgx.Tx, t
 }
 
 // FindOpenRedshiftLineageConflictsTx mocks the Redshift lineage conflict query (no default: testify panics if unexpected).
-func (m *MockConfigStore) FindOpenRedshiftLineageConflictsTx(ctx context.Context, tx pgx.Tx, selfID, family string, tuples []config.RedshiftTuple, accountID *string, horizon time.Duration) ([]config.RedshiftLineageConflict, error) {
-	m.record("FindOpenRedshiftLineageConflictsTx", ctx, tx, selfID, family, tuples, accountID, horizon)
-	args := m.Called(ctx, tx, selfID, family, tuples, accountID, horizon)
+func (m *MockConfigStore) FindOpenRedshiftLineageConflictsTx(ctx context.Context, tx pgx.Tx, selfID, family string, tuples []config.RedshiftTuple, accountID *string, horizon time.Duration, stamp time.Time, window time.Duration) ([]config.RedshiftLineageConflict, error) {
+	m.record("FindOpenRedshiftLineageConflictsTx", ctx, tx, selfID, family, tuples, accountID, horizon, stamp, window)
+	args := m.Called(ctx, tx, selfID, family, tuples, accountID, horizon, stamp, window)
 	if args.Get(0) == nil {
 		return nil, args.Error(1)
 	}

internal/purchase/armed_redrive_test.go

@@ -113,7 +113,7 @@ func armedHarness(t *testing.T, serviceType common.ServiceType) (*Manager, *Mock
 		Return(config.StampResult{At: &stampedAt, WithinHorizon: true}, nil).Maybe()
 	store.On("LockRedshiftTuplesTx", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe()
 	store.On("FindOpenRedshiftLineageConflictsTx", mock.Anything, mock.Anything, mock.Anything, mock.Anything,
-		mock.Anything, mock.Anything, mock.Anything).Return(nil, nil).Maybe()
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	email.On("SendPurchaseConfirmation", mock.Anything, mock.AnythingOfType("email.NotificationData")).Return(nil).Maybe()
 
 	factory.On("CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything).Return(prov, nil).Maybe()

internal/purchase/errnotes.go

@@ -34,6 +34,7 @@ var serverNotePatterns = []string{
 	regexp.QuoteMeta(redshiftConflictRefusal),
 	regexp.QuoteMeta(redshiftHorizonRefusal),
 	regexp.QuoteMeta(redshiftNoStampRefusal),
+	regexp.QuoteMeta(redshiftRecentBuyRefusal),
 	`EC2 recommendation for \S+ has zonal scope but no AvailabilityZone: refusing to buy a zonal RI in an arbitrary AZ; re-fetch the recommendation`,
 	`zonal RI offering needs an availability zone; re-fetch the recommendation`,
 	`email notifier not configured for this deployment`,

internal/purchase/errnotes_test.go

@@ -53,6 +53,7 @@ func TestSanitizeStoredError_AllowlistedNotesPassThroughVerbatim(t *testing.T) {
 		redshiftConflictRefusal,
 		redshiftHorizonRefusal,
 		redshiftNoStampRefusal,
+		redshiftRecentBuyRefusal,
 		"EC2 recommendation for m5.large has zonal scope but no AvailabilityZone: refusing to buy a zonal RI in an arbitrary AZ; re-fetch the recommendation",
 		"zonal RI offering needs an availability zone; re-fetch the recommendation",
 		"email notifier not configured for this deployment",
@@ -205,7 +206,7 @@ func TestSanitizeStoredError_ResolveNoteExchangeIDIsClosed(t *testing.T) {
 // The Redshift gate texts are fixed: an appended execution id or raw text must
 // not pass verbatim.
 func TestSanitizeStoredError_RedshiftGateRefusalsAreClosed(t *testing.T) {
-	for _, text := range []string{redshiftConflictRefusal, redshiftHorizonRefusal, redshiftNoStampRefusal} {
+	for _, text := range []string{redshiftConflictRefusal, redshiftHorizonRefusal, redshiftNoStampRefusal, redshiftRecentBuyRefusal} {
 		assert.Equal(t, text, SanitizeStoredError(text))
 		for _, variant := range []string{
 			text + " (execution 00000000-0000-0000-0000-000000000000, running)",