Add plain postgres ready and build connectivity waits into schema commands

plain-dev/plain/dev/core.py

@@ -74,6 +74,9 @@ def setup(
             "PLAIN_SERVER_ACCESS_LOG_FIELDS": '["method", "url", "status", "duration_ms", "size"]',
             "FORCE_COLOR": "1",
             "PYTHONWARNINGS": "default::DeprecationWarning,default::PendingDeprecationWarning",
+            # Dev databases can take a while on first run (Docker image pull);
+            # give schema commands more patience than the production default.
+            "PLAIN_POSTGRES_WAIT_TIMEOUT": "300",
             **os.environ,
         }
 
@@ -156,18 +159,14 @@ def run(self, *, reinstall_ssl: bool = False) -> int:
         self.run_preflight()
 
         if find_spec("plain.postgres"):
-            print_event("Waiting for database...", newline=False)
-            subprocess.run(
-                [sys.executable, "-m", "plain", "postgres", "wait"],
-                env=self.plain_env,
-                check=True,
-            )
-
-            # Backup before syncing if sync would make changes
+            # Backup before syncing if sync would make changes. sync --check
+            # waits for the database itself (POSTGRES_WAIT_TIMEOUT); capture
+            # only stdout so its stderr wait-progress lines stream live.
+            print_event("Waiting for database...")
             check_result = subprocess.run(
                 [sys.executable, "-m", "plain", "postgres", "sync", "--check"],
                 env=self.plain_env,
-                capture_output=True,
+                stdout=subprocess.PIPE,
             )
 
             if check_result.returncode != 0:
@@ -184,11 +183,15 @@ def run(self, *, reinstall_ssl: bool = False) -> int:
                     click.secho("skipped", dim=True)
 
             print_event("Syncing database...")
-            subprocess.run(
+            sync_result = subprocess.run(
                 [sys.executable, "-m", "plain", "postgres", "sync"],
-                env=self.plain_env,
-                check=True,
+                # The --check above already waited for the database — fail
+                # fast here instead of waiting a second full window.
+                env={**self.plain_env, "PLAIN_POSTGRES_WAIT_TIMEOUT": "0"},
             )
+            if sync_result.returncode != 0:
+                click.secho("Database sync failed — see the output above.", fg="red")
+                return 1
 
         print_event("Starting app...")
 

plain-postgres/plain/postgres/README.md

@@ -12,6 +12,7 @@
     - [Structural migrations](#structural-migrations)
     - [Data migrations](#data-migrations)
     - [Convergence](#convergence)
+    - [Readiness](#readiness)
 - [Fields](#fields)
 - [Relationships](#relationships)
 - [Constraints](#constraints)
@@ -556,6 +557,8 @@ plain postgres sync
 
 In development (`DEBUG=True`), sync auto-generates migrations before applying them. In production, it only applies existing migrations and converges.
 
+Sync also waits for the database to accept connections before starting (up to `POSTGRES_WAIT_TIMEOUT` seconds, default 60), so a release phase or dev environment where the database is still coming up doesn't need a separate wait step in front of it. Configuration errors — bad credentials, a broken `POSTGRES_URL` — fail immediately instead of retrying. `migrations apply` and `converge` do the same.
+
 | Command                        | Purpose                                                               |
 | ------------------------------ | --------------------------------------------------------------------- |
 | `plain postgres sync`          | Create + apply migrations + converge (the one command for everything) |
@@ -775,6 +778,57 @@ Environment overrides: every setting accepts `PLAIN_POSTGRES_*` env vars, so you
 PLAIN_POSTGRES_MIGRATION_STATEMENT_TIMEOUT=30s plain migrations apply
 ```
 
+### Schema lock
+
+Schema-changing commands — `plain postgres sync`, `plain migrations apply`, `plain postgres converge`, and `plain postgres drop-unknown-tables` — serialize on a single session-level advisory lock, so two deploy processes running at once (a retried migrate job, overlapping release phases) can't interleave schema changes. You don't have to do anything to get this.
+
+A second process warns and waits, retrying until the holder finishes:
+
+```python
+# app/settings.py — defaults shown (waits up to an hour total)
+POSTGRES_SCHEMA_LOCK_RETRY_INTERVAL = 5.0
+POSTGRES_SCHEMA_LOCK_MAX_RETRIES = 720
+```
+
+The wait is generous by default because a legitimate holder can be mid index build. If the budget runs out, the command fails with the holder's `pid` so you can see what's blocking. A crashed holder is not a problem — the lock releases automatically when its database session closes.
+
+The lock is held on its own connection, separate from the one running DDL, so non-transactional operations (`CREATE INDEX CONCURRENTLY`, `VALIDATE CONSTRAINT`) work normally while it's held. Session-level locks don't survive transaction-mode poolers like pgbouncer — if your `POSTGRES_URL` points at one, set [`POSTGRES_MANAGEMENT_URL`](#bypassing-a-connection-pooler-for-management-operations) to a direct connection.
+
+The lock connection sits idle while your DDL runs, so it enables TCP keepalives to survive NAT and load-balancer idle timeouts. A server-side `idle_session_timeout` would still kill it (releasing the lock mid-run) — don't set one for the role that runs migrations. `sync` re-verifies the lock between its migrate and converge phases and stops with a clear error if the session died.
+
+To see the lock live: `SELECT * FROM pg_locks WHERE locktype = 'advisory' AND objid = 1047265496`.
+
+### Readiness
+
+`plain postgres ready` checks that the database is ready for this code to serve: the database is reachable, all migrations are applied, and every model's table and columns exist. It's built for serving entrypoints — gate `plain server` or `plain jobs worker` on it so a process whose image got ahead of the database (a skipped migrate job, a restored backup, a rollback) fails with a clear diagnosis instead of serving mystery errors.
+
+```bash
+# In a server entrypoint or Kubernetes initContainer
+plain postgres ready --wait
+plain server
+```
+
+The exit code tells the caller what kind of problem it is:
+
+| Exit | Meaning                                                                                                                                                             |
+| ---- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `0`  | Ready to serve.                                                                                                                                                     |
+| `1`  | Not ready, but retryable — migrations pending, schema missing objects, or the database is unreachable. A scheduled migrate (or the server coming back) resolves it. |
+| `2`  | Configuration error a human must fix — bad credentials, database doesn't exist. Retrying will never fix it.                                                         |
+
+`--wait` polls until ready, so it standardizes the entrypoint poll-loop that otherwise gets hand-rolled around `migrate --check`-style commands. Configuration errors still exit `2` immediately, and `--timeout <seconds>` bounds the wait (an in-flight connection attempt finishes first, so the exit can run up to ~10 seconds past the deadline against an unresponsive host). Use `--json` (without `--wait`) for machine-readable output — the payload includes the exit code alongside the classification.
+
+A few behaviors are deliberate:
+
+- **Data migrations don't gate.** A pending migration whose operations are all `RunPython` warns instead of blocking — a long backfill is legitimately pending for hours while new processes serve fine.
+- **`DEBUG` warns instead of gating.** Mid-development you routinely have a new model with no migration yet; the dev server should tell you, not refuse to boot. This is the command's exit-code policy — the JSON `status` and the Python API below always report the true classification.
+- **Existence only.** The schema check verifies tables and columns exist — the things the ORM can't run without. Index, constraint, and type drift is [convergence](#convergence)'s job and never blocks serving.
+- **Unknown connection failures are retryable.** Only clearly-permanent errors (bad password, missing database, a broken `POSTGRES_URL`) exit `2` — a database restarting mid-deploy must never read as a configuration problem.
+
+The check runs against `POSTGRES_URL` — the connection the app serves through. Release phases don't need a readiness gate in front of them: the schema commands (`sync`, `migrations apply`, `converge`) wait for the database themselves, retrying connection failures for up to `POSTGRES_WAIT_TIMEOUT` seconds (default 60) while configuration errors fail immediately with the same classification `ready` uses.
+
+In Python, the same check is [`check_database_ready()`](./readiness.py#check_database_ready), which returns a classified [`ReadinessResult`](./readiness.py#ReadinessResult).
+
 ## Fields
 
 You can use many field types for different data:
@@ -1450,6 +1504,7 @@ The connection is configured with a single URL (`POSTGRES_URL`). `DATABASE_URL`
 | `POSTGRES_MIGRATION_STATEMENT_TIMEOUT`   | `str`         | `"3s"`                  | `PLAIN_POSTGRES_MIGRATION_STATEMENT_TIMEOUT`   |
 | `POSTGRES_CONVERGENCE_LOCK_TIMEOUT`      | `str`         | `"3s"`                  | `PLAIN_POSTGRES_CONVERGENCE_LOCK_TIMEOUT`      |
 | `POSTGRES_CONVERGENCE_STATEMENT_TIMEOUT` | `str`         | `"3s"`                  | `PLAIN_POSTGRES_CONVERGENCE_STATEMENT_TIMEOUT` |
+| `POSTGRES_WAIT_TIMEOUT`                  | `float`       | `60.0`                  | `PLAIN_POSTGRES_WAIT_TIMEOUT`                  |
 
 See [`default_settings.py`](./default_settings.py) for more details.
 

plain-postgres/plain/postgres/cli/converge.py

@@ -5,7 +5,11 @@
 import click
 
 from ..convergence import execute_plan, plan_convergence
-from .decorators import database_management_command
+from .decorators import (
+    cli_schema_lock,
+    cli_wait_for_database,
+    database_management_command,
+)
 
 
 @click.command()
@@ -28,6 +32,8 @@ def converge(yes: bool) -> None:
     Each fix is applied and committed independently so partial
     failures don't block subsequent fixes.
     """
+    cli_wait_for_database()
+
     plan = plan_convergence()
     items = plan.executable()
     success = True
@@ -48,7 +54,27 @@ def converge(yes: bool) -> None:
 
         click.echo()
 
-        result = execute_plan(items)
+        with cli_schema_lock():
+            # Re-plan under the lock — another converge or sync may have
+            # already applied (or changed) these fixes while we waited.
+            confirmed = {item.describe() for item in items}
+            plan = plan_convergence()
+            items = plan.executable()
+
+            new_items = [item for item in items if item.describe() not in confirmed]
+            if new_items and not yes:
+                # The plan grew while we waited at the prompt or for the
+                # lock — don't execute work the operator never approved.
+                click.secho(
+                    "The schema changed while waiting — new fixes not in the confirmed plan:",
+                    fg="red",
+                    bold=True,
+                )
+                for item in new_items:
+                    click.secho(f"  {item.describe()}", fg="red")
+                raise click.ClickException("Re-run to review the updated plan.")
+
+            result = execute_plan(items)
 
         for r in result.results:
             if r.ok:

plain-postgres/plain/postgres/cli/core.py

@@ -4,20 +4,19 @@
 import signal
 import subprocess
 import sys
-import time
 from collections import defaultdict
 
 import click
-import psycopg
 
 from plain.cli import register_cli
 
 from ..database_url import postgres_cli_args, postgres_cli_env
 from ..db import get_connection
 from ..dialect import quote_name
 from .converge import converge
-from .decorators import database_management_command
+from .decorators import cli_schema_lock, database_management_command
 from .diagnose import diagnose
+from .ready import ready
 from .schema import schema
 from .sync import sync
 
@@ -30,6 +29,7 @@ def cli() -> None:
 
 cli.add_command(converge)
 cli.add_command(diagnose)
+cli.add_command(ready)
 cli.add_command(schema)
 cli.add_command(sync)
 
@@ -126,36 +126,24 @@ def drop_unknown_tables(yes: bool) -> None:
         if not click.confirm(f"Drop {tables_label} (CASCADE)? This cannot be undone."):
             return
 
-    with conn.cursor() as cursor:
-        for table in unknown_tables:
-            click.echo(f"  Dropping {table}...", nl=False)
-            cursor.execute(f"DROP TABLE IF EXISTS {quote_name(table)} CASCADE")
-            click.echo(" OK")
-
-    click.secho(f"✓ Dropped {tables_label}.", fg="green")
-
-
-@cli.command()
-def wait() -> None:
-    """Wait for the database to be ready"""
-    attempts = 0
-    while True:
-        attempts += 1
-        waiting_for = False
-
-        try:
-            get_connection().ensure_connection()
-        except psycopg.OperationalError:
-            waiting_for = True
-
-        if waiting_for:
-            if attempts > 1:
-                # After the first attempt, start printing them
-                click.secho(
-                    f"Waiting for database (attempt {attempts})",
-                    fg="yellow",
-                )
-            time.sleep(1.5)
-        else:
-            click.secho("✔ Database ready", fg="green")
-            break
+    with cli_schema_lock():
+        # Re-check under the lock — while we sat at the prompt or waited for
+        # the lock, another process may have claimed one of these names for a
+        # real table (e.g. a migration creating a model table). Only drop
+        # what was unknown at confirm time AND is still unknown now.
+        still_unknown = set(get_unknown_tables(conn))
+        dropped_count = 0
+        with conn.cursor() as cursor:
+            for table in unknown_tables:
+                if table not in still_unknown:
+                    click.secho(f"  Skipping {table} — no longer unknown.", fg="yellow")
+                    continue
+                click.echo(f"  Dropping {table}...", nl=False)
+                cursor.execute(f"DROP TABLE IF EXISTS {quote_name(table)} CASCADE")
+                click.echo(" OK")
+                dropped_count += 1
+
+    click.secho(
+        f"✓ Dropped {dropped_count} table{'s' if dropped_count != 1 else ''}.",
+        fg="green",
+    )

plain-postgres/plain/postgres/cli/decorators.py

@@ -1,10 +1,99 @@
 from __future__ import annotations
 
 import functools
-from collections.abc import Callable
+import time
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
 from typing import Any
 
-from ..db import use_management_connection
+import click
+import psycopg
+
+from plain.runtime import settings
+
+from ..connection import DatabaseConnection
+from ..db import get_connection, use_management_connection
+from ..readiness import (
+    ReadinessStatus,
+    _classify_connection_failure,
+    _with_connect_timeout,
+)
+from ..schema_lock import SchemaLockLost, SchemaLockTimeout, schema_lock
+from ..sources import DirectSource
+
+
+@contextmanager
+def cli_schema_lock() -> Iterator[Callable[[], None]]:
+    """`schema_lock()` for CLI commands: a timeout or lost lock surfaces as a
+    clean one-line error instead of a traceback burying it. Yields the lock's
+    verify callable for multi-phase commands to check between phases."""
+    try:
+        with schema_lock() as verify:
+            yield verify
+    except (SchemaLockTimeout, SchemaLockLost) as e:
+        raise click.ClickException(str(e)) from e
+
+
+_WAIT_RETRY_INTERVAL_SECONDS = 2.0
+
+
+def cli_wait_for_database() -> None:
+    """Wait for the command's database to accept connections.
+
+    Schema commands call this first so a database that's still starting
+    (a deploy, dev services coming up, a failover) is retried for up to
+    `POSTGRES_WAIT_TIMEOUT` seconds instead of failing instantly. It probes
+    whatever connection the command will use — the management connection
+    when one is configured. Configuration errors (bad credentials, bad URL,
+    invalid options) fail immediately with a clean message; retrying can't
+    fix those.
+    """
+    conn = get_connection()
+    if conn.connection is not None:
+        # Already connected (e.g. a command invoked from another command
+        # that already waited) — nothing to wait for.
+        return
+
+    # Probe on a separate short-lived connection with a bounded connect
+    # timeout, so an unroutable host can't hang an attempt for minutes.
+    probe = DatabaseConnection(DirectSource(_with_connect_timeout(conn.settings_dict)))
+    deadline = time.monotonic() + settings.POSTGRES_WAIT_TIMEOUT
+    attempts = 0
+
+    try:
+        while True:
+            try:
+                probe.ensure_connection()
+                return
+            except psycopg.OperationalError as e:
+                if _classify_connection_failure(e) is ReadinessStatus.CONFIG_ERROR:
+                    raise click.ClickException(str(e).strip()) from e
+                # `or` fallback: an empty error message would otherwise crash
+                # the progress line's splitlines()[0].
+                error = str(e).strip() or "connection error"
+            except psycopg.ProgrammingError as e:
+                # An invalid connection option in the URL — config-shaped.
+                raise click.ClickException(str(e).strip()) from e
+
+            attempts += 1
+            if time.monotonic() >= deadline:
+                raise click.ClickException(
+                    f"Database not reachable after {attempts} attempt(s) "
+                    f"(POSTGRES_WAIT_TIMEOUT={settings.POSTGRES_WAIT_TIMEOUT:g}):\n"
+                    f"{error}"
+                )
+            # Progress goes to stderr so wrappers that capture stdout (like
+            # plain-dev's sync --check) still stream the waiting lines live.
+            click.secho(
+                f"Waiting for database (attempt {attempts}): {error.splitlines()[0]}",
+                fg="yellow",
+                err=True,
+            )
+            time.sleep(
+                min(_WAIT_RETRY_INTERVAL_SECONDS, max(0.0, deadline - time.monotonic()))
+            )
+    finally:
+        probe.close()
 
 
 def database_management_command[F: Callable[..., Any]](f: F) -> F: