diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 75363819f..71f7706e6 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -819,6 +819,15 @@ type StoreInterface interface { // whether it is within horizon on the database clock. StampFirstDispatchTx(ctx context.Context, tx pgx.Tx, executionID string, horizon time.Duration) (StampResult, error) + // LockRedshiftTuplesTx takes a transaction-scoped advisory lock per + // distinct tuple, in sorted order, serializing dispatches that share one. + LockRedshiftTuplesTx(ctx context.Context, tx pgx.Tx, tuples []RedshiftTuple) error + + // 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) + // 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 diff --git a/internal/config/store_postgres_redshift_lineage.go b/internal/config/store_postgres_redshift_lineage.go new file mode 100644 index 000000000..03bb113dd --- /dev/null +++ b/internal/config/store_postgres_redshift_lineage.go @@ -0,0 +1,133 @@ +package config + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "time" + + "github.com/jackc/pgx/v5" +) + +// RedshiftTuple identifies one Redshift reserved-node purchase. Go's dedupe +// window matches on exactly these fields (account is implicit in the client). +type RedshiftTuple struct { + Region string `json:"region"` + NodeType string `json:"node_type"` + Term int `json:"term"` + Payment string `json:"payment"` + Count int `json:"count"` + // CloudAccountID is the rec's own account; nil falls back to the accountID argument. + CloudAccountID *string `json:"cloud_account_id,omitempty"` +} + +// LockKey is the advisory lock key of the tuple. It leaves the account out +// because a NULL account on either side of the conflict query matches any. +func (t RedshiftTuple) LockKey() string { + return fmt.Sprintf("redshift-tuple:%s|%s|%d|%s|%d", t.Region, t.NodeType, t.Term, t.Payment, t.Count) +} + +// LockRedshiftTuplesTx takes a transaction-scoped advisory lock per tuple in +// sorted key order, so two dispatches in different sessions cannot deadlock. +// It must be the first thing run in tx (the isolation level can only be set +// before any query). Parallels planStepLockKeys + LockPlanProvidersTx. +func (s *PostgresStore) LockRedshiftTuplesTx(ctx context.Context, tx pgx.Tx, tuples []RedshiftTuple) error { + keys := make([]string, 0, len(tuples)) + for _, t := range tuples { + keys = append(keys, t.LockKey()) + } + sort.Strings(keys) + if _, err := tx.Exec(ctx, "SET TRANSACTION ISOLATION LEVEL READ COMMITTED"); err != nil { + return fmt.Errorf("lock redshift tuples: set isolation: %w", err) + } + if _, err := tx.Exec(ctx, fmt.Sprintf("SET LOCAL lock_timeout = %d", ladderLockTimeout.Milliseconds())); err != nil { + return fmt.Errorf("lock redshift tuples: set lock_timeout: %w", err) + } + for _, k := range keys { + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, k); err != nil { + return fmt.Errorf("lock redshift tuple %q: %w", k, err) + } + } + // The timeout was for the advisory waits only; do not leak it into later statements. + if _, err := tx.Exec(ctx, "SET LOCAL lock_timeout TO DEFAULT"); err != nil { + return fmt.Errorf("lock redshift tuples: reset lock_timeout: %w", err) + } + return nil +} + +// 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). +type RedshiftLineageConflict struct { + ExecutionID string + Status string + FirstDispatchAt *time.Time + SameFamily bool + Error string + 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) { + if family == "" { + return nil, fmt.Errorf("find open redshift lineage conflicts: empty family") + } + if len(tuples) == 0 { + return nil, nil + } + tuplesJSON, err := json.Marshal(tuples) + if err != nil { + return nil, fmt.Errorf("marshal redshift tuples: %w", err) + } + 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 + FROM purchase_executions e + WHERE e.execution_id <> $1 + AND e.first_dispatch_at IS NOT NULL + AND ( + e.status = 'running' + 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)) + ) + AND EXISTS ( + SELECT 1 + FROM jsonb_array_elements(CASE WHEN jsonb_typeof(e.recommendations) = 'array' THEN e.recommendations ELSE '[]'::jsonb END) r + JOIN jsonb_to_recordset($5::jsonb) AS t(region text, node_type text, term int, payment text, count int, cloud_account_id uuid) + 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)) + 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) + if err != nil { + return nil, fmt.Errorf("find open redshift lineage conflicts: %w", err) + } + defer rows.Close() + + var out []RedshiftLineageConflict + 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 { + return nil, fmt.Errorf("scan redshift lineage conflict: %w", err) + } + if err := json.Unmarshal(recsJSON, &c.Recommendations); err != nil { + return nil, fmt.Errorf("decode recommendations of execution %s: %w", c.ExecutionID, err) + } + out = append(out, c) + } + return out, rows.Err() +} diff --git a/internal/config/store_postgres_redshift_lineage_integration_test.go b/internal/config/store_postgres_redshift_lineage_integration_test.go new file mode 100644 index 000000000..97da919af --- /dev/null +++ b/internal/config/store_postgres_redshift_lineage_integration_test.go @@ -0,0 +1,329 @@ +//go:build integration +// +build integration + +package config + +import ( + "context" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const lineageTestHorizon = 24 * time.Hour + +var lineageTuple = RedshiftTuple{Region: "us-east-1", NodeType: "ra3.xlplus", Term: 1, Payment: "all-upfront", Count: 2} + +// seedLineageRow inserts an execution holding one selected Redshift rec that +// matches lineageTuple, in the given lineage family, status and stamp age. +func seedLineageRow(ctx context.Context, t *testing.T, store *PostgresStore, family, status string, stampAge *time.Duration) *PurchaseExecution { + t.Helper() + exec := newNoSubmissionExec(ctx, t, store, status) + exec.Recommendations = []RecommendationRecord{{ + Provider: "aws", Service: "redshift", Region: lineageTuple.Region, ResourceType: lineageTuple.NodeType, + Term: lineageTuple.Term, Payment: lineageTuple.Payment, Count: lineageTuple.Count, Selected: true, + }} + require.NoError(t, store.SavePurchaseExecution(ctx, exec)) + var stamp *time.Time + if stampAge != nil { + ts := time.Now().Add(-*stampAge) + stamp = &ts + } + _, err := store.db.Exec(ctx, `UPDATE purchase_executions SET idempotency_key = $2, first_dispatch_at = $3 WHERE execution_id = $1`, + exec.ExecutionID, family, stamp) + require.NoError(t, err) + return exec +} + +func findLineageConflicts(ctx context.Context, t *testing.T, store *PostgresStore, self, family string) []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) + return err + })) + return got +} + +func ptrDur(d time.Duration) *time.Duration { return &d } + +func conflictIDs(cs []RedshiftLineageConflict) []string { + ids := make([]string, 0, len(cs)) + for _, c := range cs { + ids = append(ids, c.ExecutionID) + } + return ids +} + +func TestFindOpenRedshiftLineageConflicts(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + recent, old := ptrDur(time.Hour), ptrDur(48*time.Hour) + // Rows of one subtest must not leak into the next: they all share a tuple. + reset := func(t *testing.T) { + t.Helper() + _, err := store.db.Exec(ctx, `DELETE FROM purchase_executions`) + require.NoError(t, err) + } + + t.Run("same_family_running_other_row_conflicts", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-a", "running", recent) + other := seedLineageRow(ctx, t, store, "fam-a:acct", "running", recent) + got := findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-a") + require.Len(t, got, 1) + assert.Equal(t, other.ExecutionID, got[0].ExecutionID) + assert.True(t, got[0].SameFamily) + assert.Equal(t, "running", got[0].Status) + require.NotNil(t, got[0].FirstDispatchAt) + require.Len(t, got[0].Recommendations, 1) + }) + t.Run("same_family_self_excluded", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-b", "running", recent) + assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-b")) + }) + t.Run("same_family_non_running_excluded", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-c", "running", recent) + for _, st := range []string{"failed", "pending", "approved"} { + seedLineageRow(ctx, t, store, "fam-c:"+st, st, recent) + } + assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-c")) + }) + t.Run("other_family_running_any_stamp_age_conflicts", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-d", "running", recent) + other := seedLineageRow(ctx, t, store, "fam-d-other", "running", old) + got := findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-d") + require.Len(t, got, 1) + assert.Equal(t, []string{other.ExecutionID}, conflictIDs(got)) + assert.False(t, got[0].SameFamily) + }) + t.Run("other_family_pending_with_stamp_conflicts", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-e", "running", recent) + other := seedLineageRow(ctx, t, store, "fam-e-other", "pending", recent) + assert.Equal(t, []string{other.ExecutionID}, conflictIDs(findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-e"))) + }) + t.Run("paused_conflicts", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-f", "running", recent) + other := seedLineageRow(ctx, t, store, "fam-f-other", "paused", recent) + assert.Equal(t, []string{other.ExecutionID}, conflictIDs(findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-f"))) + }) + t.Run("approved_and_scheduled_conflict", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-g", "running", recent) + a := seedLineageRow(ctx, t, store, "fam-g-a", "approved", recent) + s := seedLineageRow(ctx, t, store, "fam-g-s", "scheduled", recent) + n := seedLineageRow(ctx, t, store, "fam-g-n", "notified", recent) + assert.ElementsMatch(t, []string{a.ExecutionID, s.ExecutionID, n.ExecutionID}, + conflictIDs(findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-g"))) + }) + t.Run("stamp_older_than_H_does_not_conflict_when_not_running", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-h", "running", recent) + seedLineageRow(ctx, t, store, "fam-h-other", "pending", old) + assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-h")) + }) + t.Run("row_without_stamp_ignored", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-i", "running", recent) + seedLineageRow(ctx, t, store, "fam-i-other", "pending", nil) + assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-i")) + }) + t.Run("UnstampedRunningRowIsNotAConflict", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-j", "running", recent) + seedLineageRow(ctx, t, store, "fam-j-other", "running", nil) + assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-j")) + }) + t.Run("row_account_null_falls_back_to_rec_account", func(t *testing.T) { + reset(t) + a1, a2 := "11111111-1111-1111-1111-111111111111", "22222222-2222-2222-2222-222222222222" + self := seedLineageRow(ctx, t, store, "fam-k", "running", recent) + other := seedLineageRow(ctx, t, store, "fam-k-other", "running", recent) + _, err := store.db.Exec(ctx, `UPDATE purchase_executions SET recommendations = jsonb_set(recommendations, '{0,cloud_account_id}', to_jsonb($2::text)) WHERE execution_id = $1`, other.ExecutionID, a1) + require.NoError(t, err) + find := func(acct string) []RedshiftLineageConflict { + tp := lineageTuple + tp.CloudAccountID = &acct + 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) + return e + })) + return got + } + assert.Empty(t, find(a2)) + assert.Equal(t, []string{other.ExecutionID}, conflictIDs(find(a1))) + }) + t.Run("null_recommendations_json_does_not_error", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-l", "running", recent) + other := seedLineageRow(ctx, t, store, "fam-l-other", "running", recent) + _, err := store.db.Exec(ctx, `UPDATE purchase_executions SET recommendations = 'null'::jsonb WHERE execution_id = $1`, other.ExecutionID) + require.NoError(t, err) + assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-l")) + }) + // candidate runs the query for self/family with one tuple and account arg. + candidate := func(self, family string, tp RedshiftTuple, arg *string) []RedshiftLineageConflict { + 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) + return e + })) + return got + } + setRec := func(t *testing.T, id, expr string, args ...any) { + t.Helper() + _, err := store.db.Exec(ctx, `UPDATE purchase_executions SET recommendations = `+expr+` WHERE execution_id = $1`, append([]any{id}, args...)...) + require.NoError(t, err) + } + + t.Run("tuple_field_mismatch_does_not_conflict", func(t *testing.T) { + cases := map[string]func(*RedshiftTuple){ + "region": func(tp *RedshiftTuple) { tp.Region = "eu-west-1" }, + "node_type": func(tp *RedshiftTuple) { tp.NodeType = "ra3.4xlarge" }, + "term": func(tp *RedshiftTuple) { tp.Term = 3 }, + "payment": func(tp *RedshiftTuple) { tp.Payment = "no-upfront" }, + "count_up": func(tp *RedshiftTuple) { tp.Count = 3 }, + "count_down": func(tp *RedshiftTuple) { tp.Count = 1 }, + } + for name, mutate := range cases { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-m", "running", recent) + seedLineageRow(ctx, t, store, "fam-m-other", "running", recent) + tp := lineageTuple + mutate(&tp) + assert.Empty(t, candidate(self.ExecutionID, "fam-m", tp, nil), name) + } + }) + t.Run("row_count_3_vs_candidate_2_and_reverse_do_not_conflict", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-n", "running", recent) + other := seedLineageRow(ctx, t, store, "fam-n-other", "running", recent) + setRec(t, other.ExecutionID, `jsonb_set(recommendations, '{0,count}', '3')`) + assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-n")) + tp := lineageTuple + tp.Count = 3 + assert.Equal(t, []string{other.ExecutionID}, conflictIDs(candidate(self.ExecutionID, "fam-n", tp, nil))) + tp.Count = 4 + assert.Empty(t, candidate(self.ExecutionID, "fam-n", tp, nil)) + setRec(t, other.ExecutionID, `jsonb_set(recommendations, '{0,count}', '2')`) + tp.Count = 3 + assert.Empty(t, candidate(self.ExecutionID, "fam-n", tp, nil)) + }) + t.Run("unselected_rec_does_not_conflict", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-o", "running", recent) + other := seedLineageRow(ctx, t, store, "fam-o-other", "running", recent) + setRec(t, other.ExecutionID, `jsonb_set(recommendations, '{0,selected}', 'false')`) + assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-o")) + }) + t.Run("provider_and_service_filters", func(t *testing.T) { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-p", "running", recent) + az := seedLineageRow(ctx, t, store, "fam-p-az", "running", recent) + setRec(t, az.ExecutionID, `jsonb_set(jsonb_set(recommendations, '{0,provider}', '"azure"'), '{0,service}', '"data-warehouse"')`) + other := seedLineageRow(ctx, t, store, "fam-p-ec2", "running", recent) + setRec(t, other.ExecutionID, `jsonb_set(recommendations, '{0,service}', '"ec2"')`) + assert.Empty(t, findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-p")) + aws := seedLineageRow(ctx, t, store, "fam-p-aws", "running", recent) + setRec(t, aws.ExecutionID, `jsonb_set(recommendations, '{0,service}', '"data-warehouse"')`) + assert.Equal(t, []string{aws.ExecutionID}, conflictIDs(findLineageConflicts(ctx, t, store, self.ExecutionID, "fam-p"))) + }) + t.Run("account_matrix", 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) + } + // Each of the four slots is nil, a1 or a2. The row account, else the rec + // account, is compared with the tuple account, else the argument. + slots := []*string{nil, &a1, &a2} + eff := func(first, second *string) *string { + if first != nil { + return first + } + return second + } + for _, rowA := range slots { + for _, recA := range slots { + for _, tupA := range slots { + for _, argA := range slots { + reset(t) + self := seedLineageRow(ctx, t, store, "fam-q", "running", recent) + other := seedLineageRow(ctx, t, store, "fam-q-other", "running", recent) + _, err := store.db.Exec(ctx, `UPDATE purchase_executions SET cloud_account_id = $2 WHERE execution_id = $1`, other.ExecutionID, rowA) + require.NoError(t, err) + if recA != nil { + setRec(t, other.ExecutionID, `jsonb_set(recommendations, '{0,cloud_account_id}', to_jsonb($2::text))`, *recA) + } + tp := lineageTuple + tp.CloudAccountID = tupA + l, r := eff(rowA, recA), eff(tupA, argA) + 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) + } + } + } + } + }) + t.Run("non_uuid_rec_account_is_a_wildcard_not_an_error", func(t *testing.T) { + reset(t) + a1 := "11111111-1111-1111-1111-111111111111" + self := seedLineageRow(ctx, t, store, "fam-r", "running", recent) + other := seedLineageRow(ctx, t, store, "fam-r-other", "running", recent) + setRec(t, other.ExecutionID, `jsonb_set(recommendations, '{0,cloud_account_id}', '"not-a-uuid"')`) + tp := lineageTuple + tp.CloudAccountID = &a1 + assert.Equal(t, []string{other.ExecutionID}, conflictIDs(candidate(self.ExecutionID, "fam-r", tp, nil))) + }) + 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) + assert.Error(t, err) + return nil + })) + }) +} + +func TestLockRedshiftTuplesTx_BlocksSameTupleAndSetsReadCommitted(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + other := lineageTuple + other.Count = 9 + tx1, err := store.db.Begin(ctx) + require.NoError(t, err) + defer func() { _ = tx1.Rollback(ctx) }() + _, err = tx1.Exec(ctx, "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ") + require.NoError(t, err, "start above read committed so the lock helper's own SET is what is proven") + require.NoError(t, store.LockRedshiftTuplesTx(ctx, tx1, []RedshiftTuple{lineageTuple, other})) + var iso string + require.NoError(t, tx1.QueryRow(ctx, "SHOW transaction_isolation").Scan(&iso)) + assert.Equal(t, "read committed", iso) + + tx2, err := store.db.Begin(ctx) + require.NoError(t, err) + defer func() { _ = tx2.Rollback(ctx) }() + short, cancel := context.WithTimeout(ctx, 300*time.Millisecond) + defer cancel() + assert.Error(t, store.LockRedshiftTuplesTx(short, tx2, []RedshiftTuple{lineageTuple}), "a held tuple blocks a second transaction") + + tx3, err := store.db.Begin(ctx) + require.NoError(t, err) + defer func() { _ = tx3.Rollback(ctx) }() + third := lineageTuple + third.Count = 11 + assert.NoError(t, store.LockRedshiftTuplesTx(ctx, tx3, []RedshiftTuple{third}), "a different tuple does not block") +} diff --git a/internal/config/types.go b/internal/config/types.go index c3f3e63d0..ef68f7371 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -407,7 +407,7 @@ type PurchaseExecution struct { // (CHECK in migration 000130). CreatedBySystem bool `json:"-" dynamodbav:"-"` // FirstDispatchAt is the time of the first dispatch of a purchase lineage. - // Server-only; written once at INSERT, never updated (migration 000154). + // Server-only; insert-only in the upsert, set once by the dispatch stamp (migration 000154). FirstDispatchAt *time.Time `json:"-" dynamodbav:"-"` // RetryExecutionID points from a *failed* execution to the new // execution created when the user clicked Retry (issue #47). Set diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 3664f30f6..cde58ae63 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -569,6 +569,26 @@ func (m *MockConfigStore) StampFirstDispatchTx(ctx context.Context, tx pgx.Tx, e return res, args.Error(1) } +// LockRedshiftTuplesTx mocks the Redshift tuple advisory locks (no default: testify panics if unexpected). +func (m *MockConfigStore) LockRedshiftTuplesTx(ctx context.Context, tx pgx.Tx, tuples []config.RedshiftTuple) error { + m.record("LockRedshiftTuplesTx", ctx, tx, tuples) + return m.Called(ctx, tx, tuples).Error(0) +} + +// 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) + if args.Get(0) == nil { + return nil, args.Error(1) + } + res, ok := args.Get(0).([]config.RedshiftLineageConflict) + if !ok { + m.reporter.wrongReturnType("FindOpenRedshiftLineageConflictsTx", "[]config.RedshiftLineageConflict", args.Get(0)) + } + return res, 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) diff --git a/internal/purchase/armed_redrive_test.go b/internal/purchase/armed_redrive_test.go index a4acafe9e..ce7fbdf24 100644 --- a/internal/purchase/armed_redrive_test.go +++ b/internal/purchase/armed_redrive_test.go @@ -111,6 +111,9 @@ func armedHarness(t *testing.T, serviceType common.ServiceType) (*Manager, *Mock stampedAt := time.Now().UTC() store.On("StampFirstDispatchTx", mock.Anything, mock.Anything, mock.Anything, mock.Anything). 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() 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 263d35fa4..355c5bdce 100644 --- a/internal/purchase/errnotes.go +++ b/internal/purchase/errnotes.go @@ -31,6 +31,9 @@ var serverNotePatterns = []string{ `target already met: target \d+, committed \d+(?:; ` + skippedNotePattern + `)?; \d+ failed row\(s\) of unknown outcome counted as committed`, `selected recommendations do not resolve to a single cloud account: (?:\d+ selected recommendation\(s\) target \d+ cloud accounts \(` + idListPattern + `\); submit one purchase per account|\d+ selected recommendation\(s\) carry no cloud_account_id while \d+ target account ` + idPattern + `)`, `purchase was not successful`, + regexp.QuoteMeta(redshiftConflictRefusal), + regexp.QuoteMeta(redshiftHorizonRefusal), + regexp.QuoteMeta(redshiftNoStampRefusal), `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 cccc1e40e..c26a7b7f0 100644 --- a/internal/purchase/errnotes_test.go +++ b/internal/purchase/errnotes_test.go @@ -50,6 +50,9 @@ func TestSanitizeStoredError_AllowlistedNotesPassThroughVerbatim(t *testing.T) { fmt.Errorf("%w: 2 selected recommendation(s) target 2 cloud accounts (a, b); submit one purchase per account", errAmbiguousAccountScope).Error(), fmt.Errorf("%w: 1 selected recommendation(s) carry no cloud_account_id while 2 target account a", errAmbiguousAccountScope).Error(), "purchase was not successful", + redshiftConflictRefusal, + redshiftHorizonRefusal, + redshiftNoStampRefusal, "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", @@ -198,3 +201,18 @@ func TestSanitizeStoredError_ResolveNoteExchangeIDIsClosed(t *testing.T) { assert.Equal(t, ErrCategoryMessage(ErrCategoryUnknown), SanitizeStoredError(in), id) } } + +// 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} { + assert.Equal(t, text, SanitizeStoredError(text)) + for _, variant := range []string{ + text + " (execution 00000000-0000-0000-0000-000000000000, running)", + text + ": pq: connection refused", + text + "\nraw", + } { + assert.Equal(t, ErrCategoryMessage(ErrCategoryUnknown), SanitizeStoredError(variant), variant) + } + } +} diff --git a/internal/purchase/ramp_step_progress_integration_test.go b/internal/purchase/ramp_step_progress_integration_test.go index 59031e1ee..5f38bb41a 100644 --- a/internal/purchase/ramp_step_progress_integration_test.go +++ b/internal/purchase/ramp_step_progress_integration_test.go @@ -54,6 +54,7 @@ func rampStepStore(ctx context.Context, t *testing.T) (*config.PostgresStore, *p // the real store with a provider that always sells. type rampStepFixture struct { store *config.PostgresStore + pool *pgxpool.Pool manager *Manager email *MockEmailSender plan *config.PurchasePlan @@ -80,7 +81,7 @@ type rampStepFixture struct { func newRampStepFixture(ctx context.Context, t *testing.T) *rampStepFixture { t.Helper() - store, _ := rampStepStore(ctx, t) + store, pool := rampStepStore(ctx, t) plan := &config.PurchasePlan{ Name: "Ramp Step Plan", @@ -131,7 +132,7 @@ func newRampStepFixture(ctx context.Context, t *testing.T) *rampStepFixture { require.Equal(t, "role_arn", accounts[1].AWSAuthMode, "accounts[1] must be a failing account") require.Equal(t, "role_arn", accounts[2].AWSAuthMode, "accounts[2] must be a failing account") - f := &rampStepFixture{store: store, plan: plan, accounts: accounts} + f := &rampStepFixture{store: store, pool: pool, plan: plan, accounts: accounts} mockEmail := new(MockEmailSender) f.email = mockEmail diff --git a/internal/purchase/redshift_gate_integration_test.go b/internal/purchase/redshift_gate_integration_test.go new file mode 100644 index 000000000..e0f699d33 --- /dev/null +++ b/internal/purchase/redshift_gate_integration_test.go @@ -0,0 +1,183 @@ +//go:build integration + +package purchase + +// Real PostgreSQL plus a counting fake provider; nothing reaches a cloud API. +// The tests set the package-level afterRedshiftStampHook, so none is parallel. + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +type redshiftGateFixture struct { + *rampStepFixture + mgr *Manager + bought atomic.Int32 + // beforeBuy runs inside each fake provider call. + beforeBuy func() +} + +func newRedshiftGateFixture(ctx context.Context, t *testing.T) *redshiftGateFixture { + t.Helper() + f := &redshiftGateFixture{rampStepFixture: newRampStepFixture(ctx, t)} + f.repairAccounts(ctx, t) + factory := new(MockProviderFactory) + provider := new(MockProvider) + client := new(MockServiceClient) + factory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(provider, nil).Maybe() + provider.On("GetServiceClient", mock.Anything, mock.Anything, mock.Anything).Return(client, nil).Maybe() + client.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.AnythingOfType("common.PurchaseOptions")). + Run(func(mock.Arguments) { + f.bought.Add(1) + if f.beforeBuy != nil { + f.beforeBuy() + } + }). + Return(common.PurchaseResult{Success: true, CommitmentID: "rs-ok"}, nil).Maybe() + allowFailedNotification(f.email) + f.mgr = &Manager{config: f.store, email: f.email, providerFactory: factory, + credStore: awsAccessKeyCredStore(), dashboardURL: "https://dashboard.example.com"} + t.Cleanup(func() { afterRedshiftStampHook = nil }) + return f +} + +// redshiftRow saves a per-account row in its own lineage. +func (f *redshiftGateFixture) redshiftRow(ctx context.Context, t *testing.T, status string) *config.PurchaseExecution { + t.Helper() + return f.redshiftRowKeyed(ctx, t, status, uuid.New().String()+":"+f.accounts[0].ID, false) +} + +// redshiftRowKeyed saves a row with the given idempotency key; root rows carry +// no cloud_account_id, so their recs fan out per account. +func (f *redshiftGateFixture) redshiftRowKeyed(ctx context.Context, t *testing.T, status, key string, root bool) *config.PurchaseExecution { + t.Helper() + acct := f.accounts[0].ID + rec := config.RecommendationRecord{ + Provider: "aws", Service: "redshift", ResourceType: "ra3.xlplus", Region: "us-east-1", + Term: 1, Payment: "no-upfront", Count: 2, UpfrontCost: 300, Selected: true, CloudAccountID: &acct, + } + row := &config.PurchaseExecution{ + PlanID: f.plan.ID, ExecutionID: uuid.New().String(), IdempotencyKey: key, + Status: status, StepNumber: 3, ScheduledDate: time.Now(), CloudAccountID: &acct, + Recommendations: []config.RecommendationRecord{rec}, + } + if root { + row.CloudAccountID = nil + } + require.NoError(t, f.store.SavePurchaseExecution(ctx, row)) + return row +} + +func (f *redshiftGateFixture) stampOf(ctx context.Context, t *testing.T, id string) *time.Time { + t.Helper() + var at *time.Time + require.NoError(t, f.pool.QueryRow(ctx, `SELECT first_dispatch_at FROM purchase_executions WHERE execution_id = $1`, id).Scan(&at)) + return at +} + +// A conflicting open lineage refuses the dispatch and takes the stamp with it. +func TestRedshiftDispatchGate_ConflictRollsBackStamp(t *testing.T) { + ctx := context.Background() + f := newRedshiftGateFixture(ctx, t) + open := f.redshiftRow(ctx, t, "running") + _, err := f.pool.Exec(ctx, `UPDATE purchase_executions SET first_dispatch_at = now() WHERE execution_id = $1`, open.ExecutionID) + require.NoError(t, err) + 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, redshiftConflictRefusal, SanitizeStoredError(got.Error), "users see the fixed text verbatim") + assert.Nil(t, f.stampOf(ctx, t, refused.ExecutionID), "the refusal rolled the stamp back") +} + +// Two lineages of one tuple dispatch at once. The first to hold the tuple lock +// waits in the hook until the other is blocked on it, then both gates finish: +// without the lock both would stamp, see no conflict and buy. +func TestRedshiftDispatchGate_TwoLineagesRaceOnlyOneBuys(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + f := newRedshiftGateFixture(ctx, t) + f.raceOnlyOneBuys(ctx, t, f.redshiftRow(ctx, t, "approved"), f.redshiftRow(ctx, t, "approved")) +} + +// A root row (key R) and a per-account successor (key R:acct) share one +// family, yet only one may buy. A blanket same-family exclusion would let both. +func TestRedshiftDispatchGate_RootAndAccountSuccessorsOfOneFamilyDoNotBothBuy(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + f := newRedshiftGateFixture(ctx, t) + family := uuid.New().String() + f.raceOnlyOneBuys(ctx, t, + f.redshiftRowKeyed(ctx, t, "approved", family, true), + f.redshiftRowKeyed(ctx, t, "approved", family+":"+f.accounts[0].ID, false)) +} + +func (f *redshiftGateFixture) raceOnlyOneBuys(ctx context.Context, t *testing.T, rows ...*config.PurchaseExecution) { + t.Helper() + var once sync.Once + afterRedshiftStampHook = func() { + once.Do(func() { + assert.EventuallyWithT(t, func(c *assert.CollectT) { + var waiting int + require.NoError(c, f.pool.QueryRow(ctx, `SELECT count(*) FROM pg_locks WHERE locktype = 'advisory' AND NOT granted + AND database = (SELECT oid FROM pg_database WHERE datname = current_database())`).Scan(&waiting)) + assert.Positive(c, waiting) + }, 5*time.Second, 20*time.Millisecond, "the second gate never blocked on the tuple lock") + }) + } + // The winner keeps the provider call open until the loser is refused, so + // the loser's conflict query sees a running row, not a completed one. + f.beforeBuy = func() { + assert.EventuallyWithT(t, func(c *assert.CollectT) { + var failed int + require.NoError(c, f.pool.QueryRow(ctx, `SELECT count(*) FROM purchase_executions WHERE plan_id = $1 AND status = 'failed'`, f.plan.ID).Scan(&failed)) + assert.Positive(c, failed) + }, 5*time.Second, 20*time.Millisecond, "the loser was never refused") + } + + start := make(chan struct{}) + var wg sync.WaitGroup + for _, row := range rows { + wg.Add(1) + go func() { + defer wg.Done() + <-start + _, _ = f.mgr.claimAndExecute(ctx, rowStatusCopy(row)) + }() + } + close(start) + wg.Wait() + + assert.Equal(t, int32(1), f.bought.Load(), "exactly one lineage reaches the provider") + var winners, losers int + for _, row := range rows { + got := rowStatus(ctx, t, f.rampStepFixture, row.ExecutionID) + switch got.Status { + case "completed": + winners++ + assert.NotNil(t, f.stampOf(ctx, t, row.ExecutionID)) + case config.StatusFailed: + losers++ + assert.True(t, got.NoSubmission) + assert.Nil(t, f.stampOf(ctx, t, row.ExecutionID), "the loser's stamp was rolled back") + } + } + assert.Equal(t, 1, winners) + assert.Equal(t, 1, losers) +} diff --git a/internal/purchase/redshift_gate_test.go b/internal/purchase/redshift_gate_test.go new file mode 100644 index 000000000..3bff68a43 --- /dev/null +++ b/internal/purchase/redshift_gate_test.go @@ -0,0 +1,197 @@ +package purchase + +import ( + "context" + "errors" + "fmt" + "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" +) + +func gateRec(account string, count int) config.RecommendationRecord { + r := redshiftScopedRec(account) + r.Region, r.Term, r.Payment, r.Count = "us-east-1", 1, "no-upfront", count + return r +} + +type gateRun struct { + err error + exec *config.PurchaseExecution + finds []config.RedshiftTuple + family string + locked []config.RedshiftTuple +} + +// runGate drives the gate with a stamp result and the conflicts the query returns. +func runGate(t *testing.T, exec *config.PurchaseExecution, stamp config.StampResult, conflicts []config.RedshiftLineageConflict) *gateRun { + t.Helper() + run := &gateRun{exec: exec} + store := new(MockConfigStore) + store.On("LockRedshiftTuplesTx", mock.Anything, mock.Anything, mock.Anything). + 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) }). + Return(conflicts, nil).Maybe() + run.err = (&Manager{config: store}).stampRedshiftDispatch(context.Background(), exec) + return run +} + +func stamped() config.StampResult { + at := time.Now().UTC().Add(-time.Hour) + return config.StampResult{At: &at, WithinHorizon: true} +} + +func gateExec(attempt int, recs ...config.RecommendationRecord) *config.PurchaseExecution { + return &config.PurchaseExecution{ + ExecutionID: "rs-self", IdempotencyKey: "fam-1:acct-A", RetryAttemptN: attempt, Recommendations: recs, + } +} + +func isBeforeSubmission(err error) bool { + var bs *beforeSubmissionError + return errors.As(err, &bs) +} + +func TestRedshiftDispatchGate_ConflictRefusesBeforeSubmission(t *testing.T) { + for _, attempt := range []int{0, 1} { + exec := gateExec(attempt, gateRec("acct-A", 2)) + run := runGate(t, exec, stamped(), []config.RedshiftLineageConflict{{ExecutionID: "other", Status: "running"}}) + + require.Error(t, run.err) + assert.True(t, isBeforeSubmission(run.err), "attempt %d", attempt) + assert.Equal(t, redshiftConflictRefusal, run.err.Error(), "the text names no other execution") + assert.Nil(t, exec.FirstDispatchAt, "a refused dispatch records no stamp") + assert.Equal(t, "fam-1", run.family) + } +} + +func TestRedshiftDispatchGate_QueriesOnlySelectedRedshiftTuples(t *testing.T) { + unselected := gateRec("acct-A", 7) + unselected.Selected = false + ec2 := scopedTestRec("acct-A") + exec := gateExec(0, gateRec("acct-A", 2), gateRec("acct-A", 3), unselected, ec2) + run := runGate(t, exec, stamped(), nil) + + require.NoError(t, run.err) + require.NotNil(t, exec.FirstDispatchAt) + require.Len(t, run.finds, 2) + assert.Equal(t, 2, run.finds[0].Count) + assert.Equal(t, 3, run.finds[1].Count) +} + +func TestRedshiftDispatchGate_SameFamilySiblingFailedRowAllowed(t *testing.T) { + exec := gateExec(1, gateRec("acct-A", 2)) + run := runGate(t, exec, stamped(), nil) + + require.NoError(t, run.err) + assert.Equal(t, "fam-1", run.family, "the family root lets the query exclude same-family non-running rows") +} + +func TestRedshiftDispatchGate_RetrySuccessorWithoutStampRefusedNotBeforeSubmission(t *testing.T) { + exec := gateExec(1, gateRec("acct-A", 2)) + run := runGate(t, exec, config.StampResult{}, nil) + + require.Error(t, run.err) + assert.False(t, isBeforeSubmission(run.err), "no_submission must stay false") + assert.Nil(t, exec.FirstDispatchAt) +} + +func TestRedshiftDispatchGate_PastHorizonRefusedAtEveryAttempt(t *testing.T) { + for _, attempt := range []int{0, 1} { + at := time.Now().UTC().Add(-30 * time.Hour) + exec := gateExec(attempt, gateRec("acct-A", 2)) + run := runGate(t, exec, config.StampResult{At: &at, WithinHorizon: false}, nil) + + require.Error(t, run.err) + assert.False(t, isBeforeSubmission(run.err), "attempt %d: an old stamp proves an earlier dispatch", attempt) + assert.Equal(t, redshiftHorizonRefusal, run.err.Error()) + assert.Nil(t, exec.FirstDispatchAt) + } +} + +func TestRedshiftDispatchGate_LocksEveryTupleAndPassesRecAccounts(t *testing.T) { + recB := gateRec("acct-B", 2) + exec := gateExec(0, gateRec("acct-A", 2), recB) + run := runGate(t, exec, stamped(), nil) + + require.NoError(t, run.err) + require.Len(t, run.locked, 2) + require.Len(t, run.finds, 2) + require.NotNil(t, run.finds[1].CloudAccountID) + assert.Equal(t, *recB.CloudAccountID, *run.finds[1].CloudAccountID) +} + +func TestRedshiftDispatchGate_StoreErrorsByAttempt(t *testing.T) { + for _, failing := range []string{"LockRedshiftTuplesTx", "FindOpenRedshiftLineageConflictsTx"} { + for _, attempt := range []int{0, 1} { + store := new(MockConfigStore) + boom := errors.New("db down") + lockErr, findErr := error(nil), error(nil) + if failing == "LockRedshiftTuplesTx" { + lockErr = boom + } else { + findErr = boom + } + 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() + exec := gateExec(attempt, gateRec("acct-A", 2)) + + err := (&Manager{config: store}).stampRedshiftDispatch(context.Background(), exec) + + require.ErrorIs(t, err, boom) + assert.Equal(t, attempt == 0, isBeforeSubmission(err), "%s attempt %d", failing, attempt) + assert.Nil(t, exec.FirstDispatchAt) + } + } +} + +func TestRedshiftDispatchGate_CallOrderIsLockStampFind(t *testing.T) { + store := new(MockConfigStore) + var order []string + note := func(name string) func(mock.Arguments) { return func(mock.Arguments) { order = append(order, name) } } + 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) + + require.NoError(t, (&Manager{config: store}).stampRedshiftDispatch(context.Background(), gateExec(0, gateRec("acct-A", 2)))) + assert.Equal(t, []string{"lock", "stamp", "find"}, order) +} + +func TestRedshiftTuple_LockKeyDiffersPerField(t *testing.T) { + base := config.RedshiftTuple{Region: "r", NodeType: "n", Term: 1, Payment: "p", Count: 2} + for name, mut := range map[string]func(*config.RedshiftTuple){ + "region": func(x *config.RedshiftTuple) { x.Region = "r2" }, + "node type": func(x *config.RedshiftTuple) { x.NodeType = "n2" }, + "term": func(x *config.RedshiftTuple) { x.Term = 3 }, + "payment": func(x *config.RedshiftTuple) { x.Payment = "p2" }, + "count": func(x *config.RedshiftTuple) { x.Count = 3 }, + } { + other := base + mut(&other) + assert.NotEqual(t, base.LockKey(), other.LockKey(), name) + } +} + +func TestRedshiftTuple_LockKeyIgnoresAccount(t *testing.T) { + a, b := "a", "b" + t1 := config.RedshiftTuple{Region: "r", NodeType: "n", Term: 1, Payment: "p", Count: 2, CloudAccountID: &a} + t2 := t1 + t2.CloudAccountID = &b + assert.Equal(t, t1.LockKey(), t2.LockKey()) + t2.Count = 3 + assert.NotEqual(t, t1.LockKey(), t2.LockKey()) +} + +func TestRedshiftHorizonRefusal_NamesTheHorizon(t *testing.T) { + assert.Contains(t, redshiftHorizonRefusal, fmt.Sprintf(" %.0fh ", redshiftRedriveHorizon.Hours())) +} diff --git a/internal/purchase/redshift_lineage.go b/internal/purchase/redshift_lineage.go index 548de4545..0094e5f88 100644 --- a/internal/purchase/redshift_lineage.go +++ b/internal/purchase/redshift_lineage.go @@ -2,10 +2,13 @@ package purchase import ( "context" + "errors" "fmt" + "strings" "time" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" "github.com/jackc/pgx/v5" ) @@ -33,26 +36,59 @@ func hasRedshiftRec(exec *config.PurchaseExecution) bool { return false } -// stampRedshiftDispatch fixes the lineage's first dispatch time before any -// Redshift buy and records it on exec, so every dispatch of the lineage (and -// every fan-out account row, which copies exec) passes the same RequestedAt. -// Other executions are left untouched. A failure at attempt 0 provably -// submitted nothing; a successor may have, so its refusal is a plain error. +// afterRedshiftStampHook runs inside the gate transaction right after the +// stamp. Tests set it to observe the lock wait; it is nil in production. +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" +) + +// 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) + +// redshiftTuples lists the tuples of the selected Redshift recs. +func redshiftTuples(exec *config.PurchaseExecution) []config.RedshiftTuple { + var tuples []config.RedshiftTuple + for i := range exec.Recommendations { + r := &exec.Recommendations[i] + if r.Selected && isRedshiftRec(*r) { + tuples = append(tuples, config.RedshiftTuple{ + Region: r.Region, NodeType: r.ResourceType, Term: r.Term, + Payment: r.Payment, Count: r.Count, CloudAccountID: r.CloudAccountID, + }) + } + } + return tuples +} + +// stampRedshiftDispatch is the dispatch gate for a Redshift execution. In one +// transaction it locks each tuple, stamps the lineage's first dispatch (which +// becomes RequestedAt for every dispatch and fan-out account row), and refuses +// when another open lineage holds a matching tuple; the refusal rolls the +// stamp back. Other executions pass untouched. A refusal at attempt 0 or by a +// conflict provably submitted nothing; a stamp-less or past-horizon successor +// may have, so that refusal is a plain error. +// The gate needs READ COMMITTED: each statement sees a fresh snapshot after the +// lock wait. now() is the tx start, so a waiter's stamp predates its wait. func (m *Manager) stampRedshiftDispatch(ctx context.Context, exec *config.PurchaseExecution) error { if !hasRedshiftRec(exec) { return nil } var res config.StampResult err := m.config.WithTx(ctx, func(tx pgx.Tx) error { - var stampErr error - res, stampErr = m.config.StampFirstDispatchTx(ctx, tx, exec.ExecutionID, redshiftRedriveHorizon) - return stampErr + var gateErr error + res, gateErr = m.redshiftGate(ctx, tx, exec) + return gateErr }) - if err == nil && res.At == nil { - err = fmt.Errorf("execution %s has no first dispatch time and none can be minted", exec.ExecutionID) - } if err != nil { - if exec.RetryAttemptN == 0 { + var refused *beforeSubmissionError + if exec.RetryAttemptN == 0 && !errors.As(err, &refused) && !errors.Is(err, errRedshiftPastHorizon) { return beforeSubmission(err) } return err @@ -62,6 +98,38 @@ func (m *Manager) stampRedshiftDispatch(ctx context.Context, exec *config.Purcha return nil } +func (m *Manager) redshiftGate(ctx context.Context, tx pgx.Tx, exec *config.PurchaseExecution) (config.StampResult, error) { + tuples := redshiftTuples(exec) + if err := m.config.LockRedshiftTuplesTx(ctx, tx, tuples); err != nil { + return config.StampResult{}, err + } + res, err := m.config.StampFirstDispatchTx(ctx, tx, exec.ExecutionID, redshiftRedriveHorizon) + if err != nil { + return res, err + } + if res.At == nil { + logging.Warnf("Redshift gate: execution %s has no first dispatch time and none can be minted", exec.ExecutionID) + return res, errors.New(redshiftNoStampRefusal) + } + if !res.WithinHorizon { + logging.Warnf("Redshift gate: execution %s first dispatched at %s, past the horizon", exec.ExecutionID, res.At.UTC().Format(time.RFC3339)) + return res, errRedshiftPastHorizon + } + if afterRedshiftStampHook != nil { + afterRedshiftStampHook() + } + family, _, _ := strings.Cut(idempotencyLineageKey(exec), ":") + conflicts, err := m.config.FindOpenRedshiftLineageConflictsTx(ctx, tx, exec.ExecutionID, family, tuples, exec.CloudAccountID, redshiftRedriveHorizon) + if err != nil { + return res, err + } + 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)) + } + return res, nil +} + // 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 d71973f7b..8e38e28e2 100644 --- a/internal/purchase/requested_at_test.go +++ b/internal/purchase/requested_at_test.go @@ -163,6 +163,9 @@ func runRedshiftFinalize(t *testing.T, exec *config.PurchaseExecution, accounts } mockStore.GetPlanAccountsFn = func(context.Context, string) ([]config.CloudAccount, error) { return accounts, nil } 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) res, stampErr := stamp() mockStore.On("StampFirstDispatchTx", mock.Anything, mock.Anything, mock.Anything, mock.Anything). Run(func(args mock.Arguments) { @@ -310,6 +313,7 @@ func TestStampRedshiftDispatch_NilAtRefusalByAttempt(t *testing.T) { wantBeforeSubmit bool }{{0, true}, {1, false}} { store := new(MockConfigStore) + store.On("LockRedshiftTuplesTx", mock.Anything, mock.Anything, mock.Anything).Return(nil) store.On("StampFirstDispatchTx", mock.Anything, mock.Anything, "rs-x", mock.Anything). Return(config.StampResult{At: nil}, nil).Once() m := &Manager{config: store}