diff --git a/docs/invariants.md b/docs/invariants.md index ffa9591..831520a 100644 --- a/docs/invariants.md +++ b/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) diff --git a/internal/cli/cli.go b/internal/cli/cli.go index 710a346..1b06c34 100644 --- a/internal/cli/cli.go +++ b/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. diff --git a/internal/cli/migrate.go b/internal/cli/migrate.go index 3e431c2..21648e6 100644 --- a/internal/cli/migrate.go +++ b/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: diff --git a/internal/cli/migrate_test.go b/internal/cli/migrate_test.go index bd0df77..6d7fac9 100644 --- a/internal/cli/migrate_test.go +++ b/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) }) diff --git a/pkg/executor/optimistic.go b/pkg/executor/optimistic.go index 2096a27..bb8255c 100644 --- a/pkg/executor/optimistic.go +++ b/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 { diff --git a/pkg/executor/optimistic_integration_test.go b/pkg/executor/optimistic_integration_test.go index e277b80..bce576a 100644 --- a/pkg/executor/optimistic_integration_test.go +++ b/pkg/executor/optimistic_integration_test.go @@ -54,19 +54,19 @@ func columnType(t *testing.T, pool *pgxpool.Pool, schema, table, column string) return typ } -func TestAttemptNativeCommitsInstantChange(t *testing.T) { +func TestExecuteNativeCommitsInstantChange(t *testing.T) { pool, schema := newPool(t) _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema)) require.NoError(t, err) pt := mustPreflight(t, pool, schema, "t") st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int NOT NULL DEFAULT 0", schema)) - require.NoError(t, executor.AttemptNative(t.Context(), pool, pt, st, budget)) + require.NoError(t, executor.ExecuteNative(t.Context(), pool, pt, st, budget, executor.DefaultRetryPolicy())) assert.Equal(t, "integer", columnType(t, pool, schema, "t", "age"), "the committed change must be visible") } -func TestAttemptNativeCancelsWhenLockBlocked(t *testing.T) { +func TestExecuteNativeCancelsWhenLockBlocked(t *testing.T) { pool, schema := newPool(t) _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema)) require.NoError(t, err) @@ -83,15 +83,17 @@ func TestAttemptNativeCancelsWhenLockBlocked(t *testing.T) { require.NoError(t, err) st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int", schema)) - err = executor.AttemptNative(t.Context(), pool, pt, st, budget) + err = executor.ExecuteNative(t.Context(), pool, pt, st, budget, executor.DefaultRetryPolicy()) var budgetErr *executor.BudgetError require.ErrorAs(t, err, &budgetErr) assert.Equal(t, executor.CauseLock, budgetErr.Cause) assert.Equal(t, budget.LockTimeout, budgetErr.Budget) + assert.Equal(t, executor.DefaultRetryPolicy().MaxAttempts, budgetErr.Attempts, + "an actually blocked DDL must exhaust the bounded retry policy") } -func TestAttemptNativeCancelsRewriteAndLeavesTableUnchanged(t *testing.T) { +func TestExecuteNativeCancelsRewriteAndLeavesTableUnchanged(t *testing.T) { pool, schema := newPool(t) _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v text)", schema)) require.NoError(t, err) @@ -105,7 +107,7 @@ func TestAttemptNativeCancelsRewriteAndLeavesTableUnchanged(t *testing.T) { // int -> bigint forces a full table rewrite under ACCESS EXCLUSIVE. st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) tight := executor.Budget{LockTimeout: budget.LockTimeout, StatementTimeout: 50 * time.Millisecond} - err = executor.AttemptNative(t.Context(), pool, pt, st, tight) + err = executor.ExecuteNative(t.Context(), pool, pt, st, tight, executor.DefaultRetryPolicy()) var budgetErr *executor.BudgetError require.ErrorAs(t, err, &budgetErr) @@ -119,7 +121,7 @@ func TestAttemptNativeCancelsRewriteAndLeavesTableUnchanged(t *testing.T) { assert.Equal(t, 300000, count) } -func TestAttemptNativeSurfacesOperationalErrors(t *testing.T) { +func TestExecuteNativeSurfacesOperationalErrors(t *testing.T) { pool, schema := newPool(t) _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema)) require.NoError(t, err) @@ -128,7 +130,7 @@ func TestAttemptNativeSurfacesOperationalErrors(t *testing.T) { // Dropping a column that does not exist is a plain SQL error, not a // budget overrun. st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t DROP COLUMN nope", schema)) - err = executor.AttemptNative(t.Context(), pool, pt, st, budget) + err = executor.ExecuteNative(t.Context(), pool, pt, st, budget, executor.DefaultRetryPolicy()) require.Error(t, err) var budgetErr *executor.BudgetError assert.NotErrorAs(t, err, &budgetErr) @@ -136,7 +138,7 @@ func TestAttemptNativeSurfacesOperationalErrors(t *testing.T) { // Sub-millisecond budgets are as unbounded as zero ones: they truncate to // PostgreSQL's 0ms, which disables the corresponding limit entirely. -func TestAttemptNativeRejectsUnboundedBudgets(t *testing.T) { +func TestExecuteNativeRejectsUnboundedBudgets(t *testing.T) { pool, schema := newPool(t) _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema)) require.NoError(t, err) @@ -151,7 +153,7 @@ func TestAttemptNativeRejectsUnboundedBudgets(t *testing.T) { } for name, b := range unbounded { t.Run(name, func(t *testing.T) { - require.Error(t, executor.AttemptNative(t.Context(), pool, pt, st, b)) + require.Error(t, executor.ExecuteNative(t.Context(), pool, pt, st, b, executor.DefaultRetryPolicy())) }) } @@ -165,7 +167,7 @@ func TestAttemptNativeRejectsUnboundedBudgets(t *testing.T) { }) _, err = blocker.Exec(t.Context(), fmt.Sprintf("LOCK TABLE %s.t IN ACCESS EXCLUSIVE MODE", schema)) require.NoError(t, err) - err = executor.AttemptNative(t.Context(), pool, pt, st, executor.Budget{LockTimeout: time.Millisecond, StatementTimeout: time.Second}) + err = executor.ExecuteNative(t.Context(), pool, pt, st, executor.Budget{LockTimeout: time.Millisecond, StatementTimeout: time.Second}, executor.DefaultRetryPolicy()) var budgetErr *executor.BudgetError require.ErrorAs(t, err, &budgetErr) assert.Equal(t, executor.CauseLock, budgetErr.Cause) @@ -173,7 +175,7 @@ func TestAttemptNativeRejectsUnboundedBudgets(t *testing.T) { // INV: ST-7 — a preflight proof for one table can never execute a statement // against another, and a statement without a table target never executes. -func TestAttemptNativeRefusesTargetMismatch(t *testing.T) { +func TestExecuteNativeRefusesTargetMismatch(t *testing.T) { pool, schema := newPool(t) for _, ddl := range []string{ fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema), @@ -186,7 +188,7 @@ func TestAttemptNativeRefusesTargetMismatch(t *testing.T) { t.Run("statement targets a different table", func(t *testing.T) { st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.victim ADD COLUMN a int", schema)) - err := executor.AttemptNative(t.Context(), pool, pt, st, budget) + err := executor.ExecuteNative(t.Context(), pool, pt, st, budget, executor.DefaultRetryPolicy()) require.ErrorIs(t, err, executor.ErrInvariantViolation) var n int @@ -200,13 +202,13 @@ func TestAttemptNativeRefusesTargetMismatch(t *testing.T) { // Fail-closed: the proof verified schema.t, the statement names a // bare t that search_path could resolve elsewhere. st := mustParse(t, "ALTER TABLE t ADD COLUMN a int") - err := executor.AttemptNative(t.Context(), pool, pt, st, budget) + err := executor.ExecuteNative(t.Context(), pool, pt, st, budget, executor.DefaultRetryPolicy()) require.ErrorIs(t, err, executor.ErrInvariantViolation) }) t.Run("statement without a table target", func(t *testing.T) { st := mustParse(t, "CREATE TABLE elsewhere (id int)") - err := executor.AttemptNative(t.Context(), pool, pt, st, budget) + err := executor.ExecuteNative(t.Context(), pool, pt, st, budget, executor.DefaultRetryPolicy()) require.ErrorIs(t, err, executor.ErrInvariantViolation) }) } diff --git a/pkg/executor/retry_internal_test.go b/pkg/executor/retry_internal_test.go new file mode 100644 index 0000000..f0a0761 --- /dev/null +++ b/pkg/executor/retry_internal_test.go @@ -0,0 +1,82 @@ +package executor + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestExecuteWithLockRetry(t *testing.T) { + policy := RetryPolicy{MaxAttempts: 4, InitialBackoff: 10 * time.Millisecond, MaxBackoff: 25 * time.Millisecond} + var sleeps []time.Duration + attempts := 0 + err := executeWithLockRetry(t.Context(), policy, func(context.Context) error { + attempts++ + if attempts < 4 { + return &BudgetError{Cause: CauseLock, Budget: time.Millisecond} + } + return nil + }, func(_ context.Context, delay time.Duration) error { + sleeps = append(sleeps, delay) + return nil + }) + require.NoError(t, err) + assert.Equal(t, 4, attempts) + assert.Equal(t, []time.Duration{10 * time.Millisecond, 20 * time.Millisecond, 25 * time.Millisecond}, sleeps) +} + +func TestExecuteWithLockRetryExhausted(t *testing.T) { + policy := RetryPolicy{MaxAttempts: 3, InitialBackoff: time.Millisecond, MaxBackoff: time.Millisecond} + err := executeWithLockRetry(t.Context(), policy, func(context.Context) error { + return &BudgetError{Cause: CauseLock, Budget: 5 * time.Millisecond} + }, func(context.Context, time.Duration) error { return nil }) + var budgetErr *BudgetError + require.ErrorAs(t, err, &budgetErr) + assert.Equal(t, 3, budgetErr.Attempts) + assert.Contains(t, err.Error(), "after 3 bounded attempts") +} + +func TestExecuteWithLockRetryDoesNotRetryOtherFailures(t *testing.T) { + tests := []struct { + name string + err error + }{ + {name: "statement timeout", err: &BudgetError{Cause: CauseStatement, Budget: time.Second}}, + {name: "operational error", err: &pgconn.PgError{Code: "42601"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + attempts := 0 + err := executeWithLockRetry(t.Context(), DefaultRetryPolicy(), func(context.Context) error { + attempts++ + return tt.err + }, func(context.Context, time.Duration) error { return errors.New("unexpected sleep") }) + require.ErrorIs(t, err, tt.err) + assert.Equal(t, 1, attempts) + }) + } +} + +func TestRetryPolicyRejectsUnboundedValues(t *testing.T) { + tests := []RetryPolicy{ + {MaxAttempts: 0, InitialBackoff: time.Millisecond, MaxBackoff: time.Second}, + {MaxAttempts: 1, InitialBackoff: -time.Millisecond, MaxBackoff: time.Second}, + {MaxAttempts: 1, InitialBackoff: time.Second, MaxBackoff: time.Millisecond}, + // Retries with no backoff would re-enter the lock queue back-to-back. + {MaxAttempts: 3, InitialBackoff: 0, MaxBackoff: 0}, + {MaxAttempts: 2, InitialBackoff: 0, MaxBackoff: time.Second}, + } + for _, policy := range tests { + require.Error(t, policy.validate()) + } +} + +// A single attempt never sleeps, so it needs no backoff to be bounded. +func TestRetryPolicyAcceptsSingleAttemptWithoutBackoff(t *testing.T) { + require.NoError(t, RetryPolicy{MaxAttempts: 1}.validate()) +} diff --git a/pkg/verdict/verdict.go b/pkg/verdict/verdict.go index 676093e..75306df 100644 --- a/pkg/verdict/verdict.go +++ b/pkg/verdict/verdict.go @@ -78,6 +78,10 @@ type Verdict struct { // Cause narrows a budget refusal to the budget that fired; empty // otherwise. Cause Cause `json:"cause,omitempty"` + // Attempts is how many bounded attempts ran before a lock-budget + // refusal, so automation can tell an exhausted bounded retry from a + // single cancelled attempt; zero for every other verdict. + Attempts int `json:"attempts,omitempty"` // Statement is the submitted SQL. Statement string `json:"statement"` // Table is the target table (schema-qualified when the statement was), @@ -114,6 +118,9 @@ func (v Verdict) String() string { fmt.Fprintf(&b, "\n table: %s", v.Table) } fmt.Fprintf(&b, "\n statement: %s", v.Statement) + if v.Attempts > 0 { + fmt.Fprintf(&b, "\n attempts: %d", v.Attempts) + } if v.Detail != "" { fmt.Fprintf(&b, "\n detail: %s", v.Detail) } diff --git a/pkg/verdict/verdict_test.go b/pkg/verdict/verdict_test.go index 73dacae..c4cd9fc 100644 --- a/pkg/verdict/verdict_test.go +++ b/pkg/verdict/verdict_test.go @@ -16,6 +16,7 @@ func TestJSONRoundTrip(t *testing.T) { Table: "t", Detail: "the optimistic attempt exceeded its statement budget", SaferIdiom: "ADD CONSTRAINT ... NOT VALID; VALIDATE CONSTRAINT", + Attempts: 3, } s, err := v.JSON() require.NoError(t, err) @@ -31,6 +32,7 @@ func TestJSONOmitsEmptyOptionalFields(t *testing.T) { assert.NotContains(t, s, "reason") assert.NotContains(t, s, "table") assert.NotContains(t, s, "safer_idiom") + assert.NotContains(t, s, "attempts") } // Reason and Cause values are the machine contract automation switches on: @@ -70,3 +72,16 @@ func TestStringRefusedIncludesReasonAndIdiom(t *testing.T) { assert.Contains(t, s, "refused (index-statement)") assert.Contains(t, s, "CREATE INDEX CONCURRENTLY") } + +func TestStringIncludesAttemptsWhenSet(t *testing.T) { + v := Verdict{ + Outcome: OutcomeRefused, + Reason: ReasonBudgetExceeded, + Statement: "ALTER TABLE t ADD COLUMN x int", + Attempts: 3, + } + assert.Contains(t, v.String(), "attempts: 3") + + v.Attempts = 0 + assert.NotContains(t, v.String(), "attempts") +}