From e34e26030eb788d79042faf2b62540495fd265de Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Wed, 4 Nov 2026 21:13:04 +0100 Subject: [PATCH 1/4] feat(config): add purchase_executions.first_dispatch_at Part of #702 (Redshift re-drive idempotency, PR1 commit 1 of 3). - Migration 000158 adds a nullable first_dispatch_at timestamptz. - PurchaseExecution.FirstDispatchAt is server-only (json:"-") and is written once at INSERT; the upsert's DO UPDATE SET leaves it out, so a stale in-memory copy can never clear or move it. - Every full-row projection and the shared scanner read it. Nothing sets it yet; a later commit stamps it before the first dispatch and passes it to the library as PurchaseOptions.RequestedAt. Co-Authored-By: Claude Opus 5.5 --- internal/config/store_postgres.go | 28 +++++----- ...gres_first_dispatch_at_integration_test.go | 55 +++++++++++++++++++ .../config/store_postgres_history_page.go | 2 +- .../config/store_postgres_pgxmock_test.go | 38 ++++++------- .../config/store_postgres_ramp_baseline.go | 2 +- internal/config/store_postgres_ramp_step.go | 2 +- internal/config/types.go | 3 + ...hase_executions_first_dispatch_at.down.sql | 1 + ...rchase_executions_first_dispatch_at.up.sql | 1 + ...chase_executions_first_dispatch_at_test.go | 48 ++++++++++++++++ 10 files changed, 145 insertions(+), 35 deletions(-) create mode 100644 internal/config/store_postgres_first_dispatch_at_integration_test.go create mode 100644 internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.down.sql create mode 100644 internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.up.sql create mode 100644 internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at_test.go diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 0cc918310..74158da48 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -1178,8 +1178,8 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx, created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29) + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30) ON CONFLICT (execution_id) DO UPDATE SET status = $3, notification_sent = $6, @@ -1216,7 +1216,7 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx, planIDArg = execution.PlanID } - // created_by_user_id, created_by_system and retry_attempt_n are INSERT-only: the original + // created_by_user_id, created_by_system, first_dispatch_at and retry_attempt_n are INSERT-only: the original // creator and the retry-chain position must not be rewritten by an // ON CONFLICT update (e.g. the scheduler upserting status transitions). // They are omitted from the DO UPDATE SET clause above. @@ -1259,6 +1259,7 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx, execution.ScheduledExecutionAt, execution.NoSubmission, execution.CreatedBySystem, + execution.FirstDispatchAt, ) if err != nil { @@ -1418,7 +1419,7 @@ func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, q executi created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at ` records, err := queryExecutionsOn(ctx, q, query, executionID, toStatus, fromStatuses, actor, cancelActor, sawEmpty, tokenHash) @@ -1682,7 +1683,7 @@ func (s *PostgresStore) GetExecutionsByStatusesFiltered(ctx context.Context, q E created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions WHERE ` + strings.Join(conds, "\n\t\t AND ") + ` ORDER BY (status = 'pending' AND scheduled_date > now()) ASC, @@ -1794,7 +1795,7 @@ func (s *PostgresStore) GetPlannedExecutions(ctx context.Context, statuses []str created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions WHERE status = ANY($1) ORDER BY scheduled_date ASC NULLS LAST, id ASC @@ -1819,7 +1820,7 @@ func (s *PostgresStore) GetStaleApprovedExecutions(ctx context.Context, olderTha created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions WHERE status = 'approved' AND updated_at < NOW() - $1::interval ` @@ -1861,7 +1862,7 @@ func (s *PostgresStore) ListStuckExecutions(ctx context.Context, statuses []stri created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions WHERE status = ANY($1) AND updated_at < NOW() - $2::interval @@ -1882,7 +1883,7 @@ func (s *PostgresStore) GetPendingExecutions(ctx context.Context) ([]PurchaseExe created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions WHERE status IN ('pending', 'notified') AND (expires_at IS NULL OR expires_at > NOW()) @@ -1964,7 +1965,7 @@ func (s *PostgresStore) ListRecentSubmitsTx(ctx context.Context, tx pgx.Tx, crea created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions WHERE source = $3 AND status IN (` + submitDedupeStatuses + `) @@ -1999,7 +2000,7 @@ func getExecutionByIDOn(ctx context.Context, q executionQuerier, executionID str created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions WHERE execution_id = $1 ` @@ -2045,7 +2046,7 @@ func (s *PostgresStore) GetExecutionByPlanAndDate(ctx context.Context, planID st created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions WHERE plan_id = $1 AND scheduled_date = $2 ` @@ -2249,6 +2250,7 @@ func scanExecutionRows(rows pgx.Rows) ([]PurchaseExecution, error) { &scheduledExecutionAt, &exec.NoSubmission, &exec.CreatedBySystem, + &exec.FirstDispatchAt, ) if err != nil { return nil, fmt.Errorf("failed to scan execution: %w", err) @@ -2290,7 +2292,7 @@ func (s *PostgresStore) GetScheduledExecutionsDue(ctx context.Context) ([]Purcha created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions WHERE status = 'scheduled' AND scheduled_execution_at IS NOT NULL diff --git a/internal/config/store_postgres_first_dispatch_at_integration_test.go b/internal/config/store_postgres_first_dispatch_at_integration_test.go new file mode 100644 index 000000000..b1fa70d41 --- /dev/null +++ b/internal/config/store_postgres_first_dispatch_at_integration_test.go @@ -0,0 +1,55 @@ +//go:build integration +// +build integration + +package config + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// first_dispatch_at is written only by the INSERT; a stale struct upserted +// later (nil or a different value) must not move or clear it. +func TestSavePurchaseExecution_FirstDispatchAtIsInsertOnly(t *testing.T) { + ctx := context.Background() + store := setupRampStepStore(ctx, t) + + stamp := time.Now().Add(-time.Hour).UTC().Truncate(time.Microsecond) + exec := &PurchaseExecution{ + ExecutionID: uuid.New().String(), IdempotencyKey: uuid.New().String(), + Status: "pending", StepNumber: 1, ScheduledDate: time.Now(), FirstDispatchAt: &stamp, + } + require.NoError(t, store.SavePurchaseExecution(ctx, exec)) + + got, err := store.GetExecutionByID(ctx, exec.ExecutionID) + require.NoError(t, err) + require.NotNil(t, got.FirstDispatchAt, "insert persists the stamp") + assert.True(t, stamp.Equal(*got.FirstDispatchAt)) + + other := stamp.Add(30 * time.Minute) + for _, stale := range []*time.Time{nil, &other} { + exec.FirstDispatchAt = stale + exec.Status = "running" + require.NoError(t, store.SavePurchaseExecution(ctx, exec)) + got, err = store.GetExecutionByID(ctx, exec.ExecutionID) + require.NoError(t, err) + assert.Equal(t, "running", got.Status, "the upsert still updates status") + require.NotNil(t, got.FirstDispatchAt, "upsert must not clear the stamp") + assert.True(t, stamp.Equal(*got.FirstDispatchAt), "upsert must not move the stamp") + } + + // An unstamped row stays unstamped when a later upsert carries a value. + plain := &PurchaseExecution{ExecutionID: uuid.New().String(), IdempotencyKey: uuid.New().String(), + Status: "pending", StepNumber: 1, ScheduledDate: time.Now()} + require.NoError(t, store.SavePurchaseExecution(ctx, plain)) + plain.FirstDispatchAt = &other + require.NoError(t, store.SavePurchaseExecution(ctx, plain)) + got, err = store.GetExecutionByID(ctx, plain.ExecutionID) + require.NoError(t, err) + assert.Nil(t, got.FirstDispatchAt, "upsert must not set the stamp") +} diff --git a/internal/config/store_postgres_history_page.go b/internal/config/store_postgres_history_page.go index fa5092a24..4d3f29315 100644 --- a/internal/config/store_postgres_history_page.go +++ b/internal/config/store_postgres_history_page.go @@ -223,7 +223,7 @@ func (s *PostgresStore) ListHistoryExecutionsPage(ctx context.Context, q History created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions%s ORDER BY scheduled_date DESC, execution_id DESC LIMIT $%d diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go index ab640f0ac..ed67b9ed6 100644 --- a/internal/config/store_postgres_pgxmock_test.go +++ b/internal/config/store_postgres_pgxmock_test.go @@ -724,7 +724,7 @@ func TestPGXMock_GetExecutionByID_Success(t *testing.T) { "executed_by_user_id", "executed_at", "pre_approval_skip_reason", "idempotency_key", "scheduled_execution_at", - "no_submission", "created_by_system", + "no_submission", "created_by_system", "first_dispatch_at", } rows := pgxmock.NewRows(cols).AddRow( "plan-1", "exec-1", "pending", 1, now, @@ -734,9 +734,9 @@ func TestPGXMock_GetExecutionByID_Success(t *testing.T) { nil, nil, 0, sql.NullTime{}, nil, sql.NullTime{}, nil, - nil, // idempotency_key (NULL: legacy-row scan path, migration 000066) - sql.NullTime{}, // scheduled_execution_at (NULL: not on the pre-fire delay path) - false, false, // no_submission, created_by_system + nil, // idempotency_key (NULL: legacy-row scan path, migration 000066) + sql.NullTime{}, // scheduled_execution_at (NULL: not on the pre-fire delay path) + false, false, nil, // no_submission, created_by_system, first_dispatch_at ) mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg()).WillReturnRows(rows) @@ -776,7 +776,7 @@ func TestPGXMock_GetExecutionByID_WithTimestamps(t *testing.T) { "executed_by_user_id", "executed_at", "pre_approval_skip_reason", "idempotency_key", "scheduled_execution_at", - "no_submission", "created_by_system", + "no_submission", "created_by_system", "first_dispatch_at", } successorID := "exec-3" rows := pgxmock.NewRows(cols).AddRow( @@ -792,7 +792,7 @@ func TestPGXMock_GetExecutionByID_WithTimestamps(t *testing.T) { nil, sql.NullTime{}, nil, "idem-key-exec-2", // idempotency_key non-NULL: exercises the scan path sql.NullTime{}, // scheduled_execution_at (NULL: not on the pre-fire delay path) - false, false, // no_submission, created_by_system + false, false, nil, // no_submission, created_by_system, first_dispatch_at ) mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg()).WillReturnRows(rows) @@ -844,7 +844,7 @@ func TestPGXMock_GetPlannedExecutions_ProjectsAllScanColumns(t *testing.T) { "approval_token_expires_at", "executed_by_user_id", "executed_at", "pre_approval_skip_reason", "idempotency_key", "scheduled_execution_at", - "no_submission", "created_by_system", + "no_submission", "created_by_system", "first_dispatch_at", } schedAt := now.Add(2 * time.Hour) rows := pgxmock.NewRows(cols).AddRow( @@ -856,8 +856,8 @@ func TestPGXMock_GetPlannedExecutions_ProjectsAllScanColumns(t *testing.T) { sql.NullTime{}, nil, sql.NullTime{}, nil, "idem-key-planned", - sql.NullTime{}, // scheduled_execution_at (NULL: not on the pre-fire delay path) - false, false, // no_submission, created_by_system + sql.NullTime{}, // scheduled_execution_at (NULL: not on the pre-fire delay path) + false, false, nil, // no_submission, created_by_system, first_dispatch_at ).AddRow( "plan-1", "exec-2", "pending", 1, now, sql.NullTime{}, "tok-456", recsJSON, @@ -868,13 +868,13 @@ func TestPGXMock_GetPlannedExecutions_ProjectsAllScanColumns(t *testing.T) { nil, sql.NullTime{}, nil, "idem-key-delayed", sql.NullTime{Time: schedAt, Valid: true}, // scheduled_execution_at populated (pre-fire delay path) - false, false, // no_submission, created_by_system + false, false, nil, // no_submission, created_by_system, first_dispatch_at ) // Regexp matcher: only matches if the issued SELECT projects both // idempotency_key and scheduled_execution_at. The alternation forces both - // names to appear; a projection missing either column fails to match, the - // mock returns no rows, and the test catches the column-count drift. - mock.ExpectQuery(`idempotency_key.*scheduled_execution_at|scheduled_execution_at.*idempotency_key`). + // names to appear, and first_dispatch_at must follow; a projection missing any + // column fails to match, the mock returns no rows, and the test catches the column-count drift. + mock.ExpectQuery(`(idempotency_key.*scheduled_execution_at|scheduled_execution_at.*idempotency_key).*first_dispatch_at`). WithArgs(pgxmock.AnyArg(), pgxmock.AnyArg()). WillReturnRows(rows) @@ -927,7 +927,7 @@ func TestPGXMock_GetExecutionByID_ProjectsCoalescedCancelledBy(t *testing.T) { "executed_by_user_id", "executed_at", "pre_approval_skip_reason", "idempotency_key", "scheduled_execution_at", - "no_submission", "created_by_system", + "no_submission", "created_by_system", "first_dispatch_at", } // Simulates the exact row shape the atomic-cancel paths produce: the // actor lives only in canceled_by (legacy cancelled_by is NULL). What @@ -944,7 +944,7 @@ func TestPGXMock_GetExecutionByID_ProjectsCoalescedCancelledBy(t *testing.T) { nil, sql.NullTime{}, nil, nil, sql.NullTime{}, - false, false, + false, false, nil, ) mock.ExpectQuery(`COALESCE\(canceled_by,\s*cancelled_by\)`). WithArgs(pgxmock.AnyArg()). @@ -2897,9 +2897,9 @@ func stuckExecRow(execID, status string, scheduled time.Time) []any { nil, nil, 0, sql.NullTime{}, nil, sql.NullTime{}, nil, - nil, // idempotency_key (NULL: legacy-row scan path, migration 000066) - sql.NullTime{}, // scheduled_execution_at (NULL: not on the pre-fire delay path) - false, false, // no_submission, created_by_system + nil, // idempotency_key (NULL: legacy-row scan path, migration 000066) + sql.NullTime{}, // scheduled_execution_at (NULL: not on the pre-fire delay path) + false, false, nil, // no_submission, created_by_system, first_dispatch_at } } @@ -2914,7 +2914,7 @@ func stuckExecCols() []string { "executed_by_user_id", "executed_at", "pre_approval_skip_reason", "idempotency_key", "scheduled_execution_at", - "no_submission", "created_by_system", + "no_submission", "created_by_system", "first_dispatch_at", } } diff --git a/internal/config/store_postgres_ramp_baseline.go b/internal/config/store_postgres_ramp_baseline.go index 40e76a31a..580340e1a 100644 --- a/internal/config/store_postgres_ramp_baseline.go +++ b/internal/config/store_postgres_ramp_baseline.go @@ -113,7 +113,7 @@ func (s *PostgresStore) ListPlanCommittedTx(ctx context.Context, tx pgx.Tx, plan created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions WHERE plan_id = $1`, planID) if err != nil { diff --git a/internal/config/store_postgres_ramp_step.go b/internal/config/store_postgres_ramp_step.go index 70b9656a2..a28428187 100644 --- a/internal/config/store_postgres_ramp_step.go +++ b/internal/config/store_postgres_ramp_step.go @@ -331,7 +331,7 @@ func (s *PostgresStore) ListExecutionsForPlanStepTx(ctx context.Context, tx pgx. created_by_user_id, retry_execution_id, retry_attempt_n, approval_token_expires_at, executed_by_user_id, executed_at, pre_approval_skip_reason, - idempotency_key, scheduled_execution_at, no_submission, created_by_system + idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at FROM purchase_executions WHERE plan_id = $1 AND step_number = $2 ORDER BY retry_attempt_n, execution_id`, planID, stepNumber) diff --git a/internal/config/types.go b/internal/config/types.go index 90a1b9ca5..390a6e1d4 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -400,6 +400,9 @@ type PurchaseExecution struct { // request, never rewritten by the upsert. Implies CreatedByUserID == nil // (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 000158). + FirstDispatchAt *time.Time `json:"-" dynamodbav:"-"` // RetryExecutionID points from a *failed* execution to the new // execution created when the user clicked Retry (issue #47). Set // only on the original failed row; NULL on every other row including diff --git a/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.down.sql b/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.down.sql new file mode 100644 index 000000000..1dc029f0b --- /dev/null +++ b/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.down.sql @@ -0,0 +1 @@ +ALTER TABLE purchase_executions DROP COLUMN IF EXISTS first_dispatch_at; diff --git a/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.up.sql b/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.up.sql new file mode 100644 index 000000000..cd286a5f2 --- /dev/null +++ b/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.up.sql @@ -0,0 +1 @@ +ALTER TABLE purchase_executions ADD COLUMN IF NOT EXISTS first_dispatch_at timestamptz NULL; diff --git a/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at_test.go b/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at_test.go new file mode 100644 index 000000000..7c8378154 --- /dev/null +++ b/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at_test.go @@ -0,0 +1,48 @@ +//go:build integration + +package migrations_test + +import ( + "context" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Migration 000158 adds the nullable purchase_executions.first_dispatch_at: +// existing rows stay NULL and the down migration drops it. +func TestMigration_0158_FirstDispatchAt(t *testing.T) { + t.Parallel() + ctx := context.Background() + migrationsPath := getMigrationsPath() + + container := testhelpers.RequireEmptyDatabase(ctx, t) + pool := container.DB.Pool() + + columnExists := func() bool { + var exists bool + require.NoError(t, pool.QueryRow(ctx, `SELECT EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_name = 'purchase_executions' AND column_name = 'first_dispatch_at')`).Scan(&exists)) + return exists + } + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 158))) + assert.False(t, columnExists(), "column absent before 000158") + _, err := pool.Exec(ctx, `INSERT INTO purchase_executions (status, scheduled_date) VALUES ('pending', NOW())`) + require.NoError(t, err) + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 158)) + assert.True(t, columnExists(), "column present after 000158") + var nullRows, total int + require.NoError(t, pool.QueryRow(ctx, + `SELECT count(*) FILTER (WHERE first_dispatch_at IS NULL), count(*) FROM purchase_executions`).Scan(&nullRows, &total)) + assert.Equal(t, 1, total) + assert.Equal(t, total, nullRows, "existing rows stay NULL") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 158))) + assert.False(t, columnExists(), "the down migration drops the column") +} From 53dc230ae77df99e073ee5b38e1c7826b533ffa8 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Wed, 4 Nov 2026 21:40:36 +0100 Subject: [PATCH 2/4] feat(config): stamp the first dispatch of a purchase lineage Part of #702 (Redshift re-drive idempotency, PR1 commit 2 of 3). StampFirstDispatchTx sets first_dispatch_at on the claimed running row in the caller's transaction and returns it with a DB-clock horizon check: - COALESCE keeps an existing stamp, so a re-drive sees the same time. - A fresh stamp is minted only at attempt 0, or when the predecessor ended no_submission at attempt 0. A retry of a possibly submitted lineage with no stamp gets none, which callers treat as a refusal. - Taking the caller's transaction lets the dispatch gate stamp and check conflicts atomically, so a rollback undoes the mint. Not called yet; the next commit wires it into dispatch. Co-Authored-By: Claude Opus 5.5 --- internal/config/interfaces.go | 5 + .../config/store_postgres_first_dispatch.go | 50 +++++ ...ostgres_first_dispatch_integration_test.go | 173 ++++++++++++++++++ internal/mocks/stores.go | 11 ++ 4 files changed, 239 insertions(+) create mode 100644 internal/config/store_postgres_first_dispatch.go create mode 100644 internal/config/store_postgres_first_dispatch_integration_test.go diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 482b5a42f..ec2d93781 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -792,6 +792,11 @@ type StoreInterface interface { SaveExecutionIfStatusTx(ctx context.Context, tx pgx.Tx, execution *PurchaseExecution, expected []string) error RecoverReapedExecutionTx(ctx context.Context, tx pgx.Tx, execution *PurchaseExecution) error + // StampFirstDispatchTx sets the lineage's first_dispatch_at on the + // running row (kept once set; minted only per the mint rule) and reports + // whether it is within horizon on the database clock. + StampFirstDispatchTx(ctx context.Context, tx pgx.Tx, executionID string, horizon time.Duration) (StampResult, 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_first_dispatch.go b/internal/config/store_postgres_first_dispatch.go new file mode 100644 index 000000000..ca8de1ca8 --- /dev/null +++ b/internal/config/store_postgres_first_dispatch.go @@ -0,0 +1,50 @@ +package config + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" +) + +// StampResult is the outcome of StampFirstDispatchTx. At is nil when no stamp +// exists and none may be minted; WithinHorizon is judged on the database +// clock and is false when At is nil. +type StampResult struct { + At *time.Time + WithinHorizon bool +} + +// StampFirstDispatchTx stamps a lineage's first dispatch time on the claimed +// running row, with the database clock, and returns it. An existing stamp is +// kept. A stamp is minted only at attempt 0, or when the predecessor is a +// no_submission attempt 0 (the whole lineage never submitted); otherwise a +// stamp-less successor gets At == nil and the caller must refuse it. A row +// that is not running returns ErrExecutionNotInExpectedStatus. +// +// It takes a tx so the dispatch gate can stamp and check conflicts atomically. +func (s *PostgresStore) StampFirstDispatchTx(ctx context.Context, tx pgx.Tx, executionID string, horizon time.Duration) (StampResult, error) { + var res StampResult + var within *bool + err := tx.QueryRow(ctx, ` + UPDATE purchase_executions + SET first_dispatch_at = COALESCE(first_dispatch_at, CASE + WHEN retry_attempt_n = 0 OR EXISTS ( + SELECT 1 FROM purchase_executions p + WHERE p.retry_execution_id = purchase_executions.execution_id + AND p.no_submission AND p.retry_attempt_n = 0 + ) THEN now() END) + WHERE execution_id = $1 AND status = 'running' + RETURNING first_dispatch_at, first_dispatch_at > now() - make_interval(secs => $2)`, + executionID, horizon.Seconds()).Scan(&res.At, &within) + if errors.Is(err, pgx.ErrNoRows) { + return StampResult{}, fmt.Errorf("%w: execution %s is not running", ErrExecutionNotInExpectedStatus, executionID) + } + if err != nil { + return StampResult{}, fmt.Errorf("stamp first dispatch of execution %s: %w", executionID, err) + } + res.WithinHorizon = within != nil && *within + return res, nil +} diff --git a/internal/config/store_postgres_first_dispatch_integration_test.go b/internal/config/store_postgres_first_dispatch_integration_test.go new file mode 100644 index 000000000..46906de3b --- /dev/null +++ b/internal/config/store_postgres_first_dispatch_integration_test.go @@ -0,0 +1,173 @@ +//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 stampTestHorizon = 24 * time.Hour + +func stampTx(ctx context.Context, t *testing.T, store *PostgresStore, id string) (StampResult, error) { + t.Helper() + var res StampResult + err := store.WithTx(ctx, func(tx pgx.Tx) error { + var e error + res, e = store.StampFirstDispatchTx(ctx, tx, id, stampTestHorizon) + return e + }) + return res, err +} + +func storedStamp(ctx context.Context, t *testing.T, store *PostgresStore, id string) *time.Time { + t.Helper() + var v *time.Time + require.NoError(t, store.db.QueryRow(ctx, `SELECT first_dispatch_at FROM purchase_executions WHERE execution_id = $1`, id).Scan(&v)) + return v +} + +// seedPredecessor saves a failed, stamp-less row pointing at successorID. +func seedPredecessor(ctx context.Context, t *testing.T, store *PostgresStore, successorID string, attemptN int, noSubmission bool) { + t.Helper() + pred := newNoSubmissionExec(ctx, t, store, "failed") + _, err := store.db.Exec(ctx, `UPDATE purchase_executions SET retry_execution_id = $2, retry_attempt_n = $3, no_submission = $4 WHERE execution_id = $1`, + pred.ExecutionID, successorID, attemptN, noSubmission) + require.NoError(t, err) +} + +func seedRunningSuccessor(ctx context.Context, t *testing.T, store *PostgresStore, attemptN int, stamp *time.Time) *PurchaseExecution { + t.Helper() + succ := newNoSubmissionExec(ctx, t, store, "running") + _, err := store.db.Exec(ctx, `UPDATE purchase_executions SET retry_attempt_n = $2, first_dispatch_at = $3 WHERE execution_id = $1`, succ.ExecutionID, attemptN, stamp) + require.NoError(t, err) + return succ +} + +func TestStampFirstDispatch_MintsOnceAndIsStable(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + exec := newNoSubmissionExec(ctx, t, store, "running") + + first, err := stampTx(ctx, t, store, exec.ExecutionID) + require.NoError(t, err) + require.NotNil(t, first.At) + assert.True(t, first.WithinHorizon) + assert.WithinDuration(t, time.Now(), *first.At, time.Minute) + + time.Sleep(20 * time.Millisecond) + second, err := stampTx(ctx, t, store, exec.ExecutionID) + require.NoError(t, err) + require.NotNil(t, second.At) + assert.True(t, first.At.Equal(*second.At), "a second stamp must return the first one") + assert.True(t, first.At.Equal(*storedStamp(ctx, t, store, exec.ExecutionID))) +} + +func TestStampFirstDispatch_RetrySuccessorKeepsCopiedStamp(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + copied := time.Now().Add(-2 * time.Hour).UTC().Truncate(time.Microsecond) + succ := seedRunningSuccessor(ctx, t, store, 1, &copied) + + res, err := stampTx(ctx, t, store, succ.ExecutionID) + require.NoError(t, err) + require.NotNil(t, res.At) + assert.True(t, copied.Equal(*res.At), "a copied stamp must not be re-minted") + assert.True(t, res.WithinHorizon) +} + +func TestStampFirstDispatch_RetryWithoutStampIsNotMinted(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + succ := seedRunningSuccessor(ctx, t, store, 1, nil) + seedPredecessor(ctx, t, store, succ.ExecutionID, 0, false) + + res, err := stampTx(ctx, t, store, succ.ExecutionID) + require.NoError(t, err) + assert.Nil(t, res.At) + assert.False(t, res.WithinHorizon) + assert.Nil(t, storedStamp(ctx, t, store, succ.ExecutionID)) +} + +func TestStampFirstDispatch_RetryOfNoSubmissionAttempt0Mints(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + succ := seedRunningSuccessor(ctx, t, store, 1, nil) + seedPredecessor(ctx, t, store, succ.ExecutionID, 0, true) + + res, err := stampTx(ctx, t, store, succ.ExecutionID) + require.NoError(t, err) + require.NotNil(t, res.At) + assert.True(t, res.WithinHorizon) +} + +// A chain of length >= 2 may have submitted at attempt 0 even though the +// immediate predecessor carries no_submission, so it must not mint. +func TestStampFirstDispatch_RetryOfNoSubmissionAttempt1DoesNotMint(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + succ := seedRunningSuccessor(ctx, t, store, 2, nil) + seedPredecessor(ctx, t, store, succ.ExecutionID, 1, true) + + res, err := stampTx(ctx, t, store, succ.ExecutionID) + require.NoError(t, err) + assert.Nil(t, res.At) + assert.Nil(t, storedStamp(ctx, t, store, succ.ExecutionID)) +} + +func TestStampFirstDispatch_NotRunning(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + for _, status := range []string{"approved", "failed", "completed"} { + exec := newNoSubmissionExec(ctx, t, store, status) + _, err := stampTx(ctx, t, store, exec.ExecutionID) + require.ErrorIs(t, err, ErrExecutionNotInExpectedStatus, status) + assert.Nil(t, storedStamp(ctx, t, store, exec.ExecutionID), status) + } +} + +// The horizon is judged by the database clock, not the app's. +func TestStampFirstDispatch_WithinHorizonUsesDBClock(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + old := time.Now().Add(-stampTestHorizon - time.Hour).UTC() + recent := time.Now().Add(-stampTestHorizon + time.Hour).UTC() + + pastSucc := seedRunningSuccessor(ctx, t, store, 1, &old) + res, err := stampTx(ctx, t, store, pastSucc.ExecutionID) + require.NoError(t, err) + require.NotNil(t, res.At) + assert.False(t, res.WithinHorizon) + + inSucc := seedRunningSuccessor(ctx, t, store, 1, &recent) + res, err = stampTx(ctx, t, store, inSucc.ExecutionID) + require.NoError(t, err) + assert.True(t, res.WithinHorizon) +} + +// now() is the tx start, so a stamp seeded at now() - H inside the same tx is +// exactly H old: that is outside the horizon. +func TestStampFirstDispatch_ExactHorizonIsOutside(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + succ := seedRunningSuccessor(ctx, t, store, 1, nil) + + var res StampResult + require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { + if _, err := tx.Exec(ctx, `UPDATE purchase_executions SET first_dispatch_at = now() - make_interval(secs => $2) WHERE execution_id = $1`, + succ.ExecutionID, stampTestHorizon.Seconds()); err != nil { + return err + } + var err error + res, err = store.StampFirstDispatchTx(ctx, tx, succ.ExecutionID, stampTestHorizon) + return err + })) + require.NotNil(t, res.At) + assert.False(t, res.WithinHorizon) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 0c9db1eb9..5937f85e8 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -539,6 +539,17 @@ func (m *MockConfigStore) RecoverReapedExecutionTx(ctx context.Context, tx pgx.T return args.Error(0) } +// StampFirstDispatchTx mocks the first-dispatch stamp (no default: testify panics if unexpected). +func (m *MockConfigStore) StampFirstDispatchTx(ctx context.Context, tx pgx.Tx, executionID string, horizon time.Duration) (config.StampResult, error) { + m.record("StampFirstDispatchTx", ctx, tx, executionID, horizon) + args := m.Called(ctx, tx, executionID, horizon) + res, ok := args.Get(0).(config.StampResult) + if !ok { + m.reporter.wrongReturnType("StampFirstDispatchTx", "config.StampResult", 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) From 3a6caba5aac55c46beb6d2c633ac3e647a459ab6 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Wed, 4 Nov 2026 23:15:28 +0100 Subject: [PATCH 3/4] fix(purchase): pass the first-dispatch stamp to Redshift buys Part of #702 (Redshift re-drive idempotency, PR1 commit 3 of 3). executeAndFinalize stamps first_dispatch_at for an execution with a selected Redshift rec, after the pre-execution refusals and before the provider call, and the stamp becomes PurchaseOptions.RequestedAt in place of the per-dispatch time.Now() from #1138. A re-drive of the same lineage therefore hands the library's DescribeReservedNodes dedupe the same window start as the first dispatch. - Fan-out per-account rows inherit the root stamp by struct copy and are never stamped themselves. - A store error at attempt 0 fails before submission. A missing stamp on a later attempt is a plain error, so a possibly submitted lineage is never re-stamped. - A Redshift rec that reaches the provider without a stamp is refused instead of buying with a zero RequestedAt. Redshift re-drive and retry stay refused in this PR. Co-Authored-By: Claude Opus 5.5 --- internal/purchase/armed_redrive_test.go | 3 + internal/purchase/coverage_extra_test.go | 3 + internal/purchase/execution.go | 11 +- internal/purchase/manager.go | 3 + internal/purchase/redshift_lineage.go | 76 ++++++++ internal/purchase/requested_at_test.go | 232 +++++++++++++++++++++++ 6 files changed, 326 insertions(+), 2 deletions(-) create mode 100644 internal/purchase/redshift_lineage.go diff --git a/internal/purchase/armed_redrive_test.go b/internal/purchase/armed_redrive_test.go index fe123f936..a4acafe9e 100644 --- a/internal/purchase/armed_redrive_test.go +++ b/internal/purchase/armed_redrive_test.go @@ -108,6 +108,9 @@ func armedHarness(t *testing.T, serviceType common.ServiceType) (*Manager, *Mock store.On("SavePurchaseHistory", mock.Anything, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil).Maybe() store.On("CompletePlanStep", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{}, nil).Maybe() + stampedAt := time.Now().UTC() + store.On("StampFirstDispatchTx", mock.Anything, mock.Anything, mock.Anything, mock.Anything). + Return(config.StampResult{At: &stampedAt, WithinHorizon: true}, 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/coverage_extra_test.go b/internal/purchase/coverage_extra_test.go index ddaa4186d..f742c7fe1 100644 --- a/internal/purchase/coverage_extra_test.go +++ b/internal/purchase/coverage_extra_test.go @@ -1378,9 +1378,12 @@ func TestManager_ExecuteSinglePurchase_DetailsByService(t *testing.T) { require.NotNil(t, detailsBlob, "test seed must produce a non-empty Details blob") plan := &config.PurchasePlan{ID: "plan-" + tc.name, Name: "Plan"} + stampedAt := time.Now().UTC() exec := &config.PurchaseExecution{ ExecutionID: "exec-" + tc.name, PlanID: "plan-" + tc.name, + // Redshift dispatch requires the lineage stamp executeAndFinalize mints. + FirstDispatchAt: &stampedAt, Recommendations: []config.RecommendationRecord{ { Provider: "aws", diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 56a821888..ecf287ae7 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -642,6 +642,9 @@ func (m *Manager) processPurchaseRecommendations(ctx context.Context, exec *conf Source: m.normalizePurchaseSource(exec), ExecutionID: exec.ExecutionID, } + if exec.FirstDispatchAt != nil { + opts.RequestedAt = exec.FirstDispatchAt.UTC() + } // Load GlobalConfig to pick up the OfferingClass setting for EC2 RI // purchases. A load error is fatal: proceeding with an empty OfferingClass @@ -1218,8 +1221,12 @@ func (m *Manager) executeSinglePurchase(ctx context.Context, rec config.Recommen // (which the AWS SDK retries up to 3× × 30s by default) versus // something earlier in the flow — issue #667. tCall := time.Now() - // go's Redshift dedupe needs the dispatch time to bound its DescribeReservedNodes match window. - opts.RequestedAt = time.Now().UTC() + // go's Redshift dedupe bounds its DescribeReservedNodes match window by RequestedAt, which must + // be the lineage's first dispatch time, never this call's. + opts, err = withDispatchTime(rec, opts) + if err != nil { + return common.PurchaseResult{}, err + } result, err := serviceClient.PurchaseCommitment(recCtx, recommendation, opts) elapsed := time.Since(tCall) if err != nil { diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index 1f75a81b0..e0c9b1122 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -326,6 +326,9 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE // Last line of defense before money moves (issue #1718). Every executor // entry point funnels through here, so one check covers all of them. execErr := m.preExecutionRefusal(execCtx, exec) + if execErr == nil { + execErr = m.stampRedshiftDispatch(execCtx, exec) + } if execErr == nil { execErr = m.executePurchase(execCtx, exec) } diff --git a/internal/purchase/redshift_lineage.go b/internal/purchase/redshift_lineage.go new file mode 100644 index 000000000..548de4545 --- /dev/null +++ b/internal/purchase/redshift_lineage.go @@ -0,0 +1,76 @@ +package purchase + +import ( + "context" + "fmt" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/jackc/pgx/v5" +) + +// redshiftRedriveHorizon is how long after a lineage's first dispatch a +// successor may still be stamped as within horizon. +const redshiftRedriveHorizon = 24 * time.Hour + +// isRedshiftRec reports whether rec dispatches to the AWS Redshift purchase +// path. It goes through mapServiceSlug, the table the executor dispatches on. +func isRedshiftRec(rec config.RecommendationRecord) bool { + if rec.Provider != "" && rec.Provider != string(common.ProviderAWS) { + return false + } + svc, ok := mapServiceSlug(rec.Service) + return ok && (svc == common.ServiceRedshift || svc == common.ServiceDataWarehouse) +} + +func hasRedshiftRec(exec *config.PurchaseExecution) bool { + for i := range exec.Recommendations { + if exec.Recommendations[i].Selected && isRedshiftRec(exec.Recommendations[i]) { + return true + } + } + 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. +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 + }) + 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 { + return beforeSubmission(err) + } + return err + } + at := res.At.UTC() + exec.FirstDispatchAt = &at + return 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) { + if !opts.RequestedAt.IsZero() { + return opts, nil + } + if isRedshiftRec(rec) { + return opts, fmt.Errorf("purchase[%s]: Redshift dispatch without a first dispatch time", opts.ExecutionID) + } + opts.RequestedAt = time.Now().UTC() + return opts, nil +} diff --git a/internal/purchase/requested_at_test.go b/internal/purchase/requested_at_test.go index c1fdd803d..d71973f7b 100644 --- a/internal/purchase/requested_at_test.go +++ b/internal/purchase/requested_at_test.go @@ -2,6 +2,7 @@ package purchase import ( "context" + "errors" "sync" "testing" "time" @@ -29,8 +30,11 @@ func runFanOutCapturingOpts(t *testing.T, service common.ServiceType, mk func(st mockServiceClient := new(MockServiceClient) assertMocksOnCleanup(t, mockStore, mockEmail, mockFactory, mockProviderInst, mockServiceClient) + // Redshift dispatch needs the lineage stamp executeAndFinalize mints; direct callers set it. + stampedAt := time.Now().UTC() exec := &config.PurchaseExecution{ ExecutionID: "root-exec", IdempotencyKey: "root-lineage", PlanID: "plan-x", + FirstDispatchAt: &stampedAt, Source: common.PurchaseSourceWeb, Recommendations: []config.RecommendationRecord{mk("acct-A"), mk("acct-B")}, } @@ -124,3 +128,231 @@ func TestExecuteSinglePurchase_EmptyTokenOptsOtherwiseUntouched(t *testing.T) { got.RequestedAt = time.Time{} assert.Equal(t, in, got) } + +type redshiftFinalizeRun struct { + events []string + stampedIDs []string + opts []common.PurchaseOptions + saves []config.PurchaseExecution + err error +} + +// runRedshiftFinalize drives executeAndFinalize for exec with a capturing +// provider. A nil stamp registers no StampFirstDispatchTx expectation, so any +// stamp call panics the mock. +func runRedshiftFinalize(t *testing.T, exec *config.PurchaseExecution, accounts []config.CloudAccount, service common.ServiceType, + stamp func() (config.StampResult, error)) *redshiftFinalizeRun { + t.Helper() + ctx := context.Background() + mockStore := new(MockConfigStore) + mockEmail := new(MockEmailSender) + mockFactory := new(MockProviderFactory) + mockProviderInst := new(MockProvider) + mockServiceClient := new(MockServiceClient) + run := &redshiftFinalizeRun{} + var mu sync.Mutex + + mockStore.SavePurchaseExecutionFn = func(_ context.Context, e *config.PurchaseExecution) error { + mu.Lock() + run.saves = append(run.saves, *e) + mu.Unlock() + return nil + } + mockStore.GetPurchasePlanFn = func(_ context.Context, id string) (*config.PurchasePlan, error) { + return &config.PurchasePlan{ID: id, Name: "Plan X"}, nil + } + mockStore.GetPlanAccountsFn = func(context.Context, string) ([]config.CloudAccount, error) { return accounts, nil } + if stamp != nil { + res, stampErr := stamp() + mockStore.On("StampFirstDispatchTx", mock.Anything, mock.Anything, mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { + mu.Lock() + run.events = append(run.events, "stamp") + run.stampedIDs = append(run.stampedIDs, args.String(2)) + mu.Unlock() + }).Return(res, stampErr) + } + mockStore.On("CompletePlanStep", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() + mockStore.On("SavePurchaseHistory", mock.Anything, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil).Maybe() + mockEmail.On("SendPurchaseConfirmation", mock.Anything, mock.Anything).Return(nil).Maybe() + mockEmail.On("SendPurchaseFailedNotification", mock.Anything, mock.Anything).Return(nil).Maybe() + mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProviderInst, nil).Maybe() + mockProviderInst.On("GetServiceClient", mock.Anything, service, mock.Anything).Return(mockServiceClient, nil).Maybe() + mockServiceClient.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { + mu.Lock() + run.events = append(run.events, "provider") + run.opts = append(run.opts, args.Get(2).(common.PurchaseOptions)) + mu.Unlock() + }).Return(common.PurchaseResult{Success: true, CommitmentID: "ri-ok"}, nil).Maybe() + + manager := &Manager{ + config: mockStore, email: mockEmail, providerFactory: mockFactory, + credStore: awsAccessKeyCredStore(), dashboardURL: "https://dashboard.example.com", + } + run.err = manager.executeAndFinalize(ctx, exec) + return run +} + +func redshiftScopedRec(accountID string) config.RecommendationRecord { + r := scopedTestRec(accountID) + r.Service, r.ResourceType = "redshift", "ra3.xlplus" + return r +} + +func stampAt(at time.Time) func() (config.StampResult, error) { + return func() (config.StampResult, error) { return config.StampResult{At: &at, WithinHorizon: true}, nil } +} + +func TestExecuteAndFinalize_RedshiftStampsBeforeProvider(t *testing.T) { + at := time.Now().UTC().Add(-90 * time.Minute).Truncate(time.Microsecond) + exec := &config.PurchaseExecution{ + ExecutionID: "rs-exec", IdempotencyKey: "rs-lineage", Status: "running", + Recommendations: []config.RecommendationRecord{redshiftScopedRec("acct-A")}, + } + run := runRedshiftFinalize(t, exec, nil, common.ServiceRedshift, stampAt(at)) + + require.NoError(t, run.err) + assert.Equal(t, []string{"stamp", "provider"}, run.events) + assert.Equal(t, []string{"rs-exec"}, run.stampedIDs) + require.Len(t, run.opts, 1) + assert.True(t, run.opts[0].RequestedAt.Equal(at), "RequestedAt %s must be the stamp %s, not a per-dispatch clock", run.opts[0].RequestedAt, at) +} + +func TestExecuteAndFinalize_RedshiftFanOutAccountsShareRootStamp(t *testing.T) { + at := time.Now().UTC().Add(-3 * time.Hour).Truncate(time.Microsecond) + accounts := []config.CloudAccount{ + {ID: "acct-A", Name: "A", Provider: "aws", ExternalID: "111111111111", AWSAuthMode: "access_keys"}, + {ID: "acct-B", Name: "B", Provider: "aws", ExternalID: "222222222222", AWSAuthMode: "access_keys"}, + } + exec := &config.PurchaseExecution{ + ExecutionID: "root-exec", IdempotencyKey: "root-lineage", PlanID: "plan-x", Status: "running", + Recommendations: []config.RecommendationRecord{redshiftScopedRec("acct-A"), redshiftScopedRec("acct-B")}, + } + run := runRedshiftFinalize(t, exec, accounts, common.ServiceRedshift, stampAt(at)) + + require.NoError(t, run.err) + assert.Equal(t, []string{"root-exec"}, run.stampedIDs, "only the root is stamped; per-account rows inherit the stamp") + require.Len(t, run.opts, 2) + for _, o := range run.opts { + assert.True(t, o.RequestedAt.Equal(at), "every account dispatch must see the root stamp, got %s", o.RequestedAt) + } + perAccount := 0 + for _, s := range run.saves { + if s.CloudAccountID == nil { + continue + } + perAccount++ + require.NotNil(t, s.FirstDispatchAt, "per-account row %s must carry the stamp into its INSERT", s.ExecutionID) + assert.True(t, s.FirstDispatchAt.Equal(at)) + } + assert.GreaterOrEqual(t, perAccount, 2) +} + +func TestExecuteAndFinalize_RedshiftStampErrorFailsBeforeSubmission(t *testing.T) { + exec := &config.PurchaseExecution{ + ExecutionID: "rs-exec", IdempotencyKey: "rs-lineage", Status: "running", + Recommendations: []config.RecommendationRecord{redshiftScopedRec("acct-A")}, + } + run := runRedshiftFinalize(t, exec, nil, common.ServiceRedshift, func() (config.StampResult, error) { + return config.StampResult{}, errors.New("db down") + }) + + require.Error(t, run.err) + assert.Empty(t, run.opts, "no provider call after a failed stamp") + assert.Equal(t, "failed", exec.Status) + assert.True(t, exec.NoSubmission) +} + +func TestIsRedshiftRec_MatcherTable(t *testing.T) { + cases := []struct { + name string + provider string + service string + want bool + }{ + {"redshift, empty provider", "", "redshift", true}, + {"redshift, aws", "aws", "redshift", true}, + {"data-warehouse, empty provider", "", "data-warehouse", true}, + {"data-warehouse, aws", "aws", "data-warehouse", true}, + {"data-warehouse, azure", "azure", "data-warehouse", false}, + {"redshift slug, gcp", "gcp", "redshift", false}, + {"ec2, aws", "aws", "ec2", false}, + {"unknown slug, aws", "aws", "no-such-service", false}, + {"empty service, aws", "aws", "", false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rec := config.RecommendationRecord{Provider: tc.provider, Service: tc.service, Selected: true} + assert.Equal(t, tc.want, isRedshiftRec(rec)) + exec := &config.PurchaseExecution{Recommendations: []config.RecommendationRecord{rec}} + assert.Equal(t, tc.want, hasRedshiftRec(exec)) + }) + } + + t.Run("unselected redshift does not stamp", func(t *testing.T) { + rec := config.RecommendationRecord{Provider: "aws", Service: "redshift", Selected: false} + assert.False(t, hasRedshiftRec(&config.PurchaseExecution{Recommendations: []config.RecommendationRecord{rec}})) + }) + t.Run("one selected redshift among others stamps", func(t *testing.T) { + recs := []config.RecommendationRecord{ + {Provider: "aws", Service: "ec2", Selected: true}, + {Provider: "aws", Service: "redshift", Selected: false}, + {Provider: "aws", Service: "data-warehouse", Selected: true}, + } + assert.True(t, hasRedshiftRec(&config.PurchaseExecution{Recommendations: recs})) + }) +} + +func TestStampRedshiftDispatch_NilAtRefusalByAttempt(t *testing.T) { + for _, tc := range []struct { + attempt int + wantBeforeSubmit bool + }{{0, true}, {1, false}} { + store := new(MockConfigStore) + store.On("StampFirstDispatchTx", mock.Anything, mock.Anything, "rs-x", mock.Anything). + Return(config.StampResult{At: nil}, nil).Once() + m := &Manager{config: store} + exec := &config.PurchaseExecution{ + ExecutionID: "rs-x", RetryAttemptN: tc.attempt, + Recommendations: []config.RecommendationRecord{redshiftScopedRec("acct-A")}, + } + + var err error + require.NotPanics(t, func() { err = m.stampRedshiftDispatch(context.Background(), exec) }) + + require.Error(t, err) + var bs *beforeSubmissionError + assert.Equal(t, tc.wantBeforeSubmit, errors.As(err, &bs), "attempt %d", tc.attempt) + assert.Nil(t, exec.FirstDispatchAt) + store.AssertNumberOfCalls(t, "StampFirstDispatchTx", 1) + } +} + +func TestExecuteAndFinalize_NonRedshiftNeverStamps(t *testing.T) { + exec := &config.PurchaseExecution{ + ExecutionID: "ec2-exec", IdempotencyKey: "ec2-lineage", Status: "running", + Recommendations: []config.RecommendationRecord{scopedTestRec("acct-A")}, + } + before := time.Now() + run := runRedshiftFinalize(t, exec, nil, common.ServiceEC2, nil) + + require.NoError(t, run.err) + assert.Empty(t, run.stampedIDs) + require.Len(t, run.opts, 1) + assertRequestedAtStamped(t, []common.PurchaseOptions{run.opts[0], run.opts[0]}, before, time.Now()) +} + +func TestExecuteSinglePurchase_RedshiftWithoutRequestedAtRefuses(t *testing.T) { + mockFactory := new(MockProviderFactory) + mockProviderInst := new(MockProvider) + mockServiceClient := new(MockServiceClient) + mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProviderInst, nil).Maybe() + mockProviderInst.On("GetServiceClient", mock.Anything, common.ServiceRedshift, mock.Anything).Return(mockServiceClient, nil).Maybe() + m := &Manager{providerFactory: mockFactory} + + _, err := m.executeSinglePurchase(context.Background(), redshiftScopedRec("acct-A"), nil, common.PurchaseOptions{ExecutionID: "e"}) + + require.Error(t, err) + mockServiceClient.AssertNotCalled(t, "PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything) +} From fe14519072f8aebcf1ac88c01c059000ae35a2bf Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 5 Nov 2026 12:58:30 +0100 Subject: [PATCH 4/4] chore(db): renumber the first_dispatch_at migration to 000154 Part of #702. Migration numbers are now assigned at merge time as the next free number on main (000153 is the highest today). Rename only: the migration SQL is unchanged; its test and one comment follow the number. Co-Authored-By: Claude Opus 5.5 --- internal/config/types.go | 2 +- ...purchase_executions_first_dispatch_at.down.sql} | 0 ...4_purchase_executions_first_dispatch_at.up.sql} | 0 ..._purchase_executions_first_dispatch_at_test.go} | 14 +++++++------- 4 files changed, 8 insertions(+), 8 deletions(-) rename internal/database/postgres/migrations/{000158_purchase_executions_first_dispatch_at.down.sql => 000154_purchase_executions_first_dispatch_at.down.sql} (100%) rename internal/database/postgres/migrations/{000158_purchase_executions_first_dispatch_at.up.sql => 000154_purchase_executions_first_dispatch_at.up.sql} (100%) rename internal/database/postgres/migrations/{000158_purchase_executions_first_dispatch_at_test.go => 000154_purchase_executions_first_dispatch_at_test.go} (81%) diff --git a/internal/config/types.go b/internal/config/types.go index 390a6e1d4..64363e306 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -401,7 +401,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 000158). + // Server-only; written once at INSERT, never updated (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/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.down.sql b/internal/database/postgres/migrations/000154_purchase_executions_first_dispatch_at.down.sql similarity index 100% rename from internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.down.sql rename to internal/database/postgres/migrations/000154_purchase_executions_first_dispatch_at.down.sql diff --git a/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.up.sql b/internal/database/postgres/migrations/000154_purchase_executions_first_dispatch_at.up.sql similarity index 100% rename from internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at.up.sql rename to internal/database/postgres/migrations/000154_purchase_executions_first_dispatch_at.up.sql diff --git a/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at_test.go b/internal/database/postgres/migrations/000154_purchase_executions_first_dispatch_at_test.go similarity index 81% rename from internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at_test.go rename to internal/database/postgres/migrations/000154_purchase_executions_first_dispatch_at_test.go index 7c8378154..f3890d298 100644 --- a/internal/database/postgres/migrations/000158_purchase_executions_first_dispatch_at_test.go +++ b/internal/database/postgres/migrations/000154_purchase_executions_first_dispatch_at_test.go @@ -12,9 +12,9 @@ import ( "github.com/stretchr/testify/require" ) -// Migration 000158 adds the nullable purchase_executions.first_dispatch_at: +// Migration 000154 adds the nullable purchase_executions.first_dispatch_at: // existing rows stay NULL and the down migration drops it. -func TestMigration_0158_FirstDispatchAt(t *testing.T) { +func TestMigration_0154_FirstDispatchAt(t *testing.T) { t.Parallel() ctx := context.Background() migrationsPath := getMigrationsPath() @@ -30,19 +30,19 @@ func TestMigration_0158_FirstDispatchAt(t *testing.T) { return exists } - require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 158))) - assert.False(t, columnExists(), "column absent before 000158") + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 154))) + assert.False(t, columnExists(), "column absent before 000154") _, err := pool.Exec(ctx, `INSERT INTO purchase_executions (status, scheduled_date) VALUES ('pending', NOW())`) require.NoError(t, err) - require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 158)) - assert.True(t, columnExists(), "column present after 000158") + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 154)) + assert.True(t, columnExists(), "column present after 000154") var nullRows, total int require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FILTER (WHERE first_dispatch_at IS NULL), count(*) FROM purchase_executions`).Scan(&nullRows, &total)) assert.Equal(t, 1, total) assert.Equal(t, total, nullRows, "existing rows stay NULL") - require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 158))) + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 154))) assert.False(t, columnExists(), "the down migration drops the column") }