fix(purchase): refuse a Redshift dispatch near another lineage's possible buy

internal/config/store_postgres_redshift_lineage.go

@@ -71,6 +71,16 @@ func redshiftFailedClause(settleReady bool) string {
 		            AND e.first_dispatch_at > now() - make_interval(secs => $3)` + notSettled + `)`
 }
 
+// redshiftRecentBuyClause matches other-family rows that may have bought (finished, or failed
+// without the no-submission proof, settled or not) and were last written within $7 seconds of
+// the dispatch stamp $6: go could adopt their node. It is also the recent_buy column.
+func redshiftRecentBuyClause() string {
+	return `(split_part(COALESCE(NULLIF(e.idempotency_key, ''), e.execution_id::text), ':', 1) <> $2
+		            AND e.updated_at >= $6::timestamptz - make_interval(secs => $7)
+		            AND (e.status IN ('completed', 'partially_completed', 'revocation_requested')
+		                 OR (e.status = 'failed' AND NOT e.no_submission)))`
+}
+
 // redshiftSameFamilySettleClause matches a same-family row whose own attempt may have bought and
 // was last written within $8 seconds: its node can still be missing from AWS describe results.
 // A linked failed row already passed the Retry settle (and Retry bumps its updated_at).
@@ -83,20 +93,23 @@ func redshiftSameFamilySettleClause() string {
 
 // RedshiftLineageConflict is an open execution whose Redshift rec matches a
 // candidate tuple. SameFamily is true when it shares the candidate's lineage
-// root (idempotency key up to the first colon).
+// root (idempotency key up to the first colon). RecentBuy marks an other-family row that may
+// have bought and was written inside the adoption window.
 type RedshiftLineageConflict struct {
 	ExecutionID     string
 	Status          string
 	FirstDispatchAt *time.Time
 	SameFamily      bool
+	RecentBuy       bool
 	Error           string
 	Recommendations []RecommendationRecord
 }
 
 // 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 within horizon of their stamp. Other-family
-// finished rows within window of stamp; other-family retryable failed rows within horizon.
+// rows that may have bought (finished, or failed without the no-submission proof) within window
+// of stamp; other-family retryable failed rows within horizon.
 // Same-family rows: running, or failed (unlinked) or finished after a possible send, within settle.
 func (s *PostgresStore) FindOpenRedshiftLineageConflictsTx(ctx context.Context, tx pgx.Tx, selfID, family string, tuples []RedshiftTuple, accountID *string, horizon time.Duration, stamp time.Time, window, settle time.Duration) ([]RedshiftLineageConflict, error) {
 	if family == "" {
@@ -116,7 +129,8 @@ func (s *PostgresStore) FindOpenRedshiftLineageConflictsTx(ctx context.Context,
 	}
 	rows, err := tx.Query(ctx, `
 		SELECT e.execution_id, e.status, e.first_dispatch_at, COALESCE(e.error, ''), e.recommendations,
-		       split_part(COALESCE(NULLIF(e.idempotency_key, ''), e.execution_id::text), ':', 1) = $2
+		       split_part(COALESCE(NULLIF(e.idempotency_key, ''), e.execution_id::text), ':', 1) = $2,
+		       `+redshiftRecentBuyClause()+`
 		FROM purchase_executions e
 		WHERE e.execution_id <> $1
 		  AND e.first_dispatch_at IS NOT NULL
@@ -125,9 +139,7 @@ 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))
+		        OR `+redshiftRecentBuyClause()+`
 		        OR `+redshiftFailedClause(settleReady)+`
 		        OR `+redshiftSameFamilySettleClause()+`
 		      )
@@ -154,7 +166,7 @@ func (s *PostgresStore) FindOpenRedshiftLineageConflictsTx(ctx context.Context,
 	for rows.Next() {
 		var c RedshiftLineageConflict
 		var recsJSON []byte
-		if err := rows.Scan(&c.ExecutionID, &c.Status, &c.FirstDispatchAt, &c.Error, &recsJSON, &c.SameFamily); err != nil {
+		if err := rows.Scan(&c.ExecutionID, &c.Status, &c.FirstDispatchAt, &c.Error, &recsJSON, &c.SameFamily, &c.RecentBuy); err != nil {
 			return nil, fmt.Errorf("scan redshift lineage conflict: %w", err)
 		}
 		if err := json.Unmarshal(recsJSON, &c.Recommendations); err != nil {

internal/config/store_postgres_redshift_lineage_integration_test.go

@@ -44,11 +44,17 @@ func seedLineageRow(ctx context.Context, t *testing.T, store *PostgresStore, fam
 }
 
 func findLineageConflicts(ctx context.Context, t *testing.T, store *PostgresStore, self, family string) []RedshiftLineageConflict {
+	t.Helper()
+	return findLineageConflictsAt(ctx, t, store, self, family, time.Now())
+}
+
+// findLineageConflictsAt runs the query with an explicit dispatch stamp.
+func findLineageConflictsAt(ctx context.Context, t *testing.T, store *PostgresStore, self, family string, stamp time.Time) []RedshiftLineageConflict {
 	t.Helper()
 	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, time.Now(), lineageTestWindow, lineageTestSettle)
+		got, err = store.FindOpenRedshiftLineageConflictsTx(ctx, tx, self, family, []RedshiftTuple{lineageTuple}, nil, lineageTestHorizon, stamp, lineageTestWindow, lineageTestSettle)
 		return err
 	}))
 	return got
@@ -398,6 +404,7 @@ func TestFindOpenRedshiftLineageConflicts_AdoptionWindow(t *testing.T) {
 		assert.Equal(t, other.ExecutionID, got[0].ExecutionID)
 		assert.Equal(t, "completed", got[0].Status)
 		assert.False(t, got[0].SameFamily)
+		assert.True(t, got[0].RecentBuy)
 	})
 	t.Run("partially_completed_conflicts", func(t *testing.T) {
 		reset(t)
@@ -505,6 +512,8 @@ func TestFindOpenRedshiftLineageConflicts_FailedRows(t *testing.T) {
 	store := setupLadderStore(ctx, t)
 	recent, old := ptrDur(time.Hour), ptrDur(48*time.Hour)
 	// seedFailed seeds an other-family failed row and applies a column update to it.
+	// pastWindow is a stamp whose adoption window ended before the rows just written.
+	pastWindow := func() time.Time { return time.Now().Add(lineageTestWindow + time.Hour) }
 	seedFailed := func(t *testing.T, family string, stampAge *time.Duration, set string) (self, failed *PurchaseExecution) {
 		t.Helper()
 		_, err := store.db.Exec(ctx, `DELETE FROM purchase_executions`)
@@ -524,27 +533,35 @@ func TestFindOpenRedshiftLineageConflicts_FailedRows(t *testing.T) {
 
 	t.Run("failed_unlinked_conflicts", func(t *testing.T) {
 		self, failed := seedFailed(t, "fam-f1", recent, "")
-		assert.Equal(t, []string{failed.ExecutionID}, conflictIDs(findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0")))
+		got := findLineageConflictsAt(ctx, t, store, self.ExecutionID, "fam-f0", pastWindow())
+		assert.Equal(t, []string{failed.ExecutionID}, conflictIDs(got))
+		assert.False(t, got[0].RecentBuy, "the row comes from the failed clause")
 	})
 	t.Run("failed_attempt1_no_submission_conflicts", func(t *testing.T) {
 		self, failed := seedFailed(t, "fam-f2", recent, "no_submission = true, retry_attempt_n = 1")
 		assert.Equal(t, []string{failed.ExecutionID}, conflictIDs(findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0")))
 	})
 	t.Run("failed_linked_excluded", func(t *testing.T) {
 		self, _ := seedFailed(t, "fam-f3", recent, "retry_execution_id = $2")
-		assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0"))
+		assert.Empty(t, findLineageConflictsAt(ctx, t, store, self.ExecutionID, "fam-f0", pastWindow()))
 	})
 	t.Run("failed_attempt0_no_submission_excluded", func(t *testing.T) {
 		self, _ := seedFailed(t, "fam-f4", recent, "no_submission = true")
-		assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0"))
+		assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0"), "no recent buy either: the no-submission proof")
 	})
 	t.Run("failed_older_than_H_excluded", func(t *testing.T) {
 		self, _ := seedFailed(t, "fam-f5", old, "")
-		assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0"))
+		assert.Empty(t, findLineageConflictsAt(ctx, t, store, self.ExecutionID, "fam-f0", pastWindow()))
 	})
 	t.Run("failed_settled_excluded", func(t *testing.T) {
 		self, _ := seedFailed(t, "fam-f6", recent, "outcome_settled_at = now(), outcome_settled_by = 'op', outcome_settlement = 'settled'")
-		assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0"))
+		assert.Empty(t, findLineageConflictsAt(ctx, t, store, self.ExecutionID, "fam-f0", pastWindow()))
+	})
+	t.Run("failed_settled_inside_adoption_window_conflicts_as_recent_buy", func(t *testing.T) {
+		self, failed := seedFailed(t, "fam-f6b", recent, "outcome_settled_at = now(), outcome_settled_by = 'op', outcome_settlement = 'settled'")
+		got := findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0")
+		assert.Equal(t, []string{failed.ExecutionID}, conflictIDs(got))
+		assert.True(t, got[0].RecentBuy)
 	})
 	t.Run("failed_settled_still_conflicts_before_the_settlement_migration", func(t *testing.T) {
 		// Prod may run this code before migration 000155: the settled filter must drop out, not error.
@@ -558,7 +575,9 @@ func TestFindOpenRedshiftLineageConflicts_FailedRows(t *testing.T) {
 		})
 		_, err := store.db.Exec(ctx, `UPDATE schema_migrations SET version = $1, dirty = false`, settlementMigrationVersion-1)
 		require.NoError(t, err)
-		assert.Equal(t, []string{failed.ExecutionID}, conflictIDs(findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0")))
+		got := findLineageConflictsAt(ctx, t, store, self.ExecutionID, "fam-f0", pastWindow())
+		assert.Equal(t, []string{failed.ExecutionID}, conflictIDs(got))
+		assert.False(t, got[0].RecentBuy, "the row comes from the failed clause")
 	})
 	t.Run("same_family_failed_inside_settle_conflicts", func(t *testing.T) {
 		self, failed := seedFailed(t, "fam-f0:acct", recent, "")
@@ -587,15 +606,17 @@ func TestFindOpenRedshiftLineageConflicts_FailedRows(t *testing.T) {
 		self, _ := seedFailed(t, "fam-f0:acct", recent, "no_submission = true, retry_attempt_n = 1")
 		assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0"))
 	})
-	t.Run("same_family_completed_inside_settle_conflicts", func(t *testing.T) {
-		_, err := store.db.Exec(ctx, `DELETE FROM purchase_executions`)
-		require.NoError(t, err)
-		self := seedLineageRow(ctx, t, store, "fam-f0", "running", recent)
-		done := seedLineageRow(ctx, t, store, "fam-f0:acct", "completed", recent)
-		got := findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0")
-		assert.Equal(t, []string{done.ExecutionID}, conflictIDs(got))
-		assert.True(t, got[0].SameFamily)
-	})
+	for _, status := range []string{"completed", "partially_completed", "revocation_requested"} {
+		t.Run("same_family_"+status+"_inside_settle_conflicts", func(t *testing.T) {
+			_, err := store.db.Exec(ctx, `DELETE FROM purchase_executions`)
+			require.NoError(t, err)
+			self := seedLineageRow(ctx, t, store, "fam-f0", "running", recent)
+			done := seedLineageRow(ctx, t, store, "fam-f0:acct", status, recent)
+			got := findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0")
+			assert.Equal(t, []string{done.ExecutionID}, conflictIDs(got))
+			assert.True(t, got[0].SameFamily)
+		})
+	}
 	t.Run("same_family_completed_after_settle_passes", func(t *testing.T) {
 		_, err := store.db.Exec(ctx, `DELETE FROM purchase_executions`)
 		require.NoError(t, err)
@@ -617,3 +638,56 @@ func TestFindOpenRedshiftLineageConflicts_FailedRows(t *testing.T) {
 		assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f0"))
 	})
 }
+
+func TestFindOpenRedshiftLineageConflicts_RecentBuyKinds(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+	recent := ptrDur(time.Hour)
+	settled := "outcome_settled_at = now(), outcome_settled_by = 'op', outcome_settlement = 'settled'"
+	for _, tc := range []struct {
+		name, status, set string
+		recentBuy         bool
+	}{
+		{"completed", "completed", "", true},
+		{"partially_completed", "partially_completed", "", true},
+		{"revocation_requested", "revocation_requested", "", true},
+		{"failed_unsettled", "failed", "", true},
+		{"failed_settled", "failed", settled, true},
+		{"failed_retry_linked", "failed", "retry_execution_id = $2", true},
+		{"failed_no_submission", "failed", "no_submission = true", false},
+		{"failed_no_submission_attempt1", "failed", "no_submission = true, retry_attempt_n = 1", false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := store.db.Exec(ctx, `DELETE FROM purchase_executions`)
+			require.NoError(t, err)
+			self := seedLineageRow(ctx, t, store, "fam-rb", "running", recent)
+			other := seedLineageRow(ctx, t, store, "fam-rb-other", tc.status, recent)
+			if tc.set != "" {
+				args := []any{other.ExecutionID}
+				if strings.Contains(tc.set, "$2") {
+					args = append(args, self.ExecutionID)
+				}
+				_, err = store.db.Exec(ctx, `UPDATE purchase_executions SET `+tc.set+` WHERE execution_id = $1`, args...)
+				require.NoError(t, err)
+			}
+			var at time.Time
+			require.NoError(t, store.db.QueryRow(ctx, `SELECT updated_at FROM purchase_executions WHERE execution_id = $1`, other.ExecutionID).Scan(&at))
+
+			var inside RedshiftLineageConflict
+			for _, c := range findLineageConflictsAt(ctx, t, store, self.ExecutionID, "fam-rb", at) {
+				if c.ExecutionID == other.ExecutionID {
+					inside = c
+				}
+			}
+			assert.Equal(t, tc.recentBuy, inside.RecentBuy, "inside the window")
+			if tc.recentBuy {
+				assert.Equal(t, other.ExecutionID, inside.ExecutionID, "inside the window")
+			}
+
+			// One microsecond past the window edge the row is no longer a recent buy.
+			for _, c := range findLineageConflictsAt(ctx, t, store, self.ExecutionID, "fam-rb", at.Add(lineageTestWindow+time.Microsecond)) {
+				assert.False(t, c.RecentBuy, "outside the window")
+			}
+		})
+	}
+}

internal/purchase/errnotes.go

@@ -41,6 +41,7 @@ var serverNotePatterns = []string{
 	regexp.QuoteMeta(redshiftFailedConflictRefusal),
 	regexp.QuoteMeta(redshiftNoJournalRefusal),
 	regexp.QuoteMeta(redshiftRecentBuyRefusal),
+	regexp.QuoteMeta(redshiftRecentBuyRefusalLegacy),
 	regexp.QuoteMeta(redshiftSettleRefusal),
 	`EC2 recommendation for ` + resourceTypePattern + ` 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`,

internal/purchase/errnotes_test.go

@@ -60,6 +60,7 @@ func TestSanitizeStoredError_AllowlistedNotesPassThroughVerbatim(t *testing.T) {
 		redshiftFailedConflictRefusal,
 		redshiftNoJournalRefusal,
 		redshiftRecentBuyRefusal,
+		redshiftRecentBuyRefusalLegacy,
 		redshiftSettleRefusal,
 		"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",