From 3e1c4eb0df7abecd6a744c6b9ff459997ac96d75 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 16 Oct 2026 19:01:16 +0200 Subject: [PATCH] test(plans): prove legacy writers recheck plans an editor Save changed Add race tests for the legacy writers that can run while an editor Save holds the plan: the notifier's new schedule root, a retry successor, a fan-out child, execution pruning and cloud account deletion. Each test queues the writer behind a stand-in for the Save's lock footprint, then checks the outcome in both orders: a writer that waited sees the committed editor state and skips or refuses, and a writer that won is seen by the next capture. The shared footprint and lock-wait helpers move to testhelpers so the Save command's own race tests can swap in the real command. Two pre-existing gaps are recorded in known_issues (#998, #999). Refs #631 Co-Authored-By: Claude Opus 5.5 --- ...s_managed_add_accounts_integration_test.go | 20 +--- ...ler_retry_managed_race_integration_test.go | 63 +++++++++++ ...e_postgres_writer_race_integration_test.go | 95 ++++++++++++++++ .../database/postgres/testhelpers/lockwait.go | 104 ++++++++++++++++++ ...plan_step_managed_race_integration_test.go | 60 ++++++++++ known_issues/35_legacy_guard_followups.md | 27 +++++ 6 files changed, 354 insertions(+), 15 deletions(-) create mode 100644 internal/api/handler_retry_managed_race_integration_test.go create mode 100644 internal/config/store_postgres_writer_race_integration_test.go create mode 100644 internal/database/postgres/testhelpers/lockwait.go create mode 100644 internal/purchase/plan_step_managed_race_integration_test.go create mode 100644 known_issues/35_legacy_guard_followups.md diff --git a/internal/api/handler_plans_managed_add_accounts_integration_test.go b/internal/api/handler_plans_managed_add_accounts_integration_test.go index aaa5d8a2..54e2dcb5 100644 --- a/internal/api/handler_plans_managed_add_accounts_integration_test.go +++ b/internal/api/handler_plans_managed_add_accounts_integration_test.go @@ -13,7 +13,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" - "github.com/jackc/pgx/v5" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" "github.com/stretchr/testify/require" ) @@ -45,25 +45,15 @@ func (f *createConcurrencyFixture) extraAccount(ctx context.Context, t *testing. return acct.ID } -// holdPlanLock stands in for the editor command: an open transaction holding -// the plan row FOR UPDATE, rolled back at test end unless the test commits it. -func (f *createConcurrencyFixture) holdPlanLock(ctx context.Context, t *testing.T) pgx.Tx { +// holdPlanLock is the shared command footprint (plan, executions, accounts). +func (f *createConcurrencyFixture) holdPlanLock(ctx context.Context, t *testing.T) *testhelpers.Footprint { t.Helper() - command, err := f.pool.Begin(ctx) - require.NoError(t, err) - t.Cleanup(func() { _ = command.Rollback(ctx) }) - _, err = command.Exec(ctx, `SELECT 1 FROM purchase_plans WHERE id = $1 FOR UPDATE`, f.planID) - require.NoError(t, err) - return command + return testhelpers.HoldCommandFootprint(ctx, t, f.pool, f.planID) } func awaitLockWait(ctx context.Context, t *testing.T, f *createConcurrencyFixture) { t.Helper() - require.Eventually(t, func() bool { - var n int - require.NoError(t, f.pool.QueryRow(ctx, `SELECT count(*) FROM pg_stat_activity WHERE wait_event_type = 'Lock'`).Scan(&n)) - return n > 0 - }, 15*time.Second, 50*time.Millisecond, "writer must wait on the plan lock") + testhelpers.WaitForLockWait(ctx, t, f.pool, "") } const ( diff --git a/internal/api/handler_retry_managed_race_integration_test.go b/internal/api/handler_retry_managed_race_integration_test.go new file mode 100644 index 00000000..476f8e5f --- /dev/null +++ b/internal/api/handler_retry_managed_race_integration_test.go @@ -0,0 +1,63 @@ +//go:build integration + +package api + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +// W17: the retry successor insert holds a FK key-share on the plan row, so it +// queues behind the command's plan lock, and its link CAS rechecks the +// original after the wait. Retries are never refused on a managed plan: a +// failed row is fixed in the editor. +func retryRace(ctx context.Context, t *testing.T) (*createConcurrencyFixture, *config.PurchaseExecution, func() error) { + f := newCreateConcurrencyFixture(ctx, t) + failed := &config.PurchaseExecution{ExecutionID: uuid.NewString(), PlanID: f.planID, Status: "failed", StepNumber: 3, ScheduledDate: time.Now()} + require.NoError(t, f.store.SavePurchaseExecution(ctx, failed)) + successor := &config.PurchaseExecution{ExecutionID: uuid.NewString(), PlanID: f.planID, Status: "pending", StepNumber: 3, ScheduledDate: time.Now(), RetryAttemptN: 1} + return f, successor, func() error { return f.handler.persistRetrySuccessorTx(ctx, failed, successor, nil) } +} + +func TestRetryWaitsForCommandThenSucceedsAndCommandSeesIt(t *testing.T) { + ctx := context.Background() + f, successor, retry := retryRace(ctx, t) + command := f.holdPlanLock(ctx, t) + done := make(chan error, 1) + go func() { done <- retry() }() + testhelpers.WaitForLockWait(ctx, t, f.pool, "%INSERT INTO purchase_executions%") + command.SetManaged(ctx, t, f.planID) + require.NoError(t, command.Commit(ctx)) + require.NoError(t, testhelpers.Await(t, done), "a retry of a failed row is allowed on a managed plan") + + require.Contains(t, f.holdPlanLock(ctx, t).Executions, successor.ExecutionID, "the next command capture sees the successor") +} + +func TestRetryLosesToCommandThatLinkedOriginalFirst(t *testing.T) { + ctx := context.Background() + f, successor, retry := retryRace(ctx, t) + command := f.holdPlanLock(ctx, t) + done := make(chan error, 1) + go func() { done <- retry() }() + testhelpers.WaitForLockWait(ctx, t, f.pool, "%INSERT INTO purchase_executions%") + other := uuid.NewString() + _, err := command.Exec(ctx, `INSERT INTO purchase_executions (execution_id, plan_id, status, step_number, scheduled_date) VALUES ($1, $2, 'pending', 3, now())`, other, f.planID) + require.NoError(t, err) + _, err = command.Exec(ctx, `UPDATE purchase_executions SET retry_execution_id = $1 WHERE plan_id = $2 AND status = 'failed'`, other, f.planID) + require.NoError(t, err) + require.NoError(t, command.Commit(ctx)) + + ce, ok := IsClientError(testhelpers.Await(t, done)) + require.True(t, ok) + require.Equal(t, 409, ce.code) + require.Contains(t, ce.message, "already retried") + require.Equal(t, 2, f.count(ctx, t, execCount), "the lost retry leaves no successor row") + _, err = f.store.GetExecutionByID(ctx, successor.ExecutionID) + require.ErrorIs(t, err, config.ErrNotFound) +} diff --git a/internal/config/store_postgres_writer_race_integration_test.go b/internal/config/store_postgres_writer_race_integration_test.go new file mode 100644 index 00000000..695b3531 --- /dev/null +++ b/internal/config/store_postgres_writer_race_integration_test.go @@ -0,0 +1,95 @@ +//go:build integration + +package config + +import ( + "context" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/stretchr/testify/require" +) + +// Audited writers (W16, W21, W12) against the editor command's lock footprint. +// None of them gains a managed-plan refusal: each either queues behind the +// command and is accepted, or rechecks its own predicate on the committed row. + +func inBackground(f func() error) <-chan error { + done := make(chan error, 1) + go func() { done <- f() }() + return done +} + +// W16: a fan-out child row after a real purchase must never be refused. +func TestFanOutChildQueuesBehindCommandAndIsAccepted(t *testing.T) { + ctx := context.Background() + f := newRampFixture(ctx, t, 1, 1, 3) + command := testhelpers.HoldCommandFootprint(ctx, t, f.store.db, f.planID) + child := &PurchaseExecution{ExecutionID: "00000000-0000-0000-0000-0000000000c1", IdempotencyKey: "root:child", PlanID: f.planID, + CloudAccountID: &f.accounts[0], Status: StatusCompleted, StepNumber: 2} + done := inBackground(func() error { return f.store.SavePurchaseExecution(ctx, child) }) + testhelpers.WaitForLockWait(ctx, t, f.store.db, "%INSERT INTO purchase_executions%") + command.SetManaged(ctx, t, f.planID) + require.NoError(t, command.Commit(ctx)) + require.NoError(t, testhelpers.Await(t, done), "audit rows after money moved are never refused") + + require.Contains(t, testhelpers.HoldCommandFootprint(ctx, t, f.store.db, f.planID).Executions, child.ExecutionID) +} + +// W21: the prune DELETE rechecks the committed row after the wait. The +// fixture ramp is finished (3 of 3): on an unfinished ramp the retention +// trigger of migration 000124 already keeps every capacity-holding row, so +// only the SQL predicate would be unexercised. +func TestPruneRechecksRowAfterCommandCancelsIt(t *testing.T) { + for _, commandWins := range []bool{true, false} { + name := map[bool]string{true: "command first", false: "prune first"}[commandWins] + t.Run(name, func(t *testing.T) { + ctx := context.Background() + f := newRampFixture(ctx, t, 0, 3, 3) + row := f.row(ctx, t, 2, nil, "pending") + _, err := f.store.db.Exec(ctx, `UPDATE purchase_executions SET expires_at = NOW() - INTERVAL '40 days' WHERE execution_id = $1`, row.ExecutionID) + require.NoError(t, err) + prune := func() error { _, pErr := f.store.CleanupOldExecutions(ctx, 30); return pErr } + + if !commandWins { + deleted, pErr := f.store.CleanupOldExecutions(ctx, 30) + require.NoError(t, pErr) + require.EqualValues(t, 1, deleted) + require.NotContains(t, testhelpers.HoldCommandFootprint(ctx, t, f.store.db, f.planID).Executions, row.ExecutionID) + return + } + command := testhelpers.HoldCommandFootprint(ctx, t, f.store.db, f.planID) + done := inBackground(prune) + testhelpers.WaitForLockWait(ctx, t, f.store.db, "%DELETE FROM purchase_executions%") + _, err = command.Exec(ctx, `UPDATE purchase_executions SET status = 'canceled', updated_at = NOW() WHERE execution_id = $1`, row.ExecutionID) + require.NoError(t, err) + require.NoError(t, command.Commit(ctx)) + require.NoError(t, testhelpers.Await(t, done)) + require.Equal(t, 1, f.countRows(ctx, t, row.ExecutionID), "the canceled row is no longer prunable and survives") + }) + } +} + +func (f *rampFixture) countRows(ctx context.Context, t *testing.T, execID string) (n int) { + t.Helper() + require.NoError(t, f.store.db.QueryRow(ctx, `SELECT count(*) FROM purchase_executions WHERE execution_id = $1`, execID).Scan(&n)) + return n +} + +// W12: deleting an account queues behind the command's account FOR SHARE, so +// the command's captured account set cannot shrink mid-flight. +func TestAccountDeleteWaitsForCommandAccountShare(t *testing.T) { + ctx := context.Background() + f := newRampFixture(ctx, t, 1, 1, 3) + command := testhelpers.HoldCommandFootprint(ctx, t, f.store.db, f.planID) + done := inBackground(func() error { return f.store.DeleteCloudAccount(ctx, f.accounts[0]) }) + testhelpers.WaitForLockWait(ctx, t, f.store.db, "%DELETE FROM cloud_accounts%") + var n int + require.NoError(t, command.QueryRow(ctx, `SELECT count(*) FROM plan_accounts WHERE plan_id = $1`, f.planID).Scan(&n)) + require.Equal(t, 1, n, "the command still sees its account while the delete waits") + require.NoError(t, command.Commit(ctx)) + require.NoError(t, testhelpers.Await(t, done)) + + require.NoError(t, f.store.db.QueryRow(ctx, `SELECT count(*) FROM plan_accounts WHERE plan_id = $1`, f.planID).Scan(&n)) + require.Zero(t, n, "the cascade runs after the command committed (plan_accounts only; ladder_configs, the recommendations cache and credentials also cascade, none is editor-managed plan data; purchase_executions.cloud_account_id is RESTRICT)") +} diff --git a/internal/database/postgres/testhelpers/lockwait.go b/internal/database/postgres/testhelpers/lockwait.go new file mode 100644 index 00000000..8917f574 --- /dev/null +++ b/internal/database/postgres/testhelpers/lockwait.go @@ -0,0 +1,104 @@ +package testhelpers + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/require" +) + +// LockDB is the slice of a pool the lock helpers need. The config store's +// dbConn and a *pgxpool.Pool both satisfy it. +type LockDB interface { + Begin(ctx context.Context) (pgx.Tx, error) + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +} + +// Footprint stands in for the editor command's lock footprint on one plan so +// the race tests of legacy writers run before the command exists: the plan row +// FOR UPDATE, then its executions ordered FOR UPDATE, then its accounts FOR +// SHARE. It does NOT take the command's first two lock levels, the identity +// advisory lock and the sorted provider advisory locks (LockPlanProvidersTx). +// The embedded Tx is the command's transaction: Exec applies the command's +// writes (SetManaged, inserted roots) and Commit/Rollback end it. Executions +// lists the rows the footprint locked, i.e. what the command captured: a +// writer that committed before the footprint must appear here. +// +// Swap contract for the ExecutePlanScheduleCommand race tests: replace +// HoldCommandFootprint with the real command paused after its capture, wrapped +// as a *Footprint (Tx = the command's transaction, Executions = its captured +// set). Then (1) the notifier claim queues first on the provider advisory lock, +// so its WaitForLockWait pattern changes from "%FOR UPDATE%" to the advisory +// wait (wait_event 'advisory'; match the pg_advisory_xact_lock statement); +// (2) the direct Exec/QueryRow calls of the retry, prune and account-delete +// tests become the command's own writes and reads; every other assertion stays. +type Footprint struct { + pgx.Tx + Executions []string +} + +// HoldCommandFootprint opens the footprint on planID. The transaction rolls +// back at test end unless the test ended it first. +func HoldCommandFootprint(ctx context.Context, t *testing.T, db LockDB, planID string) *Footprint { + t.Helper() + tx, err := db.Begin(ctx) + require.NoError(t, err) + t.Cleanup(func() { + if rbErr := tx.Rollback(ctx); rbErr != nil && !errors.Is(rbErr, pgx.ErrTxClosed) { + t.Logf("footprint rollback: %v", rbErr) + } + }) + f := &Footprint{Tx: tx} + _, err = tx.Exec(ctx, `SELECT 1 FROM purchase_plans WHERE id = $1 FOR UPDATE`, planID) + require.NoError(t, err) + rows, err := tx.Query(ctx, `SELECT execution_id::text FROM purchase_executions WHERE plan_id = $1 ORDER BY execution_id FOR UPDATE`, planID) + require.NoError(t, err) + for rows.Next() { + var id string + require.NoError(t, rows.Scan(&id)) + f.Executions = append(f.Executions, id) + } + require.NoError(t, rows.Err()) + _, err = tx.Exec(ctx, `SELECT 1 FROM cloud_accounts WHERE id IN (SELECT account_id FROM plan_accounts WHERE plan_id = $1) ORDER BY id FOR SHARE`, planID) + require.NoError(t, err) + return f +} + +// SetManaged writes the editor marker inside the footprint's transaction. +func (f *Footprint) SetManaged(ctx context.Context, t *testing.T, planID string) { + t.Helper() + _, err := f.Exec(ctx, `UPDATE purchase_plans SET ramp_schedule = ramp_schedule || '{"dates_managed":true}' WHERE id = $1`, planID) + require.NoError(t, err) +} + +// WaitForLockWait blocks until some backend is waiting on a lock in a +// statement matching queryLike (a SQL LIKE pattern; "" matches any), which +// proves the writer under test is queued behind the footprint. +func WaitForLockWait(ctx context.Context, t *testing.T, db LockDB, queryLike string) { + t.Helper() + if queryLike == "" { + queryLike = "%" + } + require.Eventually(t, func() bool { + var n int + err := db.QueryRow(ctx, `SELECT count(*) FROM pg_stat_activity WHERE datname = current_database() + AND wait_event_type = 'Lock' AND query LIKE $1`, queryLike).Scan(&n) + return err == nil && n > 0 + }, 15*time.Second, 20*time.Millisecond, "writer must wait on the footprint lock") +} + +// Await returns the value a background writer sends on done, failing fast +// instead of hanging until the go test timeout when the writer is stuck. +func Await[T any](t *testing.T, done <-chan T) T { + t.Helper() + select { + case v := <-done: + return v + case <-time.After(30 * time.Second): + t.Fatal("writer did not finish") + panic("unreachable") + } +} diff --git a/internal/purchase/plan_step_managed_race_integration_test.go b/internal/purchase/plan_step_managed_race_integration_test.go new file mode 100644 index 00000000..4c6a128d --- /dev/null +++ b/internal/purchase/plan_step_managed_race_integration_test.go @@ -0,0 +1,60 @@ +//go:build integration + +package purchase + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/require" +) + +// W15: the notifier is a new-root inserter. It judges the marker on the plan +// row it locked, never on the copy the tick listed, so a plan the command +// converted in between gets no phantom root. + +type claimRace struct { + store *config.PostgresStore + pool *pgxpool.Pool + plan *config.PurchasePlan + sender *countingSender + m *Manager +} + +func newClaimRace(ctx context.Context, t *testing.T) *claimRace { + store, pool := rampStepStore(ctx, t) + attachGlobal(ctx, t, store) + plan := attachPlanInDB(ctx, t, store, "claim-race", time.Now().Add(24*time.Hour)) + sender := newCountingSender(ctx) + res := &fakeResolver{recs: []config.RecommendationRecord{attachRec("rec-a", 2)}} + return &claimRace{store, pool, plan, sender, &Manager{config: store, email: sender, resolver: res}} +} + +func TestClaimSkipsPlanTheCommandConvertedWhileItWaited(t *testing.T) { + ctx := context.Background() + r := newClaimRace(ctx, t) + command := testhelpers.HoldCommandFootprint(ctx, t, r.pool, r.plan.ID) + done := make(chan bool, 1) + go func() { done <- r.m.sendPlanNotification(ctx, r.plan) }() // r.plan is the stale, unmanaged copy + testhelpers.WaitForLockWait(ctx, t, r.pool, "%FOR UPDATE%") + command.SetManaged(ctx, t, r.plan.ID) + require.NoError(t, command.Commit(ctx)) + + require.False(t, testhelpers.Await(t, done)) + require.Empty(t, stepRowsInDB(ctx, t, r.store, r.plan.ID), "no phantom root on the converted plan") + require.Zero(t, r.sender.sends) +} + +func TestCommandCaptureSeesRootTheClaimCommittedFirst(t *testing.T) { + ctx := context.Background() + r := newClaimRace(ctx, t) + require.True(t, r.m.sendPlanNotification(ctx, r.plan)) + rows := stepRowsInDB(ctx, t, r.store, r.plan.ID) + require.Len(t, rows, 1) + + require.Equal(t, []string{rows[0].ExecutionID}, testhelpers.HoldCommandFootprint(ctx, t, r.pool, r.plan.ID).Executions) +} diff --git a/known_issues/35_legacy_guard_followups.md b/known_issues/35_legacy_guard_followups.md new file mode 100644 index 00000000..4b800b0f --- /dev/null +++ b/known_issues/35_legacy_guard_followups.md @@ -0,0 +1,27 @@ +# Known Issues: follow-ups from the legacy-writer guards (#631) + +> **Audit status (2026-10-16):** `2 needs triage ยท 0 resolved` + +## LOW: cleanup lambda deletes by expires_at with no status predicate + +**Files:** `cmd/cleanup-lambda/main.go` (`deleteExpired`) + +**Description:** Unlike `CleanupOldExecutions`, the lambda deletes any row whose `expires_at` lapsed, whatever its status. It cannot reach editor-managed state today: no production code assigns `PurchaseExecution.TTL`, so editor roots carry a NULL `expires_at`, and the `retain_ramp_ledger_execution` trigger (migration 000124) keeps every capacity-holding row of an unfinished ramp plan. + +**Why deferred:** Pre-existing and inert; the legacy-writer guards only prove the pruning recheck (`TestPruneRechecksRowAfterCommandCancelsIt`). + +**Issue:** #998 (follow-up of #631) + +**Status:** Needs triage. + +## MEDIUM: a failed-with-submission row blocks deleting a managed plan forever + +**Files:** `internal/config/store_postgres_managed_guard.go` (`requireManagedDeleteTx`), `internal/config/store_postgres_ramp_baseline.go` (`holdsCapacity`) + +**Description:** Deleting an editor-managed plan is refused while any execution holds capacity, including a failed row with a submission whose retry later completed, because the retry link is not consulted and failed rows are never pruned. Operators can only disable the plan. + +**Why deferred:** Fail closed is intended (unknown outcome); a reviewed settle-or-acknowledge operation is a separate feature, not a flag on delete. + +**Issue:** #999 (follow-up of #631) + +**Status:** Needs triage.