From 51d86de89cbea4a46e98f216068d79735dcf9582 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 19:21:28 +0100 Subject: [PATCH 1/4] fix(purchase): refuse a Redshift dispatch inside the adoption window Part of #702. Before buying, cloud-commitments-go adopts any owned node with the same offering and count that started at or after RequestedAt minus 5 minutes. So a new purchase dispatched shortly after another lineage's completed buy of the same tuple adopted that node and bought nothing: an under-buy, with one commitment counted twice. - The dispatch gate also counts another lineage's completed, partially completed or revocation-requested buy of the tuple as a conflict while its last save (updated_at, database clock) is within 10 minutes of this dispatch's stamp: go's 5-minute skew plus a 5-minute clock margin. The refusal happens before any purchase call, with a fixed text that names no other execution. - The conflict query now matches a candidate on the execution's account first, as dispatch does, so a differing rec account cannot hide a conflict. Co-Authored-By: Claude Opus 5.5 --- internal/config/interfaces.go | 2 +- .../config/store_postgres_redshift_lineage.go | 23 +-- ...tgres_redshift_lineage_integration_test.go | 162 +++++++++++++++++- internal/mocks/stores.go | 6 +- internal/purchase/armed_redrive_test.go | 2 +- internal/purchase/errnotes.go | 1 + internal/purchase/errnotes_test.go | 3 +- .../redshift_gate_integration_test.go | 62 +++++++ internal/purchase/redshift_gate_test.go | 39 ++++- internal/purchase/redshift_lineage.go | 28 ++- internal/purchase/requested_at_test.go | 2 +- 11 files changed, 298 insertions(+), 32 deletions(-) diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 28a491571..cae5f0833 100644 --- a/internal/config/interfaces.go +++ b/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 diff --git a/internal/config/store_postgres_redshift_lineage.go b/internal/config/store_postgres_redshift_lineage.go index 03bb113dd..268af6840 100644 --- a/internal/config/store_postgres_redshift_lineage.go +++ b/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) } diff --git a/internal/config/store_postgres_redshift_lineage_integration_test.go b/internal/config/store_postgres_redshift_lineage_integration_test.go index 97da919af..0e285b71e 100644 --- a/internal/config/store_postgres_redshift_lineage_integration_test.go +++ b/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))) + }) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index cde58ae63..c83968e59 100644 --- a/internal/mocks/stores.go +++ b/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) } diff --git a/internal/purchase/armed_redrive_test.go b/internal/purchase/armed_redrive_test.go index ce7fbdf24..ddda6e0d6 100644 --- a/internal/purchase/armed_redrive_test.go +++ b/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() diff --git a/internal/purchase/errnotes.go b/internal/purchase/errnotes.go index 355c5bdce..412893f78 100644 --- a/internal/purchase/errnotes.go +++ b/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`, diff --git a/internal/purchase/errnotes_test.go b/internal/purchase/errnotes_test.go index c26a7b7f0..d8c965f82 100644 --- a/internal/purchase/errnotes_test.go +++ b/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)", diff --git a/internal/purchase/redshift_gate_integration_test.go b/internal/purchase/redshift_gate_integration_test.go index e0f699d33..cb62c9361 100644 --- a/internal/purchase/redshift_gate_integration_test.go +++ b/internal/purchase/redshift_gate_integration_test.go @@ -181,3 +181,65 @@ func (f *redshiftGateFixture) raceOnlyOneBuys(ctx context.Context, t *testing.T, assert.Equal(t, 1, winners) assert.Equal(t, 1, losers) } + +// finishedRow saves a completed per-account row in its own lineage with a stamp. +func (f *redshiftGateFixture) finishedRow(ctx context.Context, t *testing.T, key string) *config.PurchaseExecution { + t.Helper() + row := f.redshiftRowKeyed(ctx, t, config.StatusCompleted, key, false) + _, err := f.pool.Exec(ctx, `UPDATE purchase_executions SET first_dispatch_at = now() WHERE execution_id = $1`, row.ExecutionID) + require.NoError(t, err) + return row +} + +// A node another lineage bought seconds ago is indistinguishable to AWS from a +// new buy, so the dispatch is refused before any provider call. +func TestRedshiftDispatchGate_RecentCompletedBuyRefusedBeforeSubmission(t *testing.T) { + ctx := context.Background() + f := newRedshiftGateFixture(ctx, t) + f.finishedRow(ctx, t, uuid.New().String()+":"+f.accounts[0].ID) + refused := f.redshiftRow(ctx, t, "approved") + + claimed, _ := f.mgr.claimAndExecute(ctx, rowStatusCopy(refused)) + require.True(t, claimed) + + assert.Zero(t, f.bought.Load(), "a refused lineage never reaches the provider") + got := rowStatus(ctx, t, f.rampStepFixture, refused.ExecutionID) + assert.Equal(t, config.StatusFailed, got.Status) + assert.True(t, got.NoSubmission) + assert.Equal(t, redshiftRecentBuyRefusal, SanitizeStoredError(got.Error), "users see the fixed text verbatim") + assert.Nil(t, f.stampOf(ctx, t, refused.ExecutionID), "the refusal rolled the stamp back") +} + +// A lineage whose stamp is more than the window after the other buy finished +// dispatches: the stamp is seeded in the future of the finished row's updated_at. +func TestRedshiftDispatchGate_SecondLineageAfterWindowDispatches(t *testing.T) { + ctx := context.Background() + f := newRedshiftGateFixture(ctx, t) + f.finishedRow(ctx, t, uuid.New().String()+":"+f.accounts[0].ID) + later := f.redshiftRow(ctx, t, "approved") + _, err := f.pool.Exec(ctx, `UPDATE purchase_executions SET first_dispatch_at = now() + $2 * interval '1 second' WHERE execution_id = $1`, + later.ExecutionID, (redshiftAdoptionWindow + time.Minute).Seconds()) + require.NoError(t, err) + + claimed, _ := f.mgr.claimAndExecute(ctx, rowStatusCopy(later)) + require.True(t, claimed) + + assert.Equal(t, int32(1), f.bought.Load()) + assert.Equal(t, config.StatusCompleted, rowStatus(ctx, t, f.rampStepFixture, later.ExecutionID).Status) +} + +// A completed row of the same family (a retry's earlier per-account row) is +// not another lineage, so the retry successor is not refused. +func TestRedshiftDispatchGate_SameFamilyCompletedRowNotRefused(t *testing.T) { + ctx := context.Background() + f := newRedshiftGateFixture(ctx, t) + family := uuid.New().String() + f.finishedRow(ctx, t, family+":"+f.accounts[0].ID) + successor := f.redshiftRowKeyed(ctx, t, "approved", family, true) + + claimed, _ := f.mgr.claimAndExecute(ctx, rowStatusCopy(successor)) + require.True(t, claimed) + + assert.NotEqual(t, redshiftRecentBuyRefusal, SanitizeStoredError(rowStatus(ctx, t, f.rampStepFixture, successor.ExecutionID).Error)) + assert.Positive(t, f.bought.Load(), "the same-family retry reaches the provider") +} diff --git a/internal/purchase/redshift_gate_test.go b/internal/purchase/redshift_gate_test.go index 3bff68a43..bec8778b1 100644 --- a/internal/purchase/redshift_gate_test.go +++ b/internal/purchase/redshift_gate_test.go @@ -25,6 +25,8 @@ type gateRun struct { finds []config.RedshiftTuple family string locked []config.RedshiftTuple + stamp time.Time + window time.Duration } // runGate drives the gate with a stamp result and the conflicts the query returns. @@ -36,8 +38,13 @@ func runGate(t *testing.T, exec *config.PurchaseExecution, stamp config.StampRes Run(func(a mock.Arguments) { run.locked = a.Get(2).([]config.RedshiftTuple) }).Return(nil) store.On("StampFirstDispatchTx", mock.Anything, mock.Anything, exec.ExecutionID, redshiftRedriveHorizon).Return(stamp, nil) store.On("FindOpenRedshiftLineageConflictsTx", mock.Anything, mock.Anything, exec.ExecutionID, mock.Anything, - mock.Anything, mock.Anything, redshiftRedriveHorizon). - Run(func(a mock.Arguments) { run.family = a.String(3); run.finds = a.Get(4).([]config.RedshiftTuple) }). + mock.Anything, mock.Anything, redshiftRedriveHorizon, mock.Anything, mock.Anything). + Run(func(a mock.Arguments) { + run.family = a.String(3) + run.finds = a.Get(4).([]config.RedshiftTuple) + run.stamp = a.Get(7).(time.Time) + run.window = a.Get(8).(time.Duration) + }). Return(conflicts, nil).Maybe() run.err = (&Manager{config: store}).stampRedshiftDispatch(context.Background(), exec) return run @@ -72,6 +79,30 @@ func TestRedshiftDispatchGate_ConflictRefusesBeforeSubmission(t *testing.T) { } } +func TestRedshiftDispatchGate_RecentFinishedBuyRefusedWithFixedText(t *testing.T) { + for _, status := range []string{config.StatusCompleted, config.StatusPartiallyCompleted, config.StatusRevocationRequested} { + exec := gateExec(0, gateRec("acct-A", 2)) + stamp := stamped() + run := runGate(t, exec, stamp, []config.RedshiftLineageConflict{{ExecutionID: "other-id", Status: status}}) + + require.Error(t, run.err, status) + assert.True(t, isBeforeSubmission(run.err), status) + assert.Equal(t, redshiftRecentBuyRefusal, run.err.Error(), "the text names no execution") + assert.NotContains(t, run.err.Error(), "other-id") + assert.Nil(t, exec.FirstDispatchAt, "a refused dispatch records no stamp") + assert.True(t, stamp.At.Equal(run.stamp), "the window is anchored on the lineage stamp") + assert.Equal(t, redshiftAdoptionWindow, run.window) + } +} + +func TestRedshiftDispatchGate_FinishedConflictWinsOverOpenOne(t *testing.T) { + run := runGate(t, gateExec(0, gateRec("acct-A", 2)), stamped(), []config.RedshiftLineageConflict{ + {ExecutionID: "open", Status: "running"}, {ExecutionID: "done", Status: config.StatusCompleted}}) + + require.Error(t, run.err) + assert.Equal(t, redshiftRecentBuyRefusal, run.err.Error()) +} + func TestRedshiftDispatchGate_QueriesOnlySelectedRedshiftTuples(t *testing.T) { unselected := gateRec("acct-A", 7) unselected.Selected = false @@ -142,7 +173,7 @@ func TestRedshiftDispatchGate_StoreErrorsByAttempt(t *testing.T) { store.On("LockRedshiftTuplesTx", mock.Anything, mock.Anything, mock.Anything).Return(lockErr) store.On("StampFirstDispatchTx", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(stamped(), nil).Maybe() store.On("FindOpenRedshiftLineageConflictsTx", mock.Anything, mock.Anything, mock.Anything, mock.Anything, - mock.Anything, mock.Anything, mock.Anything).Return(nil, findErr).Maybe() + mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, findErr).Maybe() exec := gateExec(attempt, gateRec("acct-A", 2)) err := (&Manager{config: store}).stampRedshiftDispatch(context.Background(), exec) @@ -161,7 +192,7 @@ func TestRedshiftDispatchGate_CallOrderIsLockStampFind(t *testing.T) { store.On("LockRedshiftTuplesTx", mock.Anything, mock.Anything, mock.Anything).Run(note("lock")).Return(nil) store.On("StampFirstDispatchTx", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Run(note("stamp")).Return(stamped(), nil) store.On("FindOpenRedshiftLineageConflictsTx", mock.Anything, mock.Anything, mock.Anything, mock.Anything, - mock.Anything, mock.Anything, mock.Anything).Run(note("find")).Return(nil, nil) + mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Run(note("find")).Return(nil, nil) require.NoError(t, (&Manager{config: store}).stampRedshiftDispatch(context.Background(), gateExec(0, gateRec("acct-A", 2)))) assert.Equal(t, []string{"lock", "stamp", "find"}, order) diff --git a/internal/purchase/redshift_lineage.go b/internal/purchase/redshift_lineage.go index 0094e5f88..ee9b5c562 100644 --- a/internal/purchase/redshift_lineage.go +++ b/internal/purchase/redshift_lineage.go @@ -43,11 +43,16 @@ var afterRedshiftStampHook func() // Fixed refusal texts: the NULL-account wildcard can match another account's row, // so they never name another execution. errnotes.go allowlists each one. const ( - redshiftConflictRefusal = "refusing to dispatch: another open purchase of the same Redshift reservation (region, node type, term, payment, count) may still complete, so wait for it to finish and then retry" - redshiftHorizonRefusal = "refusing to dispatch: this Redshift purchase was first dispatched more than 24h ago, so check the account in AWS and start a new purchase" - redshiftNoStampRefusal = "refusing to dispatch: this Redshift purchase has no first dispatch time recorded, so check the account in AWS and start a new purchase" + redshiftConflictRefusal = "refusing to dispatch: another open purchase of the same Redshift reservation (region, node type, term, payment, count) may still complete, so wait for it to finish and then retry" + redshiftHorizonRefusal = "refusing to dispatch: this Redshift purchase was first dispatched more than 24h ago, so check the account in AWS and start a new purchase" + redshiftNoStampRefusal = "refusing to dispatch: this Redshift purchase has no first dispatch time recorded, so check the account in AWS and start a new purchase" + redshiftRecentBuyRefusal = "refusing to dispatch: another purchase of the same Redshift reservation (region, node type, term, payment, count) completed too close to this purchase's first dispatch for AWS to tell the two apart, so start a new purchase at least 10 minutes after the other one completed" ) +// redshiftAdoptionWindow mirrors cloud-commitments-go's 5m Redshift dedupe skew +// (providers/aws/services/redshift/dedupe.go redshiftClockSkew) plus a 5m clock margin. +const redshiftAdoptionWindow = 10 * time.Minute + // errRedshiftPastHorizon marks a stamp older than the horizon. It is never // beforeSubmission: an old stamp proves an earlier dispatch may have landed. var errRedshiftPastHorizon = errors.New(redshiftHorizonRefusal) @@ -119,10 +124,16 @@ func (m *Manager) redshiftGate(ctx context.Context, tx pgx.Tx, exec *config.Purc afterRedshiftStampHook() } family, _, _ := strings.Cut(idempotencyLineageKey(exec), ":") - conflicts, err := m.config.FindOpenRedshiftLineageConflictsTx(ctx, tx, exec.ExecutionID, family, tuples, exec.CloudAccountID, redshiftRedriveHorizon) + conflicts, err := m.config.FindOpenRedshiftLineageConflictsTx(ctx, tx, exec.ExecutionID, family, tuples, exec.CloudAccountID, redshiftRedriveHorizon, *res.At, redshiftAdoptionWindow) if err != nil { return res, err } + for _, c := range conflicts { + if redshiftFinishedStatus(c.Status) { + logging.Warnf("Redshift gate: execution %s refused, another purchase of the tuple finished within the adoption window (execution %s, %s)", exec.ExecutionID, c.ExecutionID, c.Status) + return res, beforeSubmission(errors.New(redshiftRecentBuyRefusal)) + } + } if len(conflicts) > 0 { logging.Warnf("Redshift gate: execution %s refused, conflicts with execution %s (%s)", exec.ExecutionID, conflicts[0].ExecutionID, conflicts[0].Status) return res, beforeSubmission(errors.New(redshiftConflictRefusal)) @@ -130,6 +141,15 @@ func (m *Manager) redshiftGate(ctx context.Context, tx pgx.Tx, exec *config.Purc return res, nil } +// redshiftFinishedStatus reports the statuses whose node may already sit in AWS. +func redshiftFinishedStatus(status string) bool { + switch status { + case config.StatusCompleted, config.StatusPartiallyCompleted, config.StatusRevocationRequested: + return true + } + return false +} + // withDispatchTime defaults RequestedAt to now. A Redshift rec must arrive // with the lineage stamp: defaulting would silently move go's dedupe window. func withDispatchTime(rec config.RecommendationRecord, opts common.PurchaseOptions) (common.PurchaseOptions, error) { diff --git a/internal/purchase/requested_at_test.go b/internal/purchase/requested_at_test.go index 8e38e28e2..24eb23b77 100644 --- a/internal/purchase/requested_at_test.go +++ b/internal/purchase/requested_at_test.go @@ -165,7 +165,7 @@ func runRedshiftFinalize(t *testing.T, exec *config.PurchaseExecution, accounts if stamp != nil { mockStore.On("LockRedshiftTuplesTx", mock.Anything, mock.Anything, mock.Anything).Return(nil) mockStore.On("FindOpenRedshiftLineageConflictsTx", mock.Anything, mock.Anything, mock.Anything, mock.Anything, - mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) + mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) res, stampErr := stamp() mockStore.On("StampFirstDispatchTx", mock.Anything, mock.Anything, mock.Anything, mock.Anything). Run(func(args mock.Arguments) { From 2f0063b8b03c372b234eaa54d7ec6b0bd4bed716 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 19:22:23 +0100 Subject: [PATCH 2/4] docs(known-issues): record the Redshift adoption-window refusal Part of #702. Mark the cross-lineage adoption entry as mitigated for buys the platform records, and list what remains open. Co-Authored-By: Claude Opus 5.5 --- known_issues/39_redshift_dedupe_window.md | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/known_issues/39_redshift_dedupe_window.md b/known_issues/39_redshift_dedupe_window.md index 8f8b75b6a..73e1a4178 100644 --- a/known_issues/39_redshift_dedupe_window.md +++ b/known_issues/39_redshift_dedupe_window.md @@ -8,9 +8,20 @@ **Description:** The platform stamps `PurchaseOptions.RequestedAt` at dispatch. go's Redshift dedupe matches an existing reserved node through `DescribeReservedNodes` when its start time is at or after `RequestedAt - 5min`, and the window has no upper bound. A later purchase with the same account, region, offering and node count made within about 5 minutes of a completed one finds the earlier node, adopts it and reports success without buying anything. -**Why deferred:** The fix belongs in go (an upper bound or an identity stronger than the tuple), tracked in LeanerCloud/cloud-commitments-go#515. The platform only supplies `RequestedAt` (#702). +**Why deferred:** The full fix belongs in go (an upper bound or an identity stronger than the tuple), tracked in LeanerCloud/cloud-commitments-go#515. The platform mitigates it for buys it records (below). -**Status:** Open. Tracked in #702 and LeanerCloud/cloud-commitments-go#515. +**Mitigation (platform-recorded buys):** The Redshift dispatch gate now refuses a purchase when another lineage's completed, partially completed or revocation-requested row for the same tuple was last saved within 10 minutes of the new lineage's first dispatch time. The refusal is recorded before submission with a fixed text that names no execution, and the user starts a new purchase at least 10 minutes after the other one completed. + +**Remaining:** + +- Buys made outside the platform are invisible to the gate. +- Two identical-tuple recs in one execution still adopt each other (platform #1809). +- A lineage whose purchase record never landed (the child bought but its row save failed, `execution.go:351` and `:383`) leaves no completed row for the gate to see. +- The go-side narrowing in LeanerCloud/cloud-commitments-go#700 and the real fix in #515 are open. +- The window is measured from the other row's last save, so a later save of that row (for example a revocation request) restarts it. That only refuses longer; it never allows a double buy. +- After the Redshift re-drive guard is lifted, a retry can be refused for up to 24h while another lineage's completed node sits inside its window (liveness only, no double buy). + +**Status:** Mitigated for platform-recorded buys. Tracked in #702 and LeanerCloud/cloud-commitments-go#515. ## LOW: unknown outcome followed by a new same-tuple purchase outside the window can double-buy From 71ef8ad148ac4e4268b37fab2532dadd5b01b16b Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 19:54:44 +0100 Subject: [PATCH 3/4] test(purchase): pin the Redshift adoption window to go's skew plus margin Part of #702. A review probe set the window to 4 minutes, below go's 5-minute Redshift dedupe skew, and every test still passed, because the tests derived their expectations from the constant. Assert the window with literals: go's 5-minute skew plus a 5-minute clock margin. Also drop the finished-row helper's unused return (unparam). Co-Authored-By: Claude Opus 5.5 --- internal/purchase/redshift_gate_integration_test.go | 3 +-- internal/purchase/redshift_gate_test.go | 9 ++++++++- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/internal/purchase/redshift_gate_integration_test.go b/internal/purchase/redshift_gate_integration_test.go index cb62c9361..1bc94bb28 100644 --- a/internal/purchase/redshift_gate_integration_test.go +++ b/internal/purchase/redshift_gate_integration_test.go @@ -183,12 +183,11 @@ func (f *redshiftGateFixture) raceOnlyOneBuys(ctx context.Context, t *testing.T, } // finishedRow saves a completed per-account row in its own lineage with a stamp. -func (f *redshiftGateFixture) finishedRow(ctx context.Context, t *testing.T, key string) *config.PurchaseExecution { +func (f *redshiftGateFixture) finishedRow(ctx context.Context, t *testing.T, key string) { t.Helper() row := f.redshiftRowKeyed(ctx, t, config.StatusCompleted, key, false) _, err := f.pool.Exec(ctx, `UPDATE purchase_executions SET first_dispatch_at = now() WHERE execution_id = $1`, row.ExecutionID) require.NoError(t, err) - return row } // A node another lineage bought seconds ago is indistinguishable to AWS from a diff --git a/internal/purchase/redshift_gate_test.go b/internal/purchase/redshift_gate_test.go index bec8778b1..4584e9cd0 100644 --- a/internal/purchase/redshift_gate_test.go +++ b/internal/purchase/redshift_gate_test.go @@ -91,7 +91,7 @@ func TestRedshiftDispatchGate_RecentFinishedBuyRefusedWithFixedText(t *testing.T assert.NotContains(t, run.err.Error(), "other-id") assert.Nil(t, exec.FirstDispatchAt, "a refused dispatch records no stamp") assert.True(t, stamp.At.Equal(run.stamp), "the window is anchored on the lineage stamp") - assert.Equal(t, redshiftAdoptionWindow, run.window) + assert.Equal(t, 10*time.Minute, run.window, "go's 5m Redshift dedupe skew plus a 5m clock margin") } } @@ -226,3 +226,10 @@ func TestRedshiftTuple_LockKeyIgnoresAccount(t *testing.T) { func TestRedshiftHorizonRefusal_NamesTheHorizon(t *testing.T) { assert.Contains(t, redshiftHorizonRefusal, fmt.Sprintf(" %.0fh ", redshiftRedriveHorizon.Hours())) } + +// go adopts a node started up to 5 minutes before RequestedAt (dedupe.go redshiftClockSkew at the +// pinned version); a window below that skew plus a 5-minute clock margin reopens the adoption. +func TestRedshiftAdoptionWindow_CoversGoSkewPlusMargin(t *testing.T) { + const goRedshiftClockSkew, clockMargin = 5 * time.Minute, 5 * time.Minute + assert.GreaterOrEqual(t, redshiftAdoptionWindow, goRedshiftClockSkew+clockMargin) +} From 404ad8eb09f6ca2ec8f8aa53b67114cf2e4afa72 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 19:54:54 +0100 Subject: [PATCH 4/4] docs(known-issues): note failed rows the adoption window does not cover Part of #702. Co-Authored-By: Claude Opus 5.5 --- known_issues/39_redshift_dedupe_window.md | 1 + 1 file changed, 1 insertion(+) diff --git a/known_issues/39_redshift_dedupe_window.md b/known_issues/39_redshift_dedupe_window.md index 73e1a4178..399088995 100644 --- a/known_issues/39_redshift_dedupe_window.md +++ b/known_issues/39_redshift_dedupe_window.md @@ -19,6 +19,7 @@ - A lineage whose purchase record never landed (the child bought but its row save failed, `execution.go:351` and `:383`) leaves no completed row for the gate to see. - The go-side narrowing in LeanerCloud/cloud-commitments-go#700 and the real fix in #515 are open. - The window is measured from the other row's last save, so a later save of that row (for example a revocation request) restarts it. That only refuses longer; it never allows a double buy. +- A `failed` row without the no-submission proof (reaped mid-buy, or an unknown outcome) may hold a real node but is not refused by this window; if go adopts that node, the accounting stays one node and one completed row. PR3 counts such rows as open lineages. - After the Redshift re-drive guard is lifted, a retry can be refused for up to 24h while another lineage's completed node sits inside its window (liveness only, no double buy). **Status:** Mitigated for platform-recorded buys. Tracked in #702 and LeanerCloud/cloud-commitments-go#515.