test(plans): prove legacy writers recheck plans an editor Save changed

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 (

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)
+}

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)")
+}

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")
+	}
+}

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)
+}

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.