diff --git a/cmd/apply/apply.go b/cmd/apply/apply.go
index a8e9e70cf..62859ce12 100644
--- a/cmd/apply/apply.go
+++ b/cmd/apply/apply.go
@@ -4,11 +4,14 @@ import (
"bufio"
"context"
"database/sql"
+ "errors"
"fmt"
"os"
"path/filepath"
"strings"
+ "time"
+ "github.com/jackc/pgx/v5/pgconn"
planCmd "github.com/pgplex/pgschema/cmd/plan"
"github.com/pgplex/pgschema/cmd/util"
"github.com/pgplex/pgschema/internal/fingerprint"
@@ -19,6 +22,13 @@ import (
"github.com/spf13/cobra"
)
+// lockNotAvailableSQLState is the PostgreSQL SQLSTATE raised when a statement
+// fails to acquire a lock before lock_timeout expires.
+const lockNotAvailableSQLState = "55P03"
+
+// maxLockRetryBackoff caps the exponential backoff between lock timeout retries.
+const maxLockRetryBackoff = 30 * time.Second
+
var (
applyHost string
applyPort int
@@ -31,6 +41,8 @@ var (
applyAutoApprove bool
applyNoColor bool
applyLockTimeout string
+ applyLockRetries int
+ applyLockRetryWait time.Duration
applyApplicationName string
// Plan database connection flags (optional - for using external database instead of embedded postgres)
@@ -72,6 +84,8 @@ func init() {
ApplyCmd.Flags().BoolVar(&applyAutoApprove, "auto-approve", false, "Apply changes without prompting for approval")
ApplyCmd.Flags().BoolVar(&applyNoColor, "no-color", false, "Disable colored output")
ApplyCmd.Flags().StringVar(&applyLockTimeout, "lock-timeout", "", "Maximum time to wait for database locks (e.g., 30s, 5m, 1h)")
+ ApplyCmd.Flags().IntVar(&applyLockRetries, "lock-timeout-retries", 0, "Number of times to retry a statement that fails with a lock timeout (requires --lock-timeout)")
+ ApplyCmd.Flags().DurationVar(&applyLockRetryWait, "lock-timeout-retry-wait", 1*time.Second, "Initial wait before retrying a lock timeout failure; doubles after each retry up to 30s")
ApplyCmd.Flags().StringVar(&applyApplicationName, "application-name", "pgschema", "Application name for database connection (visible in pg_stat_activity) (env: PGAPPNAME)")
// Plan database connection flags (optional - for using external database instead of embedded postgres when using --file)
@@ -103,6 +117,8 @@ type ApplyConfig struct {
NoColor bool
Quiet bool // Suppress plan display and progress messages (useful for tests)
LockTimeout string
+ LockRetries int // Number of retries on lock timeout (55P03) failures
+ LockRetryWait time.Duration // Initial backoff before the first retry; doubles thereafter
ApplicationName string
SSLMode string
// Plan database configuration (needed when GeneratePlan checks provider SSL mode)
@@ -118,6 +134,19 @@ type ApplyConfig struct {
// If config.File is provided, provider is used to generate the plan.
// The caller is responsible for managing the provider lifecycle (creation and cleanup).
func ApplyMigration(config *ApplyConfig, provider postgres.DesiredStateProvider) error {
+ // Validate lock retry settings here too, not just in RunApply: a direct
+ // caller of ApplyMigration could otherwise set LockRetries without a
+ // LockTimeout and block indefinitely instead of getting a retryable error.
+ if config.LockRetries < 0 {
+ return fmt.Errorf("lock timeout retries must be non-negative")
+ }
+ if config.LockRetries > 0 && config.LockTimeout == "" {
+ return fmt.Errorf("lock timeout retries require a lock timeout")
+ }
+ if config.LockRetries > 0 && config.LockRetryWait <= 0 {
+ return fmt.Errorf("lock timeout retry wait must be greater than zero")
+ }
+
var migrationPlan *plan.Plan
var err error
@@ -247,13 +276,15 @@ func ApplyMigration(config *ApplyConfig, provider postgres.DesiredStateProvider)
return nil
}
+ retry := lockRetryConfig{MaxRetries: config.LockRetries, Backoff: config.LockRetryWait}
+
// Execute by groups with wait directive support
for i, group := range migrationPlan.Groups {
if !config.Quiet {
fmt.Printf("\nExecuting group %d/%d...\n", i+1, len(migrationPlan.Groups))
}
- err = executeGroup(ctx, conn, group, i+1, config.Quiet)
+ err = executeGroup(ctx, conn, group, i+1, config.Quiet, retry)
if err != nil {
return err
}
@@ -316,6 +347,18 @@ func RunApply(cmd *cobra.Command, args []string) error {
return err
}
+ // Lock timeout retries only make sense alongside a lock timeout: without one,
+ // statements block indefinitely instead of failing with a retryable error.
+ if applyLockRetries < 0 {
+ return fmt.Errorf("--lock-timeout-retries must be non-negative")
+ }
+ if applyLockRetries > 0 && applyLockTimeout == "" {
+ return fmt.Errorf("--lock-timeout-retries requires --lock-timeout to be set")
+ }
+ if applyLockRetries > 0 && applyLockRetryWait <= 0 {
+ return fmt.Errorf("--lock-timeout-retry-wait must be greater than zero")
+ }
+
// Build configuration
config := &ApplyConfig{
Host: applyHost,
@@ -327,6 +370,8 @@ func RunApply(cmd *cobra.Command, args []string) error {
AutoApprove: applyAutoApprove,
NoColor: applyNoColor,
LockTimeout: applyLockTimeout,
+ LockRetries: applyLockRetries,
+ LockRetryWait: applyLockRetryWait,
ApplicationName: applyApplicationName,
SSLMode: finalSSLMode,
}
@@ -444,7 +489,7 @@ func validateSchemaFingerprint(migrationPlan *plan.Plan, host string, port int,
}
// executeGroup executes all steps in a group, handling directives separately from SQL statements
-func executeGroup(ctx context.Context, conn *sql.DB, group plan.ExecutionGroup, groupNum int, quiet bool) error {
+func executeGroup(ctx context.Context, conn *sql.DB, group plan.ExecutionGroup, groupNum int, quiet bool, retry lockRetryConfig) error {
// Check if this group has directives
hasDirectives := false
@@ -457,15 +502,15 @@ func executeGroup(ctx context.Context, conn *sql.DB, group plan.ExecutionGroup,
if !hasDirectives {
// No directives - concatenate all SQL and execute in implicit transaction
- return executeGroupConcatenated(ctx, conn, group, groupNum, quiet)
+ return executeGroupConcatenated(ctx, conn, group, groupNum, quiet, retry)
} else {
// Has directives - execute statements individually
- return executeGroupIndividually(ctx, conn, group, groupNum, quiet)
+ return executeGroupIndividually(ctx, conn, group, groupNum, quiet, retry)
}
}
// executeGroupConcatenated concatenates all SQL statements and executes them in an implicit transaction
-func executeGroupConcatenated(ctx context.Context, conn *sql.DB, group plan.ExecutionGroup, groupNum int, quiet bool) error {
+func executeGroupConcatenated(ctx context.Context, conn *sql.DB, group plan.ExecutionGroup, groupNum int, quiet bool, retry lockRetryConfig) error {
var sqlStatements []string
// Collect all SQL statements
@@ -480,8 +525,9 @@ func executeGroupConcatenated(ctx context.Context, conn *sql.DB, group plan.Exec
fmt.Printf(" Executing %d statements in implicit transaction\n", len(sqlStatements))
}
- // Execute all statements in a single call (implicit transaction)
- _, err := util.ExecContextWithLogging(ctx, conn, concatenatedSQL, fmt.Sprintf("execute %d statements in group %d", len(sqlStatements), groupNum))
+ // Execute all statements in a single call (implicit transaction). Since the
+ // batch is atomic, retrying the whole batch on a lock timeout is safe.
+ _, err := execWithLockRetry(ctx, conn, concatenatedSQL, fmt.Sprintf("execute %d statements in group %d", len(sqlStatements), groupNum), retry, quiet)
if err != nil {
return fmt.Errorf("failed to execute concatenated statements in group %d: %w", groupNum, err)
}
@@ -490,7 +536,7 @@ func executeGroupConcatenated(ctx context.Context, conn *sql.DB, group plan.Exec
}
// executeGroupIndividually executes statements individually without transactions
-func executeGroupIndividually(ctx context.Context, conn *sql.DB, group plan.ExecutionGroup, groupNum int, quiet bool) error {
+func executeGroupIndividually(ctx context.Context, conn *sql.DB, group plan.ExecutionGroup, groupNum int, quiet bool, retry lockRetryConfig) error {
for stepIdx, step := range group.Steps {
if step.Directive != nil {
// Handle directive execution
@@ -504,7 +550,7 @@ func executeGroupIndividually(ctx context.Context, conn *sql.DB, group plan.Exec
fmt.Printf(" Executing: %s\n", truncateSQL(step.SQL, 80))
}
- _, err := util.ExecContextWithLogging(ctx, conn, step.SQL, fmt.Sprintf("execute statement in group %d, step %d", groupNum, stepIdx+1))
+ _, err := execWithLockRetry(ctx, conn, step.SQL, fmt.Sprintf("execute statement in group %d, step %d", groupNum, stepIdx+1), retry, quiet)
if err != nil {
return fmt.Errorf("failed to execute statement in group %d, step %d: %w", groupNum, stepIdx+1, err)
}
@@ -513,6 +559,56 @@ func executeGroupIndividually(ctx context.Context, conn *sql.DB, group plan.Exec
return nil
}
+// lockRetryConfig controls retry behavior for statements that fail because they
+// could not acquire a lock before lock_timeout expired (SQLSTATE 55P03).
+type lockRetryConfig struct {
+ MaxRetries int // Number of additional attempts after the first failure
+ Backoff time.Duration // Initial wait before the first retry; doubles after each subsequent retry
+}
+
+// sqlExecer is satisfied by both *sql.DB and *sql.Conn, matching util.ExecContextWithLogging.
+type sqlExecer interface {
+ ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
+}
+
+// isLockTimeoutError reports whether err is a PostgreSQL error raised because a
+// statement failed to acquire a lock before lock_timeout expired.
+func isLockTimeoutError(err error) bool {
+ var pgErr *pgconn.PgError
+ if errors.As(err, &pgErr) {
+ return pgErr.Code == lockNotAvailableSQLState
+ }
+ return false
+}
+
+// execWithLockRetry executes sqlStmt, retrying with exponential backoff when the
+// failure is a lock timeout (SQLSTATE 55P03). Any other error is returned immediately.
+func execWithLockRetry(ctx context.Context, conn sqlExecer, sqlStmt, description string, retry lockRetryConfig, quiet bool) (sql.Result, error) {
+ backoff := retry.Backoff
+
+ for attempt := 0; ; attempt++ {
+ result, err := util.ExecContextWithLogging(ctx, conn, sqlStmt, description)
+ if err == nil || !isLockTimeoutError(err) || attempt >= retry.MaxRetries {
+ return result, err
+ }
+
+ if !quiet {
+ fmt.Printf(" Lock not available, retrying in %s (retry %d/%d)...\n", backoff, attempt+1, retry.MaxRetries)
+ }
+
+ select {
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case <-time.After(backoff):
+ }
+
+ backoff *= 2
+ if backoff > maxLockRetryBackoff {
+ backoff = maxLockRetryBackoff
+ }
+ }
+}
+
// truncateSQL truncates a SQL statement for display purposes
func truncateSQL(sql string, maxLen int) string {
// Remove extra whitespace and newlines
diff --git a/cmd/apply/apply_test.go b/cmd/apply/apply_test.go
index de43d5311..a4006de3c 100644
--- a/cmd/apply/apply_test.go
+++ b/cmd/apply/apply_test.go
@@ -101,6 +101,24 @@ func TestApplyCommand(t *testing.T) {
t.Errorf("Expected default lock-timeout to be empty, got '%s'", lockTimeoutFlag.DefValue)
}
+ // Test lock-timeout-retries flag
+ lockRetriesFlag := flags.Lookup("lock-timeout-retries")
+ if lockRetriesFlag == nil {
+ t.Error("Expected --lock-timeout-retries flag to be defined")
+ }
+ if lockRetriesFlag.DefValue != "0" {
+ t.Errorf("Expected default lock-timeout-retries to be '0', got '%s'", lockRetriesFlag.DefValue)
+ }
+
+ // Test lock-timeout-retry-wait flag
+ lockRetryWaitFlag := flags.Lookup("lock-timeout-retry-wait")
+ if lockRetryWaitFlag == nil {
+ t.Error("Expected --lock-timeout-retry-wait flag to be defined")
+ }
+ if lockRetryWaitFlag.DefValue != "1s" {
+ t.Errorf("Expected default lock-timeout-retry-wait to be '1s', got '%s'", lockRetryWaitFlag.DefValue)
+ }
+
// Test application-name flag
applicationNameFlag := flags.Lookup("application-name")
if applicationNameFlag == nil {
@@ -208,11 +226,17 @@ func TestApplyCommandFlagValidation(t *testing.T) {
origUser := applyUser
origFile := applyFile
origPlan := applyPlan
+ origLockTimeout := applyLockTimeout
+ origLockRetries := applyLockRetries
+ origLockRetryWait := applyLockRetryWait
defer func() {
applyDB = origDB
applyUser = origUser
applyFile = origFile
applyPlan = origPlan
+ applyLockTimeout = origLockTimeout
+ applyLockRetries = origLockRetries
+ applyLockRetryWait = origLockRetryWait
}()
t.Run("neither file nor plan specified", func(t *testing.T) {
@@ -231,6 +255,59 @@ func TestApplyCommandFlagValidation(t *testing.T) {
}
})
+ t.Run("lock timeout retries without lock timeout", func(t *testing.T) {
+ // Reset flags
+ applyDB = "testdb"
+ applyUser = "testuser"
+ applyFile = "schema.sql"
+ applyPlan = ""
+ applyLockTimeout = ""
+ applyLockRetries = 3
+
+ err := RunApply(ApplyCmd, []string{})
+ if err == nil {
+ t.Error("Expected error when --lock-timeout-retries is set without --lock-timeout")
+ }
+ if err != nil && err.Error() != "--lock-timeout-retries requires --lock-timeout to be set" {
+ t.Errorf("Expected specific error message, got: %v", err)
+ }
+ })
+
+ t.Run("negative lock timeout retries", func(t *testing.T) {
+ applyDB = "testdb"
+ applyUser = "testuser"
+ applyFile = "schema.sql"
+ applyPlan = ""
+ applyLockTimeout = "5s"
+ applyLockRetries = -1
+
+ err := RunApply(ApplyCmd, []string{})
+ if err == nil {
+ t.Error("Expected error when --lock-timeout-retries is negative")
+ }
+ if err != nil && err.Error() != "--lock-timeout-retries must be non-negative" {
+ t.Errorf("Expected specific error message, got: %v", err)
+ }
+ })
+
+ t.Run("non-positive lock timeout retry wait", func(t *testing.T) {
+ applyDB = "testdb"
+ applyUser = "testuser"
+ applyFile = "schema.sql"
+ applyPlan = ""
+ applyLockTimeout = "5s"
+ applyLockRetries = 3
+ applyLockRetryWait = 0
+
+ err := RunApply(ApplyCmd, []string{})
+ if err == nil {
+ t.Error("Expected error when --lock-timeout-retry-wait is non-positive")
+ }
+ if err != nil && err.Error() != "--lock-timeout-retry-wait must be greater than zero" {
+ t.Errorf("Expected specific error message, got: %v", err)
+ }
+ })
+
t.Run("both file and plan specified", func(t *testing.T) {
// Create a test command to test mutual exclusivity
testCmd := &cobra.Command{
diff --git a/cmd/apply/lock_retry_integration_test.go b/cmd/apply/lock_retry_integration_test.go
new file mode 100644
index 000000000..612feb5f0
--- /dev/null
+++ b/cmd/apply/lock_retry_integration_test.go
@@ -0,0 +1,169 @@
+package apply
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/pgplex/pgschema/internal/plan"
+ "github.com/pgplex/pgschema/testutil"
+)
+
+// lockTestPlan builds a minimal single-statement migration plan for exercising
+// lock timeout retries directly via ApplyMigration, bypassing plan generation.
+func lockTestPlan(sql string) *plan.Plan {
+ return &plan.Plan{
+ Groups: []plan.ExecutionGroup{
+ {Steps: []plan.Step{{SQL: sql, Type: "table", Operation: "alter", Path: "public.locktest"}}},
+ },
+ }
+}
+
+// TestApplyCommand_LockTimeoutRetrySucceeds verifies that a statement blocked by a
+// concurrent lock eventually succeeds once --lock-timeout-retries are exhausted-but-one:
+// the lock is released while retries are still available.
+func TestApplyCommand_LockTimeoutRetrySucceeds(t *testing.T) {
+ if testing.Short() {
+ t.Skip("Skipping integration test in short mode")
+ }
+
+ ctx := context.Background()
+ embeddedPG := testutil.SetupPostgres(t)
+ defer embeddedPG.Stop()
+ conn, host, port, dbname, user, password := testutil.ConnectToPostgres(t, embeddedPG)
+ defer conn.Close()
+
+ if _, err := conn.ExecContext(ctx, "CREATE TABLE locktest (id INT)"); err != nil {
+ t.Fatalf("failed to create test table: %v", err)
+ }
+
+ // Hold an ACCESS EXCLUSIVE lock on locktest from a separate session, long enough
+ // that the first couple of attempts time out, then release it so a later retry succeeds.
+ lockConn, err := conn.Conn(ctx)
+ if err != nil {
+ t.Fatalf("failed to get dedicated connection: %v", err)
+ }
+ defer lockConn.Close()
+
+ if _, err := lockConn.ExecContext(ctx, "BEGIN"); err != nil {
+ t.Fatalf("failed to begin locking transaction: %v", err)
+ }
+ if _, err := lockConn.ExecContext(ctx, "LOCK TABLE locktest IN ACCESS EXCLUSIVE MODE"); err != nil {
+ t.Fatalf("failed to lock table: %v", err)
+ }
+
+ const holdDuration = 1200 * time.Millisecond
+ released := make(chan struct{})
+ go func() {
+ defer close(released)
+ time.Sleep(holdDuration)
+ if _, err := lockConn.ExecContext(context.Background(), "COMMIT"); err != nil {
+ t.Errorf("failed to release lock: %v", err)
+ }
+ }()
+ defer func() { <-released }()
+
+ applyConfig := &ApplyConfig{
+ Host: host,
+ Port: port,
+ DB: dbname,
+ User: user,
+ Password: password,
+ Schema: "public",
+ Plan: lockTestPlan("ALTER TABLE locktest ADD COLUMN name TEXT;"),
+ AutoApprove: true,
+ Quiet: true,
+ LockTimeout: "100ms",
+ LockRetries: 6,
+ LockRetryWait: 250 * time.Millisecond,
+ }
+
+ start := time.Now()
+ err = ApplyMigration(applyConfig, nil)
+ if err != nil {
+ t.Fatalf("expected apply to eventually succeed after the lock was released, got error: %v", err)
+ }
+ t.Logf("apply succeeded after %v", time.Since(start))
+
+ var nameColumnExists bool
+ if err := conn.QueryRowContext(ctx, `
+ SELECT EXISTS (
+ SELECT 1 FROM information_schema.columns
+ WHERE table_name = 'locktest' AND column_name = 'name'
+ )
+ `).Scan(&nameColumnExists); err != nil {
+ t.Fatalf("failed to check column existence: %v", err)
+ }
+ if !nameColumnExists {
+ t.Fatal("expected 'name' column to exist after apply succeeded")
+ }
+}
+
+// TestApplyCommand_LockTimeoutRetryExhausted verifies that apply fails with the
+// underlying lock timeout error once all configured retries are exhausted.
+func TestApplyCommand_LockTimeoutRetryExhausted(t *testing.T) {
+ if testing.Short() {
+ t.Skip("Skipping integration test in short mode")
+ }
+
+ ctx := context.Background()
+ embeddedPG := testutil.SetupPostgres(t)
+ defer embeddedPG.Stop()
+ conn, host, port, dbname, user, password := testutil.ConnectToPostgres(t, embeddedPG)
+ defer conn.Close()
+
+ if _, err := conn.ExecContext(ctx, "CREATE TABLE locktest (id INT)"); err != nil {
+ t.Fatalf("failed to create test table: %v", err)
+ }
+
+ // Hold the lock for the entire test - long enough to outlast every retry attempt.
+ lockConn, err := conn.Conn(ctx)
+ if err != nil {
+ t.Fatalf("failed to get dedicated connection: %v", err)
+ }
+ defer lockConn.Close()
+
+ if _, err := lockConn.ExecContext(ctx, "BEGIN"); err != nil {
+ t.Fatalf("failed to begin locking transaction: %v", err)
+ }
+ if _, err := lockConn.ExecContext(ctx, "LOCK TABLE locktest IN ACCESS EXCLUSIVE MODE"); err != nil {
+ t.Fatalf("failed to lock table: %v", err)
+ }
+ defer lockConn.ExecContext(context.Background(), "ROLLBACK")
+
+ applyConfig := &ApplyConfig{
+ Host: host,
+ Port: port,
+ DB: dbname,
+ User: user,
+ Password: password,
+ Schema: "public",
+ Plan: lockTestPlan("ALTER TABLE locktest ADD COLUMN name TEXT;"),
+ AutoApprove: true,
+ Quiet: true,
+ LockTimeout: "100ms",
+ LockRetries: 2,
+ LockRetryWait: 50 * time.Millisecond,
+ }
+
+ err = ApplyMigration(applyConfig, nil)
+ if err == nil {
+ t.Fatal("expected apply to fail once retries are exhausted while lock is still held")
+ }
+ if !isLockTimeoutError(err) {
+ t.Errorf("expected the returned error to wrap a lock timeout (55P03), got: %v", err)
+ }
+
+ var nameColumnExists bool
+ if err := conn.QueryRowContext(ctx, `
+ SELECT EXISTS (
+ SELECT 1 FROM information_schema.columns
+ WHERE table_name = 'locktest' AND column_name = 'name'
+ )
+ `).Scan(&nameColumnExists); err != nil {
+ t.Fatalf("failed to check column existence: %v", err)
+ }
+ if nameColumnExists {
+ t.Fatal("expected 'name' column to NOT exist since apply should have failed")
+ }
+}
diff --git a/cmd/apply/lock_retry_test.go b/cmd/apply/lock_retry_test.go
new file mode 100644
index 000000000..cf61d8ca5
--- /dev/null
+++ b/cmd/apply/lock_retry_test.go
@@ -0,0 +1,174 @@
+package apply
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/jackc/pgx/v5/pgconn"
+)
+
+// fakeResult is a minimal sql.Result implementation for tests.
+type fakeResult struct{}
+
+func (fakeResult) LastInsertId() (int64, error) { return 0, nil }
+func (fakeResult) RowsAffected() (int64, error) { return 0, nil }
+
+// fakeExecer simulates a database connection whose first failN calls fail with
+// failErr before succeeding.
+type fakeExecer struct {
+ calls int
+ failN int
+ failErr error
+}
+
+func (f *fakeExecer) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
+ f.calls++
+ if f.calls <= f.failN {
+ return nil, f.failErr
+ }
+ return fakeResult{}, nil
+}
+
+func TestIsLockTimeoutError(t *testing.T) {
+ tests := []struct {
+ name string
+ err error
+ want bool
+ }{
+ {"nil error", nil, false},
+ {"lock timeout pgerror", &pgconn.PgError{Code: lockNotAvailableSQLState}, true},
+ {"wrapped lock timeout pgerror", fmt.Errorf("exec failed: %w", &pgconn.PgError{Code: lockNotAvailableSQLState}), true},
+ {"other pgerror", &pgconn.PgError{Code: "42601"}, false},
+ {"non pgerror", errors.New("some other error"), false},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ if got := isLockTimeoutError(tt.err); got != tt.want {
+ t.Errorf("isLockTimeoutError(%v) = %v, want %v", tt.err, got, tt.want)
+ }
+ })
+ }
+}
+
+func TestExecWithLockRetry_SucceedsAfterRetries(t *testing.T) {
+ fe := &fakeExecer{failN: 2, failErr: &pgconn.PgError{Code: lockNotAvailableSQLState}}
+ retry := lockRetryConfig{MaxRetries: 3, Backoff: time.Millisecond}
+
+ _, err := execWithLockRetry(context.Background(), fe, "SELECT 1", "test", retry, true)
+ if err != nil {
+ t.Fatalf("expected success, got error: %v", err)
+ }
+ if fe.calls != 3 {
+ t.Errorf("expected 3 calls (2 failures + 1 success), got %d", fe.calls)
+ }
+}
+
+func TestExecWithLockRetry_ExhaustsRetries(t *testing.T) {
+ pgErr := &pgconn.PgError{Code: lockNotAvailableSQLState}
+ fe := &fakeExecer{failN: 100, failErr: pgErr}
+ retry := lockRetryConfig{MaxRetries: 2, Backoff: time.Millisecond}
+
+ _, err := execWithLockRetry(context.Background(), fe, "SELECT 1", "test", retry, true)
+ if err == nil {
+ t.Fatal("expected error after exhausting retries, got nil")
+ }
+ if !errors.Is(err, error(pgErr)) && !isLockTimeoutError(err) {
+ t.Errorf("expected the underlying lock timeout error to be returned, got: %v", err)
+ }
+ if fe.calls != 3 {
+ t.Errorf("expected 3 calls (1 initial + 2 retries), got %d", fe.calls)
+ }
+}
+
+func TestExecWithLockRetry_NonRetryableErrorFailsImmediately(t *testing.T) {
+ otherErr := &pgconn.PgError{Code: "42601"} // syntax_error
+ fe := &fakeExecer{failN: 100, failErr: otherErr}
+ retry := lockRetryConfig{MaxRetries: 5, Backoff: time.Millisecond}
+
+ _, err := execWithLockRetry(context.Background(), fe, "SELECT 1", "test", retry, true)
+ if err == nil {
+ t.Fatal("expected error, got nil")
+ }
+ if fe.calls != 1 {
+ t.Errorf("expected 1 call since non-retryable errors should not retry, got %d", fe.calls)
+ }
+}
+
+func TestExecWithLockRetry_NoRetriesConfigured(t *testing.T) {
+ pgErr := &pgconn.PgError{Code: lockNotAvailableSQLState}
+ fe := &fakeExecer{failN: 1, failErr: pgErr}
+ retry := lockRetryConfig{MaxRetries: 0, Backoff: time.Millisecond}
+
+ _, err := execWithLockRetry(context.Background(), fe, "SELECT 1", "test", retry, true)
+ if err == nil {
+ t.Fatal("expected error since MaxRetries is 0, got nil")
+ }
+ if fe.calls != 1 {
+ t.Errorf("expected exactly 1 call with MaxRetries=0, got %d", fe.calls)
+ }
+}
+
+func TestExecWithLockRetry_ContextCancelledDuringBackoff(t *testing.T) {
+ pgErr := &pgconn.PgError{Code: lockNotAvailableSQLState}
+ fe := &fakeExecer{failN: 100, failErr: pgErr}
+ retry := lockRetryConfig{MaxRetries: 5, Backoff: 50 * time.Millisecond}
+
+ ctx, cancel := context.WithCancel(context.Background())
+ go func() {
+ time.Sleep(10 * time.Millisecond)
+ cancel()
+ }()
+
+ _, err := execWithLockRetry(ctx, fe, "SELECT 1", "test", retry, true)
+ if !errors.Is(err, context.Canceled) {
+ t.Errorf("expected context.Canceled, got: %v", err)
+ }
+ if fe.calls != 1 {
+ t.Errorf("expected exactly 1 call before cancellation, got %d", fe.calls)
+ }
+}
+
+// TestApplyMigration_ValidatesLockRetryConfig verifies that ApplyMigration itself
+// rejects inconsistent lock retry settings, not just the RunApply CLI entry point.
+// This matters for callers that build an ApplyConfig directly and skip RunApply's
+// flag validation entirely.
+func TestApplyMigration_ValidatesLockRetryConfig(t *testing.T) {
+ tests := []struct {
+ name string
+ config *ApplyConfig
+ wantErr string
+ }{
+ {
+ name: "negative retries",
+ config: &ApplyConfig{LockRetries: -1},
+ wantErr: "lock timeout retries must be non-negative",
+ },
+ {
+ name: "retries without lock timeout",
+ config: &ApplyConfig{LockRetries: 3, LockRetryWait: time.Second},
+ wantErr: "lock timeout retries require a lock timeout",
+ },
+ {
+ name: "retries with non-positive retry wait",
+ config: &ApplyConfig{LockRetries: 3, LockTimeout: "5s", LockRetryWait: 0},
+ wantErr: "lock timeout retry wait must be greater than zero",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ err := ApplyMigration(tt.config, nil)
+ if err == nil {
+ t.Fatal("expected validation error, got nil")
+ }
+ if err.Error() != tt.wantErr {
+ t.Errorf("expected error %q, got %q", tt.wantErr, err.Error())
+ }
+ })
+ }
+}
diff --git a/docs/cli/apply.mdx b/docs/cli/apply.mdx
index ef46f54ab..a93fafd8f 100644
--- a/docs/cli/apply.mdx
+++ b/docs/cli/apply.mdx
@@ -165,6 +165,18 @@ When using File Mode (`--file`), the apply command generates a plan internally u
See [PostgreSQL lock_timeout documentation](https://www.postgresql.org/docs/current/runtime-config-client.html#GUC-LOCK-TIMEOUT).
+
+ Number of times to retry a statement that fails because it could not acquire a lock before `--lock-timeout` expired (PostgreSQL error 55P03)
+
+ Requires `--lock-timeout` to be set. Useful when applying migrations to busy tables: instead of failing on the first lock timeout, pgschema retries with exponential backoff until the lock becomes available or the retries are exhausted. See [Zero-downtime Postgres schema migrations: lock timeout and retries](https://postgres.ai/blog/20210923-zero-downtime-postgres-schema-migrations-lock-timeout-and-retries) for background on this technique.
+
+
+
+ Initial wait before retrying a lock timeout failure; doubles after each subsequent retry, capped at 30s
+
+ Only takes effect when `--lock-timeout-retries` is greater than 0.
+
+
Application name for database connection (visible in pg_stat_activity) (env: PGAPPNAME)
@@ -266,6 +278,21 @@ pgschema apply \
--lock-timeout "30s"
```
+### With Lock Timeout Retries
+
+```bash
+# Retry statements that fail to acquire a lock, instead of failing immediately.
+# Useful for busy tables where a lock is only held briefly.
+pgschema apply \
+ --host localhost \
+ --db myapp \
+ --user postgres \
+ --file schema.sql \
+ --lock-timeout "2s" \
+ --lock-timeout-retries 5 \
+ --lock-timeout-retry-wait "1s"
+```
+
### Custom Application Name
```bash