feat(executor): lock_timeout + bounded retry for native DDL

docs/invariants.md

@@ -238,7 +238,7 @@ which enforces exactly one statement through the real grammar — and refuses, b
 executes, any statement whose target table does not match the preflight proof it was handed.
 A proof for one table can never smuggle SQL against another, and a multi-statement string can
 never reach the database through the executor (pgx's simple protocol would happily run all of
-it). *Enforced:* `pkg/executor` (`AttemptNative`), `pkg/statement` (proof construction).
+it). *Enforced:* `pkg/executor` (`ExecuteNative`), `pkg/statement` (proof construction).
 *Source:* adversarial review of the optimistic front door.
 
 ## Refusals and preflight (RF)

internal/cli/cli.go

@@ -80,10 +80,13 @@ func (f DBFlags) diag() *slog.Logger {
 type MigrateCmd struct {
 	DBFlags `embed:""`
 
-	Alter        string   `help:"Imperative ALTER statement to run." name:"alter" required:""`
-	MaxTableSize byteSize `help:"Size threshold above which the optimistic attempt is skipped, measured as the table's full on-disk footprint: heap, indexes, and TOAST, all partitions (binary units: B, KiB, MiB, GiB, TiB)." default:"1GiB"`
-	DryRun       bool     `help:"Classify and route the statement, print the plan, and execute nothing."`
-	JSON         bool     `help:"Emit the verdict (or dry-run plan) as JSON."`
+	Alter          string        `help:"Imperative ALTER statement to run." name:"alter" required:""`
+	MaxTableSize   byteSize      `help:"Size threshold above which the optimistic attempt is skipped, measured as the table's full on-disk footprint: heap, indexes, and TOAST, all partitions (binary units: B, KiB, MiB, GiB, TiB)." default:"1GiB"`
+	DryRun         bool          `help:"Classify and route the statement, print the plan, and execute nothing."`
+	JSON           bool          `help:"Emit the verdict (or dry-run plan) as JSON."`
+	LockAttempts   int           `help:"Maximum bounded attempts when native DDL exceeds lock_timeout; 1 disables retry." default:"3"`
+	LockBackoff    time.Duration `help:"Initial exponential backoff between lock-timeout attempts." default:"100ms"`
+	LockBackoffMax time.Duration `help:"Maximum exponential backoff between lock-timeout attempts." default:"1s"`
 }
 
 // Run implements the migrate subcommand.

internal/cli/migrate.go

@@ -51,13 +51,14 @@ func (c *MigrateCmd) run(ctx context.Context, out io.Writer) error {
 		"table", qualified(st), "total_bytes", pt.TotalBytes(), "limit_bytes", int64(c.MaxTableSize))
 
 	budget := executor.Budget{LockTimeout: c.LockTimeout, StatementTimeout: c.StatementTimeout}
+	retry := c.retryPolicy()
 	start := time.Now()
-	err = executor.AttemptNative(ctx, pool, pt, st, budget)
+	err = executor.ExecuteNative(ctx, pool, pt, st, budget, retry)
 	elapsed := time.Since(start)
 	var budgetErr *executor.BudgetError
 	if errors.As(err, &budgetErr) {
 		logger.Debug("optimistic attempt cancelled",
-			"cause", budgetErr.Cause, "budget", budgetErr.Budget, "elapsed", elapsed)
+			"cause", budgetErr.Cause, "budget", budgetErr.Budget, "attempts", budgetErr.Attempts, "elapsed", elapsed)
 		return c.emit(out, budgetVerdict(st, budgetErr))
 	}
 	if err != nil {
@@ -73,6 +74,16 @@ func (c *MigrateCmd) run(ctx context.Context, out io.Writer) error {
 	})
 }
 
+func (c *MigrateCmd) retryPolicy() executor.RetryPolicy {
+	// Programmatic callers do not pass through Kong's default population.
+	// Preserve the safe defaults for a zero-valued command while rejecting
+	// partially configured or invalid policies in the executor.
+	if c.LockAttempts == 0 && c.LockBackoff == 0 && c.LockBackoffMax == 0 {
+		return executor.DefaultRetryPolicy()
+	}
+	return executor.RetryPolicy{MaxAttempts: c.LockAttempts, InitialBackoff: c.LockBackoff, MaxBackoff: c.LockBackoffMax}
+}
+
 // emit prints the verdict in the selected format and returns ErrRefused for
 // refusals so the exit code distinguishes them from operational errors.
 func (c *MigrateCmd) emit(out io.Writer, v verdict.Verdict) error {
@@ -161,6 +172,13 @@ func budgetVerdict(st statement.Statement, budgetErr *executor.BudgetError) verd
 	switch budgetErr.Cause {
 	case executor.CauseLock:
 		v.Cause = verdict.CauseLockBudget
+		v.Attempts = budgetErr.Attempts
+		if budgetErr.Attempts > 1 {
+			v.Detail = fmt.Sprintf("the lock was not granted within the %s lock budget on any of %d bounded "+
+				"attempts: the table is too contended for a blind attempt; nothing was executed",
+				budgetErr.Budget, budgetErr.Attempts)
+			break
+		}
 		v.Detail = fmt.Sprintf("the lock was not granted within the %s lock budget: the table is too "+
 			"contended for a blind attempt; nothing was executed", budgetErr.Budget)
 	case executor.CauseStatement:

internal/cli/migrate_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/alecthomas/kong"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -12,15 +13,59 @@ import (
 	"github.com/block/pg-sprite/pkg/verdict"
 )
 
+// parseMigrate runs args through the real command grammar so these tests
+// exercise the same flag-to-field path production does.
+func parseMigrate(t *testing.T, args ...string) *MigrateCmd {
+	t.Helper()
+	c := New()
+	k, err := kong.New(c, kong.Vars{"version": "test"})
+	require.NoError(t, err)
+	_, err = k.Parse(append([]string{
+		"migrate",
+		"--url", "postgres://user@localhost:5432/app",
+		"--alter", "ALTER TABLE t ADD COLUMN c int",
+	}, args...))
+	require.NoError(t, err)
+	return &c.Migrate
+}
+
+func TestRetryFlagsWireIntoRetryPolicy(t *testing.T) {
+	c := parseMigrate(t,
+		"--lock-attempts", "5",
+		"--lock-backoff", "250ms",
+		"--lock-backoff-max", "2s",
+	)
+	// Full-struct equality: swapping the backoff fields, inverting the
+	// zero-value fallback, or dropping the passthrough must all fail here.
+	assert.Equal(t, executor.RetryPolicy{
+		MaxAttempts:    5,
+		InitialBackoff: 250 * time.Millisecond,
+		MaxBackoff:     2 * time.Second,
+	}, c.retryPolicy())
+}
+
+func TestRetryPolicyDefaults(t *testing.T) {
+	t.Run("kong defaults match the executor defaults", func(t *testing.T) {
+		c := parseMigrate(t)
+		assert.Equal(t, executor.DefaultRetryPolicy(), c.retryPolicy())
+	})
+
+	t.Run("zero-valued command falls back to the executor defaults", func(t *testing.T) {
+		var c MigrateCmd
+		assert.Equal(t, executor.DefaultRetryPolicy(), c.retryPolicy())
+	})
+}
+
 func TestBudgetVerdict(t *testing.T) {
 	st, err := statement.ParseOne("ALTER TABLE billing.invoices ALTER COLUMN id TYPE bigint")
 	require.NoError(t, err)
 
 	t.Run("lock budget", func(t *testing.T) {
-		v := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseLock, Budget: 3 * time.Second})
+		v := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseLock, Budget: 3 * time.Second, Attempts: 3})
 		assert.Equal(t, verdict.OutcomeRefused, v.Outcome)
 		assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason)
 		assert.Equal(t, verdict.CauseLockBudget, v.Cause)
+		assert.Equal(t, 3, v.Attempts, "the exhausted attempt count must reach the verdict")
 		assert.Equal(t, "billing.invoices", v.Table)
 		assert.NotEmpty(t, v.Detail)
 	})

pkg/executor/optimistic.go

@@ -74,13 +74,56 @@ type BudgetError struct {
 	Cause BudgetCause
 	// Budget is the configured limit for that cause.
 	Budget time.Duration
+	// Attempts is the number of bounded transactions tried. It is greater
+	// than one when lock acquisition retries were exhausted.
+	Attempts int
 }
 
 // Error implements the error interface.
 func (e *BudgetError) Error() string {
+	if e.Attempts > 1 {
+		return fmt.Sprintf("execution exceeded its %s (%s) after %d bounded attempts", e.Cause, e.Budget, e.Attempts)
+	}
 	return fmt.Sprintf("execution exceeded its %s (%s) and was cancelled", e.Cause, e.Budget)
 }
 
+// RetryPolicy bounds retries after lock_timeout expires. Backoff doubles
+// after each failed attempt and is capped at MaxBackoff.
+type RetryPolicy struct {
+	MaxAttempts    int
+	InitialBackoff time.Duration
+	MaxBackoff     time.Duration
+}
+
+// DefaultRetryPolicy returns the safe native-DDL retry policy used by
+// callers that do not need to tune lock acquisition.
+func DefaultRetryPolicy() RetryPolicy {
+	return RetryPolicy{
+		MaxAttempts:    3,
+		InitialBackoff: 100 * time.Millisecond,
+		MaxBackoff:     time.Second,
+	}
+}
+
+func (p RetryPolicy) validate() error {
+	if p.MaxAttempts < 1 {
+		return fmt.Errorf("retry attempts must be at least 1, got %d", p.MaxAttempts)
+	}
+	if p.InitialBackoff < 0 {
+		return fmt.Errorf("initial retry backoff must not be negative, got %s", p.InitialBackoff)
+	}
+	// Zero backoff with retries enabled would re-enter the lock queue
+	// back-to-back, each occupancy blocking queued traffic for the full
+	// lock budget with no pause for the blocker to drain.
+	if p.MaxAttempts > 1 && p.InitialBackoff == 0 {
+		return fmt.Errorf("initial retry backoff must be positive when retries are enabled, got %d attempts with no backoff", p.MaxAttempts)
+	}
+	if p.MaxBackoff < p.InitialBackoff {
+		return fmt.Errorf("maximum retry backoff %s must be at least initial backoff %s", p.MaxBackoff, p.InitialBackoff)
+	}
+	return nil
+}
+
 // Budget bounds one optimistic attempt. Both limits must be at least
 // minBudget: an unbounded attempt is exactly the stall the front door exists
 // to prevent.
@@ -111,26 +154,39 @@ func (b Budget) validate() error {
 	return nil
 }
 
-// AttemptNative runs st once, directly, inside a transaction bounded by b.
-// The table must have passed preflight and the statement must target it —
-// both proofs make the unsafe call unrepresentable: a statement.Statement
-// can only come from ParseOne (exactly one statement, parsed by the real
+// ExecuteNative runs transactional native DDL under transaction-local
+// budgets and retries only lock_timeout failures, bounded by retry. The
+// table must have passed preflight and the statement must target it — both
+// proofs make the unsafe call unrepresentable: a statement.Statement can
+// only come from ParseOne (exactly one statement, parsed by the real
 // grammar), and a target mismatch is refused before anything executes, so a
-// proof for one table cannot smuggle SQL against another. On success the
-// change is committed: it was effectively instant. If a budget is exceeded
-// the statement is cancelled by the server, the transaction rolls back, and
-// a *BudgetError is returned. Any other failure is surfaced as an
-// operational error.
-func AttemptNative(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, st statement.Statement, b Budget) error {
+// proof for one table cannot smuggle SQL against another. Each attempt is a
+// new transaction, so neither an aborted transaction nor its settings can
+// leak through the pool. On success the change is committed: it was
+// effectively instant. If the lock budget is exhausted across all bounded
+// attempts, a *BudgetError carrying the attempt count is returned.
+// Statement timeouts and all other failures return immediately: repeating
+// work that exceeded its execution budget is not a lock-acquisition
+// strategy.
+func ExecuteNative(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, st statement.Statement, b Budget, retry RetryPolicy) error {
 	if err := b.validate(); err != nil {
 		return err
 	}
+	if err := retry.validate(); err != nil {
+		return err
+	}
 	// INV: ST-7 — the executor runs exactly the statement that was gated,
 	// and only against the table the preflight proof verified.
 	if st.Table() == "" || st.Schema() != pt.Schema() || st.Table() != pt.Table() {
 		return fmt.Errorf("%w: ST-7: statement targets %q but preflight verified %q",
 			ErrInvariantViolation, qualifiedName(st.Schema(), st.Table()), qualifiedName(pt.Schema(), pt.Table()))
 	}
+	return executeWithLockRetry(ctx, retry, func(ctx context.Context) error {
+		return executeNativeAttempt(ctx, pool, st, b)
+	}, sleepContext)
+}
+
+func executeNativeAttempt(ctx context.Context, pool *pgxpool.Pool, st statement.Statement, b Budget) error {
 	tx, err := pool.Begin(ctx)
 	if err != nil {
 		return fmt.Errorf("begin optimistic attempt: %w", err)
@@ -165,6 +221,54 @@ func AttemptNative(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflig
 	return nil
 }
 
+type sleepFunc func(context.Context, time.Duration) error
+
+func executeWithLockRetry(ctx context.Context, policy RetryPolicy, attempt func(context.Context) error, sleep sleepFunc) error {
+	for i := 1; i <= policy.MaxAttempts; i++ {
+		err := attempt(ctx)
+		if err == nil {
+			return nil
+		}
+		var budgetErr *BudgetError
+		if !errors.As(err, &budgetErr) || budgetErr.Cause != CauseLock {
+			return err
+		}
+		if i == policy.MaxAttempts {
+			budgetErr.Attempts = i
+			return budgetErr
+		}
+		if err := sleep(ctx, retryBackoff(policy, i)); err != nil {
+			return err
+		}
+	}
+	return fmt.Errorf("%w: LK-2: retry loop escaped its bounded attempts", ErrInvariantViolation)
+}
+
+func retryBackoff(policy RetryPolicy, failedAttempts int) time.Duration {
+	delay := policy.InitialBackoff
+	for i := 1; i < failedAttempts && delay < policy.MaxBackoff; i++ {
+		if delay > policy.MaxBackoff/2 {
+			return policy.MaxBackoff
+		}
+		delay *= 2
+	}
+	if delay > policy.MaxBackoff {
+		return policy.MaxBackoff
+	}
+	return delay
+}
+
+func sleepContext(ctx context.Context, delay time.Duration) error {
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
 // qualifiedName renders schema.table for error messages, omitting the dot
 // when the name is unqualified.
 func qualifiedName(schema, table string) string {