From f12f5f1bd44a6fb3f722ecf2408b931bdf10285a Mon Sep 17 00:00:00 2001 From: Dave Gaeddert Date: Thu, 16 Jul 2026 10:24:58 -0500 Subject: [PATCH 1/2] Serialize schema-changing commands with an advisory schema lock plain postgres sync, plain migrations apply, plain postgres converge, and plain postgres drop-unknown-tables now take a single session-level advisory lock (key 1047265496 = crc32(b"plain/schema")) held on a dedicated connection, so concurrent deploy processes (a retried migrate job, two operators, overlapping release phases) can't interleave schema changes. - Non-blocking acquire with retry via POSTGRES_SCHEMA_LOCK_RETRY_INTERVAL / POSTGRES_SCHEMA_LOCK_MAX_RETRIES (defaults wait up to an hour for long index builds); warns about once a minute with the holder's pid while waiting; timeouts surface as a clean CLI error, not a traceback. - Commands re-plan under the lock, so the loser of a race sees the winner's applied work instead of executing a stale plan; interactive converge aborts if the plan grew past what the operator confirmed; drop-unknown- tables re-checks and skips names that are no longer unknown. - The lock connection uses TCP keepalives, sync re-verifies the lock between its migrate and converge phases (SchemaLockLost), nested acquisition fails fast, and a dead holder self-clears when its session closes. --- plain-postgres/plain/postgres/README.md | 20 ++ plain-postgres/plain/postgres/cli/converge.py | 24 ++- plain-postgres/plain/postgres/cli/core.py | 30 ++- .../plain/postgres/cli/decorators.py | 18 +- .../plain/postgres/cli/migrations.py | 124 ++++++----- plain-postgres/plain/postgres/cli/sync.py | 10 +- .../plain/postgres/default_settings.py | 7 + plain-postgres/plain/postgres/schema_lock.py | 194 ++++++++++++++++++ .../tests/internal/test_apply_replan.py | 56 +++++ .../tests/internal/test_schema_lock.py | 138 +++++++++++++ 10 files changed, 553 insertions(+), 68 deletions(-) create mode 100644 plain-postgres/plain/postgres/schema_lock.py create mode 100644 plain-postgres/tests/internal/test_apply_replan.py create mode 100644 plain-postgres/tests/internal/test_schema_lock.py diff --git a/plain-postgres/plain/postgres/README.md b/plain-postgres/plain/postgres/README.md index cd89b83aa7..f06bb0cb1d 100644 --- a/plain-postgres/plain/postgres/README.md +++ b/plain-postgres/plain/postgres/README.md @@ -775,6 +775,26 @@ 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`. + ## Fields You can use many field types for different data: diff --git a/plain-postgres/plain/postgres/cli/converge.py b/plain-postgres/plain/postgres/cli/converge.py index 7e206e6e8e..268a84ac36 100644 --- a/plain-postgres/plain/postgres/cli/converge.py +++ b/plain-postgres/plain/postgres/cli/converge.py @@ -5,7 +5,7 @@ import click from ..convergence import execute_plan, plan_convergence -from .decorators import database_management_command +from .decorators import cli_schema_lock, database_management_command @click.command() @@ -48,7 +48,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: diff --git a/plain-postgres/plain/postgres/cli/core.py b/plain-postgres/plain/postgres/cli/core.py index b9d5398f28..0fea63c869 100644 --- a/plain-postgres/plain/postgres/cli/core.py +++ b/plain-postgres/plain/postgres/cli/core.py @@ -16,7 +16,7 @@ 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 .schema import schema from .sync import sync @@ -126,13 +126,27 @@ 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") + 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", + ) @cli.command() diff --git a/plain-postgres/plain/postgres/cli/decorators.py b/plain-postgres/plain/postgres/cli/decorators.py index 0ca17322b9..7fdd18c239 100644 --- a/plain-postgres/plain/postgres/cli/decorators.py +++ b/plain-postgres/plain/postgres/cli/decorators.py @@ -1,10 +1,26 @@ from __future__ import annotations import functools -from collections.abc import Callable +from collections.abc import Callable, Iterator +from contextlib import contextmanager from typing import Any +import click + from ..db import use_management_connection +from ..schema_lock import SchemaLockLost, SchemaLockTimeout, schema_lock + + +@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 def database_management_command[F: Callable[..., Any]](f: F) -> F: diff --git a/plain-postgres/plain/postgres/cli/migrations.py b/plain-postgres/plain/postgres/cli/migrations.py index d7dea9daf9..9580df9124 100644 --- a/plain-postgres/plain/postgres/cli/migrations.py +++ b/plain-postgres/plain/postgres/cli/migrations.py @@ -27,7 +27,7 @@ from ..migrations.state import ModelState, ProjectState from ..migrations.writer import MigrationWriter from ..registry import models_registry -from .decorators import database_management_command +from .decorators import cli_schema_lock, database_management_command if TYPE_CHECKING: from ..connection import DatabaseConnection @@ -511,68 +511,84 @@ def describe_operation(operation: Any) -> tuple[str, bool]: click.secho(package_label, dim=True) click.echo() # Add newline after package - pre_migrate_state = executor._create_project_state(with_applied_migrations=True) - if migration_plan: - # Determine whether to use atomic batch - use_atomic_batch = False - atomic_batch_message = None - if len(migration_plan) > 1: - # Check if all migrations support atomic - non_atomic_migrations = [m for m in migration_plan if not m.atomic] - - if atomic_batch is True: - # User explicitly requested atomic batch - if non_atomic_migrations: - names = ", ".join( - f"{m.package_label}.{m.name}" for m in non_atomic_migrations[:3] - ) - if len(non_atomic_migrations) > 3: - names += f", and {len(non_atomic_migrations) - 3} more" - raise click.UsageError( - f"--atomic-batch requested but these migrations have atomic=False: {names}" - ) - use_atomic_batch = True - atomic_batch_message = ( - f"Running {len(migration_plan)} migrations in atomic batch" - ) - elif atomic_batch is False: - # User explicitly disabled atomic batch - use_atomic_batch = False - if len(migration_plan) > 1: - atomic_batch_message = ( - f"Running {len(migration_plan)} migrations separately" + with cli_schema_lock(): + # Re-plan under the lock — another process may have applied some + # or all of these migrations while we waited for it. + executor = MigrationExecutor(get_connection(), migration_progress_callback) + migration_plan = executor.migration_plan(targets) + if not migration_plan: + if not quiet: + click.echo( + "No migrations to apply (another process already applied them)." ) - else: - # Auto-detect (atomic_batch is None) - # Use atomic batch by default - if not non_atomic_migrations: + return + + # Determine whether to use atomic batch + use_atomic_batch = False + atomic_batch_message = None + if len(migration_plan) > 1: + # Check if all migrations support atomic + non_atomic_migrations = [m for m in migration_plan if not m.atomic] + + if atomic_batch is True: + # User explicitly requested atomic batch + if non_atomic_migrations: + names = ", ".join( + f"{m.package_label}.{m.name}" + for m in non_atomic_migrations[:3] + ) + if len(non_atomic_migrations) > 3: + names += f", and {len(non_atomic_migrations) - 3} more" + raise click.UsageError( + f"--atomic-batch requested but these migrations have atomic=False: {names}" + ) use_atomic_batch = True atomic_batch_message = ( f"Running {len(migration_plan)} migrations in atomic batch" ) - else: + elif atomic_batch is False: + # User explicitly disabled atomic batch use_atomic_batch = False if len(migration_plan) > 1: - atomic_batch_message = f"Running {len(migration_plan)} migrations separately (some have atomic=False)" - - if not quiet: - click.echo() # Add blank line before applying + atomic_batch_message = ( + f"Running {len(migration_plan)} migrations separately" + ) + else: + # Auto-detect (atomic_batch is None) + # Use atomic batch by default + if not non_atomic_migrations: + use_atomic_batch = True + atomic_batch_message = ( + f"Running {len(migration_plan)} migrations in atomic batch" + ) + else: + use_atomic_batch = False + if len(migration_plan) > 1: + atomic_batch_message = f"Running {len(migration_plan)} migrations separately (some have atomic=False)" + + if not quiet: + click.echo() # Add blank line before applying + + if not quiet: + if atomic_batch_message: + click.secho( + f"Applying migrations ({atomic_batch_message.lower()}):", + bold=True, + ) + else: + click.secho("Applying migrations:", bold=True) - if not quiet: - if atomic_batch_message: - click.secho( - f"Applying migrations ({atomic_batch_message.lower()}):", bold=True - ) - else: - click.secho("Applying migrations:", bold=True) - post_migrate_state = executor.migrate( - targets, - plan=migration_plan, - state=pre_migrate_state.clone(), - fake=fake, - atomic_batch=use_atomic_batch, - ) + pre_migrate_state = executor._create_project_state( + with_applied_migrations=True + ) + post_migrate_state = executor.migrate( + targets, + plan=migration_plan, + state=pre_migrate_state.clone(), + fake=fake, + atomic_batch=use_atomic_batch, + ) # post_migrate signals have access to all models. Ensure that all models # are reloaded in case any are delayed. post_migrate_state.clear_delayed_models_cache() diff --git a/plain-postgres/plain/postgres/cli/sync.py b/plain-postgres/plain/postgres/cli/sync.py index 2b860e0570..ef8d78790a 100644 --- a/plain-postgres/plain/postgres/cli/sync.py +++ b/plain-postgres/plain/postgres/cli/sync.py @@ -7,7 +7,7 @@ from plain.runtime import settings from ..convergence import execute_plan, plan_convergence -from .decorators import database_management_command +from .decorators import cli_schema_lock, database_management_command @click.command() @@ -33,8 +33,12 @@ def sync(check: bool) -> None: if settings.DEBUG: _create_migrations() - _migrate() - _converge() + with cli_schema_lock() as verify_lock: + _migrate() + # If the lock session died during migrations, another process may + # already be converging — stop instead of interleaving with it. + verify_lock() + _converge() def _check() -> None: diff --git a/plain-postgres/plain/postgres/default_settings.py b/plain-postgres/plain/postgres/default_settings.py index d19ec46304..38fcd552a2 100644 --- a/plain-postgres/plain/postgres/default_settings.py +++ b/plain-postgres/plain/postgres/default_settings.py @@ -40,3 +40,10 @@ POSTGRES_MIGRATION_STATEMENT_TIMEOUT: str = "3s" POSTGRES_CONVERGENCE_LOCK_TIMEOUT: str = "3s" POSTGRES_CONVERGENCE_STATEMENT_TIMEOUT: str = "3s" + +# Retry budget for the schema advisory lock that serializes schema-changing +# commands (see schema_lock.py). Non-blocking acquire, retried; the defaults +# wait up to an hour (720 × 5s) because a legitimate holder can be mid +# index build. +POSTGRES_SCHEMA_LOCK_RETRY_INTERVAL: float = 5.0 +POSTGRES_SCHEMA_LOCK_MAX_RETRIES: int = 720 diff --git a/plain-postgres/plain/postgres/schema_lock.py b/plain-postgres/plain/postgres/schema_lock.py new file mode 100644 index 0000000000..57ce1dfcad --- /dev/null +++ b/plain-postgres/plain/postgres/schema_lock.py @@ -0,0 +1,194 @@ +"""One session-level advisory lock serializing schema-changing commands. + +`plain postgres sync`, `plain migrations apply`, `plain postgres converge`, +and `plain postgres drop-unknown-tables` all take the same lock, so concurrent +processes (a retried migrate Job, two operators, overlapping release phases) +can't interleave schema changes. + +The lock is held on a dedicated connection, separate from the connection the +DDL runs on. Session-level advisory locks aren't bound to a transaction, so +non-transactional DDL (CREATE INDEX CONCURRENTLY, VALIDATE CONSTRAINT) runs +freely on the working connection while the lock connection just sits there +holding it. If the process dies, the session closes and Postgres releases the +lock automatically — no stale-lock cleanup step. + +Advisory locks are per-database, so parallel test databases and multi-tenant +clusters don't contend. +""" + +from __future__ import annotations + +import time +from collections.abc import Callable, Iterator +from contextlib import contextmanager + +import psycopg + +from plain.exceptions import ImproperlyConfigured +from plain.logs import get_framework_logger +from plain.runtime import settings + +from .db import get_connection +from .sources import build_connection_params + +logger = get_framework_logger() + +# Guards against nested schema_lock() in the same process, which would block +# on itself until the retry budget ran out (each entry is its own database +# session, so Postgres sees two different lock requesters). Commands are +# single-threaded, so a module flag is enough. +_held_by_this_process = False + +# Fixed key so every version of Plain computes the same value (mixed-version +# deploys must contend on the same lock). Derived once from the lock's name — +# zlib.crc32(b"plain/schema") — and frozen as a literal. The name identifies +# the concept, not this module, so it stays true if the code moves; any future +# Plain advisory lock should follow the same crc32(b"plain/") +# convention. Debuggable directly: +# SELECT * FROM pg_locks WHERE locktype = 'advisory' AND objid = 1047265496 +SCHEMA_LOCK_KEY = 1047265496 + + +class SchemaLockTimeout(Exception): + """The schema advisory lock couldn't be acquired within the retry budget.""" + + +class SchemaLockLost(Exception): + """The lock session died mid-hold, so the lock may now be held elsewhere.""" + + +@contextmanager +def schema_lock() -> Iterator[Callable[[], None]]: + """Hold the schema advisory lock for the duration of the block. + + Acquisition is non-blocking with retry (`pg_try_advisory_lock`), so a + dead lock holder surfaces as a clear timeout instead of hanging forever. + Retry behavior is controlled by `POSTGRES_SCHEMA_LOCK_RETRY_INTERVAL` and + `POSTGRES_SCHEMA_LOCK_MAX_RETRIES`. + + Yields a verify callable: it raises `SchemaLockLost` if the lock session + has died (the lock releases with it, so another process may now hold it). + Multi-phase callers check it between phases — losing the lock mid-phase + is left to the environment guidance in the README (keepalives are enabled; + don't set `idle_session_timeout` on the management role). + + Not reentrant: a nested `schema_lock()` raises immediately (it would open + a second session and block on the outer one). Commands take the lock once, + at the top. + """ + global _held_by_this_process + if _held_by_this_process: + raise RuntimeError( + "Schema lock is already held by this process — schema_lock() is " + "not reentrant. Take the lock once, at the top of the command." + ) + + config = get_connection().settings_dict + params = build_connection_params(config) + # The lock connection sits idle while DDL runs elsewhere — TCP keepalives + # stop NAT/LB idle timeouts from silently killing it (and the lock with it). + params["keepalives"] = 1 + params["keepalives_idle"] = 30 + params["keepalives_interval"] = 10 + params["keepalives_count"] = 3 + + retry_interval: float = settings.POSTGRES_SCHEMA_LOCK_RETRY_INTERVAL + max_retries: int = settings.POSTGRES_SCHEMA_LOCK_MAX_RETRIES + if retry_interval <= 0: + raise ImproperlyConfigured( + "POSTGRES_SCHEMA_LOCK_RETRY_INTERVAL must be greater than 0 " + f"(got {retry_interval!r})." + ) + # Warn immediately, then re-warn about once a minute so a long wait + # reads as "still waiting" in deploy logs, not a hang. + warn_every = max(1, round(60 / retry_interval)) + + with psycopg.connect(**params, autocommit=True) as lock_conn: + attempts = 0 + while True: + row = lock_conn.execute( + "SELECT pg_try_advisory_lock(%s)", [SCHEMA_LOCK_KEY] + ).fetchone() + assert row is not None + if row[0]: + break + + attempts += 1 + # Sleeps happen after this counter bumps, so the elapsed wait is + # one interval behind the attempt count. + waited_seconds = round((attempts - 1) * retry_interval) + if attempts == 1 or attempts % warn_every == 0: + logger.warning( + "Waiting for schema lock held by another process", + extra={ + "context": { + "holder": _describe_holder(lock_conn), + "waited_seconds": waited_seconds, + } + }, + ) + if attempts >= max_retries: + raise SchemaLockTimeout( + f"Could not acquire the schema lock after {attempts} attempt(s) " + f"({waited_seconds}s). Another process is running schema " + f"changes: {_describe_holder(lock_conn)}. If that process " + "is gone, its lock releases when its database session closes." + ) + time.sleep(retry_interval) + + def verify() -> None: + """Raise SchemaLockLost if the lock session is no longer alive.""" + try: + lock_conn.execute("SELECT 1") + except psycopg.Error as e: + raise SchemaLockLost( + "The schema lock session died while work was running, so " + "the lock has been released and another process may now be " + "making schema changes. Stopping here — re-run the command." + ) from e + + _held_by_this_process = True + try: + yield verify + finally: + _held_by_this_process = False + # No explicit pg_advisory_unlock: closing the session (the + # `with psycopg.connect(...)` exit below) releases the lock, and + # an unlock attempt on a connection that died mid-block would + # raise here and mask the block's real exception. + + +def _describe_holder(lock_conn: psycopg.Connection) -> str: + """Best-effort description of the session currently holding the lock.""" + # A bigint advisory key shows up in pg_locks split across classid + # (high half) and objid (low half), with objsubid = 1. + row = lock_conn.execute( + """ + SELECT l.pid, a.application_name, a.usename, a.query_start + FROM pg_locks l + LEFT JOIN pg_stat_activity a ON a.pid = l.pid + WHERE l.locktype = 'advisory' + AND l.classid = %s + AND l.objid = %s + AND l.objsubid = 1 + AND l.granted + AND l.database = ( + SELECT oid FROM pg_database WHERE datname = current_database() + ) + LIMIT 1 + """, + [SCHEMA_LOCK_KEY >> 32, SCHEMA_LOCK_KEY & 0xFFFFFFFF], + ).fetchone() + + if row is None: + return "no holder found (it may have just released)" + + pid, application_name, usename, query_start = row + details = [f"pid={pid}"] + if application_name: + details.append(f"application={application_name}") + if usename: + details.append(f"user={usename}") + if query_start: + details.append(f"since={query_start:%Y-%m-%d %H:%M:%S}") + return ", ".join(details) diff --git a/plain-postgres/tests/internal/test_apply_replan.py b/plain-postgres/tests/internal/test_apply_replan.py new file mode 100644 index 0000000000..eadb0ce88d --- /dev/null +++ b/plain-postgres/tests/internal/test_apply_replan.py @@ -0,0 +1,56 @@ +"""The `migrations apply` command re-plans under the schema lock. + +Pins the concurrent-deploy contract: two processes both plan the same pending +migration; the loser waits on the lock, then must re-plan and see the winner's +work instead of re-applying its stale plan (which would re-run DDL that +already exists and crash). +""" + +from __future__ import annotations + +from contextlib import contextmanager + +from plain.postgres.cli.migrations import apply +from plain.postgres.db import get_connection +from plain.postgres.migrations.recorder import MigrationRecorder + + +def test_apply_replans_after_waiting_on_lock(db, monkeypatch, capsys): + recorder = MigrationRecorder(get_connection()) + + # Pick the latest applied migration of the test app and delete its record + # (schema untouched) — `apply` now sees it as pending, exactly like the + # loser of a two-process race that planned before the winner finished. + applied = [name for app, name in recorder.applied_migrations() if app == "examples"] + assert applied, "test app should have applied migrations" + latest = max(applied) + recorder.record_unapplied("examples", latest) + + # Simulate the winner: the moment the lock is acquired, the migration is + # (already) applied. If `apply` executed its stale pre-lock plan instead + # of re-planning, it would re-run this migration's DDL and crash. + import plain.postgres.cli.migrations as migrations_cli + + real_lock = migrations_cli.cli_schema_lock + + @contextmanager + def lock_with_racing_winner(): + with real_lock() as verify: + recorder.record_applied("examples", latest) + yield verify + + monkeypatch.setattr(migrations_cli, "cli_schema_lock", lock_with_racing_winner) + + apply.callback( + package_label=None, + migration_name=None, + fake=False, + plan=False, + check_unapplied=False, + no_input=True, + atomic_batch=None, + quiet=False, + ) + + out = capsys.readouterr().out + assert "another process already applied them" in out diff --git a/plain-postgres/tests/internal/test_schema_lock.py b/plain-postgres/tests/internal/test_schema_lock.py new file mode 100644 index 0000000000..cfaa80161c --- /dev/null +++ b/plain-postgres/tests/internal/test_schema_lock.py @@ -0,0 +1,138 @@ +"""Tests for the schema advisory lock that serializes schema-changing commands. + +The lock lives on its own database session (separate from the working +connection), so these tests observe it from the test connection via pg_locks. +""" + +from __future__ import annotations + +import psycopg +import pytest + +from plain.postgres.db import get_connection +from plain.postgres.schema_lock import ( + SCHEMA_LOCK_KEY, + SchemaLockTimeout, + schema_lock, +) +from plain.postgres.sources import build_connection_params + + +def _lock_holder_count() -> int: + with get_connection().cursor() as cursor: + cursor.execute( + """ + SELECT count(*) + FROM pg_locks + WHERE locktype = 'advisory' + AND classid = %s + AND objid = %s + AND objsubid = 1 + AND granted + AND database = ( + SELECT oid FROM pg_database WHERE datname = current_database() + ) + """, + [SCHEMA_LOCK_KEY >> 32, SCHEMA_LOCK_KEY & 0xFFFFFFFF], + ) + row = cursor.fetchone() + assert row is not None + return row[0] + + +def test_lock_held_during_block_and_released_after(db): + assert _lock_holder_count() == 0 + + with schema_lock(): + assert _lock_holder_count() == 1 + + assert _lock_holder_count() == 0 + + +def test_lock_reacquirable_after_release(db): + with schema_lock(): + pass + + with schema_lock(): + assert _lock_holder_count() == 1 + + +def test_acquisition_times_out_when_held_by_another_session(db, settings): + settings.POSTGRES_SCHEMA_LOCK_RETRY_INTERVAL = 0.01 + settings.POSTGRES_SCHEMA_LOCK_MAX_RETRIES = 3 + + # Hold the lock from a separate session, like another process would. + params = build_connection_params(get_connection().settings_dict) + with psycopg.connect(**params, autocommit=True) as holder: + holder.execute("SELECT pg_advisory_lock(%s)", [SCHEMA_LOCK_KEY]) + + with pytest.raises(SchemaLockTimeout) as exc_info: + with schema_lock(): + pass + + # The failed acquisition didn't disturb the holder. + assert _lock_holder_count() == 1 + + message = str(exc_info.value) + assert "3 attempt(s)" in message + assert "pid=" in message # identifies the holder + + +def test_lock_released_when_block_raises(db): + with pytest.raises(RuntimeError): + with schema_lock(): + raise RuntimeError("boom") + + assert _lock_holder_count() == 0 + + +def test_nested_acquisition_raises_immediately(db): + with schema_lock(): + with pytest.raises(RuntimeError, match="not reentrant"): + with schema_lock(): + pass + + # The failed nested entry didn't corrupt the held flag. + with schema_lock(): + assert _lock_holder_count() == 1 + + +def test_invalid_retry_interval_rejected(db, settings): + from plain.exceptions import ImproperlyConfigured + + settings.POSTGRES_SCHEMA_LOCK_RETRY_INTERVAL = 0.0 + + with pytest.raises(ImproperlyConfigured, match="must be greater than 0"): + with schema_lock(): + pass + + +def test_verify_passes_while_session_alive(db): + with schema_lock() as verify: + verify() # no exception — session is alive, lock held + + +def test_verify_raises_when_lock_session_killed(db): + from plain.postgres.schema_lock import SchemaLockLost + + with schema_lock() as verify: + # Find the lock holder's backend pid and kill it, simulating an + # idle-session timeout / NAT drop / failover. + with get_connection().cursor() as cursor: + cursor.execute( + """ + SELECT pid FROM pg_locks + WHERE locktype = 'advisory' + AND classid = %s AND objid = %s AND objsubid = 1 AND granted + AND database = ( + SELECT oid FROM pg_database WHERE datname = current_database() + ) + """, + [SCHEMA_LOCK_KEY >> 32, SCHEMA_LOCK_KEY & 0xFFFFFFFF], + ) + row = cursor.fetchone() + assert row is not None + cursor.execute("SELECT pg_terminate_backend(%s)", [row[0]]) + + with pytest.raises(SchemaLockLost, match="re-run the command"): + verify() From 185a91cf76ff6db20f5df8c4f30eaf35918c4c1e Mon Sep 17 00:00:00 2001 From: Dave Gaeddert Date: Thu, 16 Jul 2026 22:08:00 -0500 Subject: [PATCH 2/2] Add plain postgres ready and build connectivity waits into schema commands MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A readiness engine (plain/postgres/readiness.py) classifies whether the database is ready for this code to serve: reachable, migrations applied, and every model's table and columns present (existence only — drift is convergence's job). `plain postgres ready [--wait --timeout] [--json]` maps the classification to an operator contract: exit 0 = serve, 1 = not ready but retryable, 2 = configuration error a human must fix. DEBUG downgrades schema gaps to warnings at the CLI; data-only (all-RunPython) migrations warn instead of gate. Schema commands (sync, migrations apply, converge) now wait for the database themselves — connection failures retry for up to POSTGRES_WAIT_TIMEOUT seconds while configuration errors fail immediately via the same classifier — so release phases and dev environments no longer need a separate wait step. The `plain postgres wait` command is removed and plain-dev relies on sync's built-in wait. Float-annotated settings now accept int values (PEP 484 numeric tower), so PLAIN_POSTGRES_WAIT_TIMEOUT=30 and friends work naturally. --- plain-dev/plain/dev/core.py | 27 +- plain-postgres/plain/postgres/README.md | 35 ++ plain-postgres/plain/postgres/cli/converge.py | 8 +- plain-postgres/plain/postgres/cli/core.py | 30 +- .../plain/postgres/cli/decorators.py | 75 +++- .../plain/postgres/cli/migrations.py | 7 +- plain-postgres/plain/postgres/cli/ready.py | 187 ++++++++++ plain-postgres/plain/postgres/cli/sync.py | 8 +- .../plain/postgres/default_settings.py | 8 + plain-postgres/plain/postgres/readiness.py | 321 ++++++++++++++++++ .../0019_readiness_data_migration.py | 16 + .../internal/test_readiness_classification.py | 149 ++++++++ .../tests/internal/test_wait_for_database.py | 56 +++ plain-postgres/tests/public/test_ready.py | 165 +++++++++ plain/plain/runtime/user_settings.py | 10 + .../internal/test_setting_int_to_float.py | 46 +++ 16 files changed, 1104 insertions(+), 44 deletions(-) create mode 100644 plain-postgres/plain/postgres/cli/ready.py create mode 100644 plain-postgres/plain/postgres/readiness.py create mode 100644 plain-postgres/tests/app/examples/migrations/0019_readiness_data_migration.py create mode 100644 plain-postgres/tests/internal/test_readiness_classification.py create mode 100644 plain-postgres/tests/internal/test_wait_for_database.py create mode 100644 plain-postgres/tests/public/test_ready.py create mode 100644 plain/tests/internal/test_setting_int_to_float.py diff --git a/plain-dev/plain/dev/core.py b/plain-dev/plain/dev/core.py index 07e796936e..44c79874ab 100644 --- a/plain-dev/plain/dev/core.py +++ b/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...") diff --git a/plain-postgres/plain/postgres/README.md b/plain-postgres/plain/postgres/README.md index f06bb0cb1d..964e1a0f6a 100644 --- a/plain-postgres/plain/postgres/README.md +++ b/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) | @@ -795,6 +798,37 @@ The lock connection sits idle while your DDL runs, so it enables TCP keepalives 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 ` 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: @@ -1470,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. diff --git a/plain-postgres/plain/postgres/cli/converge.py b/plain-postgres/plain/postgres/cli/converge.py index 268a84ac36..1c937b44a6 100644 --- a/plain-postgres/plain/postgres/cli/converge.py +++ b/plain-postgres/plain/postgres/cli/converge.py @@ -5,7 +5,11 @@ import click from ..convergence import execute_plan, plan_convergence -from .decorators import cli_schema_lock, 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 diff --git a/plain-postgres/plain/postgres/cli/core.py b/plain-postgres/plain/postgres/cli/core.py index 0fea63c869..0034cb3aa9 100644 --- a/plain-postgres/plain/postgres/cli/core.py +++ b/plain-postgres/plain/postgres/cli/core.py @@ -4,11 +4,9 @@ import signal import subprocess import sys -import time from collections import defaultdict import click -import psycopg from plain.cli import register_cli @@ -18,6 +16,7 @@ from .converge import converge 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) @@ -147,29 +147,3 @@ def drop_unknown_tables(yes: bool) -> None: f"✓ Dropped {dropped_count} table{'s' if dropped_count != 1 else ''}.", 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 diff --git a/plain-postgres/plain/postgres/cli/decorators.py b/plain-postgres/plain/postgres/cli/decorators.py index 7fdd18c239..f0ce26861f 100644 --- a/plain-postgres/plain/postgres/cli/decorators.py +++ b/plain-postgres/plain/postgres/cli/decorators.py @@ -1,14 +1,25 @@ from __future__ import annotations import functools +import time from collections.abc import Callable, Iterator from contextlib import contextmanager from typing import Any import click +import psycopg -from ..db import use_management_connection +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 @@ -23,6 +34,68 @@ def cli_schema_lock() -> Iterator[Callable[[], None]]: 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: """Run a click command's body through `use_management_connection()`. diff --git a/plain-postgres/plain/postgres/cli/migrations.py b/plain-postgres/plain/postgres/cli/migrations.py index 9580df9124..0a1eac752c 100644 --- a/plain-postgres/plain/postgres/cli/migrations.py +++ b/plain-postgres/plain/postgres/cli/migrations.py @@ -27,7 +27,11 @@ from ..migrations.state import ModelState, ProjectState from ..migrations.writer import MigrationWriter from ..registry import models_registry -from .decorators import cli_schema_lock, database_management_command +from .decorators import ( + cli_schema_lock, + cli_wait_for_database, + database_management_command, +) if TYPE_CHECKING: from ..connection import DatabaseConnection @@ -359,6 +363,7 @@ def apply( quiet: bool, ) -> None: """Apply database migrations""" + cli_wait_for_database() def migration_progress_callback( action: str, diff --git a/plain-postgres/plain/postgres/cli/ready.py b/plain-postgres/plain/postgres/cli/ready.py new file mode 100644 index 0000000000..712a0a5ba8 --- /dev/null +++ b/plain-postgres/plain/postgres/cli/ready.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +import json +import sys +import time + +import click + +from plain.runtime import settings + +from ..readiness import ReadinessResult, ReadinessStatus, check_database_ready + +_WAIT_INTERVAL_SECONDS = 2.0 + + +@click.command() +@click.option( + "--wait", + is_flag=True, + help="Poll until the database is ready. Configuration errors still exit immediately.", +) +@click.option( + "--timeout", + type=float, + help="With --wait, give up (exit 1) after this many seconds.", +) +@click.option("--json", "output_json", is_flag=True, help="Output as JSON") +def ready(wait: bool, timeout: float | None, output_json: bool) -> None: + """Check that the database is ready for this application to serve. + + Verifies the database is reachable, all migrations are applied, and the + schema has every model's table and columns. Runs against `POSTGRES_URL` — + the connection the app serves through — so it belongs in serving + entrypoints (before `plain server` / `plain jobs worker`), not release + phases. + + Exit codes: 0 = ready, 1 = not ready but retryable (pending migrations, + schema gap, database unreachable), 2 = configuration error a human must + fix (bad credentials, database doesn't exist). + """ + if timeout is not None and not wait: + raise click.UsageError("--timeout requires --wait.") + if output_json and wait: + raise click.UsageError("--json cannot be combined with --wait.") + + if not wait: + result = check_database_ready() + exit_code = _exit_code(result) + if output_json: + # The exit code carries the CLI's policy (the DEBUG downgrade), + # which the classification fields alone can't convey. + payload = {"exit_code": exit_code, **result.to_dict()} + click.echo(json.dumps(payload, indent=2)) + else: + _print_result(result) + sys.exit(exit_code) + + deadline = time.monotonic() + timeout if timeout is not None else None + last_status: ReadinessStatus | None = None + attempts = 0 + + while True: + # Deadline check before the next attempt (but always attempt at + # least once) — an attempt can block for the full connect timeout, + # which would overshoot --timeout by a lot. + if attempts and deadline is not None and time.monotonic() >= deadline: + click.secho(f"Timed out after {timeout:g}s.", fg="red", bold=True) + sys.exit(1) + + result = check_database_ready() + attempts += 1 + exit_code = _exit_code(result) + + if exit_code != 1: + # Nothing to wait for: ready (or DEBUG warned past a gap), or a + # configuration error a human must fix. + _print_result(result) + sys.exit(exit_code) + + if result.status is not last_status: + # Full detail when the situation changes; one-liners between. + _print_result(result) + last_status = result.status + else: + click.secho( + f"Not ready (attempt {attempts}): {result.summary()}", + fg="yellow", + ) + + sleep_seconds = _WAIT_INTERVAL_SECONDS + if deadline is not None: + sleep_seconds = min(sleep_seconds, max(0.0, deadline - time.monotonic())) + time.sleep(sleep_seconds) + + +def _exit_code(result: ReadinessResult) -> int: + """Map the classification to the exit-code contract: 0 = serve, + 1 = retryable, 2 = a human must fix configuration. + + In DEBUG, schema gaps warn instead of gating — mid-development you + routinely have a new model with no migration yet, and refusing to boot + the dev server over a table you know about is hostile. Unreachable and + config errors still fail in DEBUG; there's no serving through those. + """ + match result.status: + case ReadinessStatus.READY: + return 0 + case ReadinessStatus.PENDING_MIGRATIONS | ReadinessStatus.SCHEMA_NOT_SATISFIED: + return 0 if settings.DEBUG else 1 + case ReadinessStatus.CONFIG_ERROR: + return 2 + case ReadinessStatus.UNREACHABLE: + return 1 + case _: + raise ValueError(f"Unhandled readiness status: {result.status}") + + +def _print_result(result: ReadinessResult) -> None: + match result.status: + case ReadinessStatus.READY: + click.secho("✔ Database ready", fg="green") + for name in result.pending_data_migrations: + click.secho( + f" ! {name} pending (data migration — not gating)", fg="yellow" + ) + case ReadinessStatus.PENDING_MIGRATIONS: + _print_not_ready_headline(result) + for name in result.pending_migrations: + click.echo(f" {name}") + for name in result.pending_data_migrations: + click.secho(f" {name} (data migration)", dim=True) + click.secho( + " Run `plain migrations apply` (or `plain postgres sync`) to apply them.", + dim=True, + ) + case ReadinessStatus.SCHEMA_NOT_SATISFIED: + _print_not_ready_headline(result) + for table in result.missing_tables: + click.echo(f" table {table}") + for column in result.missing_columns: + click.echo(f" column {column}") + for name in result.pending_data_migrations: + click.secho(f" {name} (data migration pending)", dim=True) + click.secho( + " No schema-affecting migrations are pending, so the database " + "may be from a different version of this code — a restored " + "backup, a rollback, or a database migrated by newer code.", + dim=True, + ) + case ReadinessStatus.UNREACHABLE: + _print_connection_error(result, "Database unreachable") + click.secho( + " This is usually temporary — the server may be starting, " + "restarting, or briefly unreachable.", + dim=True, + ) + case ReadinessStatus.CONFIG_ERROR: + _print_connection_error(result, "Database configuration error") + click.secho( + " Retrying will not fix this — check POSTGRES_URL and the " + "database server configuration.", + dim=True, + ) + + +def _print_not_ready_headline(result: ReadinessResult) -> None: + if settings.DEBUG: + click.secho( + f"! {result.summary()} — serving anyway in DEBUG, production would not", + fg="yellow", + bold=True, + ) + else: + click.secho(f"✗ Not ready: {result.summary()}", fg="red", bold=True) + + +def _print_connection_error(result: ReadinessResult, headline: str) -> None: + """Print a possibly multi-line psycopg error with a one-line headline. + + psycopg reports multi-host failures as several lines ("Multiple + connection attempts failed. All failures were: - host ..."); the first + line goes on the headline (matching `summary()`) and the rest indent + under it. + """ + click.secho(f"✗ {headline}: {result.summary()}", fg="red", bold=True) + for line in (result.connection_error or "").splitlines()[1:]: + click.secho(f" {line}", fg="red") diff --git a/plain-postgres/plain/postgres/cli/sync.py b/plain-postgres/plain/postgres/cli/sync.py index ef8d78790a..3431a7f4ea 100644 --- a/plain-postgres/plain/postgres/cli/sync.py +++ b/plain-postgres/plain/postgres/cli/sync.py @@ -7,7 +7,11 @@ from plain.runtime import settings from ..convergence import execute_plan, plan_convergence -from .decorators import cli_schema_lock, database_management_command +from .decorators import ( + cli_schema_lock, + cli_wait_for_database, + database_management_command, +) @click.command() @@ -26,6 +30,8 @@ def sync(check: bool) -> None: Undeclared indexes and constraints are automatically dropped — models are the source of truth. """ + cli_wait_for_database() + if check: _check() return diff --git a/plain-postgres/plain/postgres/default_settings.py b/plain-postgres/plain/postgres/default_settings.py index 38fcd552a2..8d3542a623 100644 --- a/plain-postgres/plain/postgres/default_settings.py +++ b/plain-postgres/plain/postgres/default_settings.py @@ -47,3 +47,11 @@ # index build. POSTGRES_SCHEMA_LOCK_RETRY_INTERVAL: float = 5.0 POSTGRES_SCHEMA_LOCK_MAX_RETRIES: int = 720 + +# How long schema commands (sync, migrations apply, converge) wait for the +# database to accept connections before giving up. Covers a database that's +# still starting (deploys, dev services, failovers) so those commands don't +# need a separate wait step in front of them. Configuration errors (bad +# credentials, bad URL) fail immediately regardless — retrying can't fix +# them. Set to 0 to fail on the first connection error. +POSTGRES_WAIT_TIMEOUT: float = 60.0 diff --git a/plain-postgres/plain/postgres/readiness.py b/plain-postgres/plain/postgres/readiness.py new file mode 100644 index 0000000000..e7998cc914 --- /dev/null +++ b/plain-postgres/plain/postgres/readiness.py @@ -0,0 +1,321 @@ +"""Classify whether the database is ready for this application to serve. + +Two checks with disjoint coverage: + +1. **Unapplied migrations** — the same plan `plain postgres sync` would run. + Catches pending data migrations that schema presence can't see. +2. **Schema satisfies models** — every model's table and concrete columns + exist. Existence only: index/constraint/nullability drift is convergence's + job, not a serving cliff. Catches what migration records can't — restored + backups, rollbacks, manual damage, an empty database. + +The result is a classification, not a verb. Whether a process should wait, +refuse, or serve anyway is deployment policy that belongs to the operator — +typically an entrypoint script wrapping `plain postgres ready`. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from enum import StrEnum +from typing import TYPE_CHECKING, Any + +import psycopg + +from plain.exceptions import ImproperlyConfigured + +from .connection import DatabaseConnection, get_migratable_models +from .db import _db_conn +from .fields.base import ColumnField +from .migrations.executor import MigrationExecutor +from .migrations.operations import RunPython +from .sources import DirectSource, _parse_runtime_url + +if TYPE_CHECKING: + from .database_url import DatabaseConfig + + +class ReadinessStatus(StrEnum): + READY = "ready" + PENDING_MIGRATIONS = "pending-migrations" + SCHEMA_NOT_SATISFIED = "schema-not-satisfied" + UNREACHABLE = "unreachable" + CONFIG_ERROR = "config-error" + + +@dataclass +class ReadinessResult: + """A readiness classification plus the facts behind it.""" + + status: ReadinessStatus + pending_migrations: list[str] = field(default_factory=list) + pending_data_migrations: list[str] = field(default_factory=list) + missing_tables: list[str] = field(default_factory=list) + missing_columns: list[str] = field(default_factory=list) + connection_error: str | None = None + + def summary(self) -> str: + """One-line description, suitable for a poll loop's log output.""" + match self.status: + case ReadinessStatus.READY: + return "ready" + case ReadinessStatus.PENDING_MIGRATIONS: + return f"{len(self.pending_migrations)} migration(s) pending" + case ReadinessStatus.SCHEMA_NOT_SATISFIED: + return ( + f"schema missing {len(self.missing_tables)} table(s) " + f"and {len(self.missing_columns)} column(s)" + ) + case _: # UNREACHABLE / CONFIG_ERROR + error = self.connection_error or "" + return error.splitlines()[0] if error else str(self.status) + + def to_dict(self) -> dict[str, Any]: + return { + "status": self.status.value, + "summary": self.summary(), + "pending_migrations": self.pending_migrations, + "pending_data_migrations": self.pending_data_migrations, + "missing_tables": self.missing_tables, + "missing_columns": self.missing_columns, + "connection_error": self.connection_error, + } + + +def check_database_ready(conn: DatabaseConnection | None = None) -> ReadinessResult: + """Run the readiness checks and classify the result. + + With no connection given, opens a short-lived direct connection against + `POSTGRES_URL` — the URL the app serves through — and closes it when + done. A direct connection (not the runtime pool) so a connection failure + surfaces as a classifiable error instead of an opaque pool timeout. + + Pass `conn` to run the checks over an existing connection instead — an + in-process caller that already holds one (tests, error diagnosis). + + Either way, the connection is installed as the active connection while + the checks run, so ORM reads inside them (the migration recorder) go + through it rather than opening the runtime pool. + """ + owns_connection = conn is None + if conn is None: + try: + conn = DatabaseConnection( + DirectSource(_with_connect_timeout(_parse_runtime_url())) + ) + except (ImproperlyConfigured, ValueError) as e: + # An unusable POSTGRES_URL — unsupported scheme, bad port, + # too-long database name — is permanent; a human must fix it. + return ReadinessResult( + status=ReadinessStatus.CONFIG_ERROR, connection_error=str(e) + ) + + token = _db_conn.set(conn) + try: + return _run_checks(conn) + finally: + _db_conn.reset(token) + if owns_connection: + conn.close() + + +# Without a bounded connect timeout, an unroutable host hangs a readiness +# probe for the full TCP timeout (minutes) with no output. +_CONNECT_TIMEOUT_SECONDS = 10 + + +def _with_connect_timeout(config: DatabaseConfig) -> DatabaseConfig: + """Copy of `config` with a bounded connect timeout for probing. + + A `connect_timeout` already present in the URL wins. Also used by + `cli_wait_for_database()` for the schema commands' connectivity wait. + """ + return { + **config, + "OPTIONS": { + "connect_timeout": _CONNECT_TIMEOUT_SECONDS, + **config.get("OPTIONS", {}), + }, + } + + +def _run_checks(conn: DatabaseConnection) -> ReadinessResult: + try: + conn.ensure_connection() + except psycopg.OperationalError as e: + return ReadinessResult( + status=_classify_connection_failure(e), + connection_error=str(e).strip(), + ) + except psycopg.ProgrammingError as e: + # psycopg rejects an unknown connection option (a typo'd URL query + # param) with ProgrammingError, not OperationalError. Config-shaped — + # only caught around connect, so a ProgrammingError from the check + # queries themselves (a bug) still propagates loudly. + return ReadinessResult( + status=ReadinessStatus.CONFIG_ERROR, + connection_error=str(e).strip(), + ) + + try: + pending, pending_data = _pending_migrations(conn) + if pending: + # Missing schema objects are expected while migrations are + # pending — the pending verdict wins, so skip the schema check. + return ReadinessResult( + status=ReadinessStatus.PENDING_MIGRATIONS, + pending_migrations=pending, + pending_data_migrations=pending_data, + ) + missing_tables, missing_columns = _missing_schema_objects(conn) + except psycopg.errors.InsufficientPrivilege as e: + # The role can't read something the checks need — typically SELECT + # on plainmigrations when a runtime/management role split misses a + # grant. A human must fix it. Kept narrow (42501, not all + # ProgrammingError) so a bug in our own check SQL still propagates + # loudly instead of classifying. + return ReadinessResult( + status=ReadinessStatus.CONFIG_ERROR, + connection_error=str(e).strip(), + ) + except psycopg.OperationalError as e: + return ReadinessResult( + status=_classify_connection_failure(e), + connection_error=str(e).strip(), + ) + + if missing_tables or missing_columns: + status = ReadinessStatus.SCHEMA_NOT_SATISFIED + else: + status = ReadinessStatus.READY + + return ReadinessResult( + status=status, + pending_data_migrations=pending_data, + missing_tables=missing_tables, + missing_columns=missing_columns, + ) + + +def _pending_migrations(conn: DatabaseConnection) -> tuple[list[str], list[str]]: + """Unapplied migrations, split into (schema-affecting, data-only). + + A migration whose operations are all `RunPython` can't change the + schema — a long backfill is legitimately pending for hours while new + processes serve fine, so those warn instead of gate. + """ + executor = MigrationExecutor(conn) + plan = executor.migration_plan(executor.loader.graph.leaf_nodes()) + + pending: list[str] = [] + data_only: list[str] = [] + for migration in plan: + if migration.operations and all( + isinstance(op, RunPython) for op in migration.operations + ): + data_only.append(str(migration)) + else: + pending.append(str(migration)) + return pending, data_only + + +def _missing_schema_objects(conn: DatabaseConnection) -> tuple[list[str], list[str]]: + """Model tables and columns missing from the database. + + Existence only, mirroring what the ORM requires to run at all: Plain's + SELECTs enumerate model columns, so a missing table or column is always + fatal to serving. Type/nullability/index drift is deliberately ignored — + that's convergence's gradient, not a serving cliff (Hibernate's + `validate` checks types too and earned its false-positive reputation + for it). + + One catalog query across all tables, rather than + `introspection.introspect_table()` per table — that path also pulls + constraints, indexes, and storage parameters, far more than existence + needs. + """ + expected: dict[str, list[str]] = {} + for model in get_migratable_models(): + columns = [] + for f in model._model_meta.local_fields: + if isinstance(f, ColumnField) and f.db_type() is not None: + columns.append(f.column) + expected[model.model_options.db_table] = columns + + with conn.cursor() as cursor: + cursor.execute( + """ + SELECT c.relname, a.attname + FROM pg_class c + JOIN pg_attribute a ON a.attrelid = c.oid + WHERE c.relname = ANY(%s) + AND c.relkind IN ('r', 'p', 'v', 'f', 'm') + AND pg_catalog.pg_table_is_visible(c.oid) + AND a.attnum > 0 AND NOT a.attisdropped + """, + [list(expected)], + ) + actual: dict[str, set[str]] = {} + for table, column in cursor.fetchall(): + actual.setdefault(table, set()).add(column) + + missing_tables: list[str] = [] + missing_columns: list[str] = [] + for table, columns in expected.items(): + if table not in actual: + missing_tables.append(table) + continue + missing_columns.extend( + f"{table}.{column}" for column in columns if column not in actual[table] + ) + return sorted(missing_tables), sorted(missing_columns) + + +# Server-sent FATAL messages that mean the configuration is wrong and a +# human must change something — retrying can never fix these. Matched as +# substrings of the connection error because psycopg3 exposes no sqlstate +# on connection-time failures (verified against psycopg 3.2: auth failure, +# bad database, refused, and DNS all raise OperationalError with +# sqlstate=None). Localized servers won't match and fall through to +# UNREACHABLE — the safe direction, since a wrongly-permanent verdict +# during a DB restart would mislead the operator, while a wrongly-retryable +# one just keeps polling with the real error printed. +_CONFIG_ERROR_MESSAGES = ( + "password authentication failed", # 28P01 + "no pg_hba.conf entry", # 28000 + '" does not exist', # 3D000 database / 28000 role +) + +# libpq rejects a bad value for a recognized connection option client-side, +# before any network I/O: 'invalid sslmode value: "bogus"', "invalid +# integer value ... for connection option ...", "invalid channel_binding +# value", etc. One pattern covers the family. +_INVALID_OPTION_VALUE_RE = re.compile(r"\binvalid \w+ value") + + +def _classify_connection_failure(error: psycopg.OperationalError) -> ReadinessStatus: + """Split connection failures into permanent (config) vs retryable. + + Everything unmatched is retryable: "server starting up", "too many + connections", refused, timeout, DNS — all resolve on their own or with + time, and defaulting to retryable is the direction that never misleads. + """ + if error.sqlstate: + # Query-time failures carry a sqlstate. Class 28 (auth) and class 3D + # (invalid catalog) are config-shaped; everything else that reaches + # here (connection loss, admin shutdown, resource limits) is + # retryable. Today 3D000 only occurs at connect time where sqlstate + # is None (psycopg 3.2), so it's handled by the message matching + # below — the sqlstate hedge is for a future psycopg that populates + # it. + if error.sqlstate.startswith(("28", "3D")): + return ReadinessStatus.CONFIG_ERROR + return ReadinessStatus.UNREACHABLE + + message = str(error) + if any(marker in message for marker in _CONFIG_ERROR_MESSAGES): + return ReadinessStatus.CONFIG_ERROR + if _INVALID_OPTION_VALUE_RE.search(message): + return ReadinessStatus.CONFIG_ERROR + return ReadinessStatus.UNREACHABLE diff --git a/plain-postgres/tests/app/examples/migrations/0019_readiness_data_migration.py b/plain-postgres/tests/app/examples/migrations/0019_readiness_data_migration.py new file mode 100644 index 0000000000..5eacd8d511 --- /dev/null +++ b/plain-postgres/tests/app/examples/migrations/0019_readiness_data_migration.py @@ -0,0 +1,16 @@ +from plain.postgres import migrations + + +def noop_backfill(models, schema_editor): + """Does nothing — exists so tests have an all-RunPython migration whose + pending state should warn, not gate, readiness.""" + + +class Migration(migrations.Migration): + dependencies = [ + ("examples", "0018_storageparametersexample"), + ] + + operations = [ + migrations.RunPython(noop_backfill), + ] diff --git a/plain-postgres/tests/internal/test_readiness_classification.py b/plain-postgres/tests/internal/test_readiness_classification.py new file mode 100644 index 0000000000..367ed026e3 --- /dev/null +++ b/plain-postgres/tests/internal/test_readiness_classification.py @@ -0,0 +1,149 @@ +"""Pins the connection-failure taxonomy in `readiness.py`. + +psycopg3 exposes no sqlstate on connection-time failures (verified against +psycopg 3.2), so classification falls back to message matching there — these +tests pin both paths and the safe default (unknown → retryable). +""" + +from __future__ import annotations + +import psycopg +import pytest + +from plain.exceptions import ImproperlyConfigured +from plain.postgres import readiness +from plain.postgres.connection import DatabaseConnection +from plain.postgres.database_url import DatabaseConfig +from plain.postgres.db import get_connection +from plain.postgres.readiness import ( + ReadinessStatus, + _classify_connection_failure, + check_database_ready, +) +from plain.postgres.sources import DirectSource + + +@pytest.mark.parametrize( + ("message", "expected"), + [ + # Connection-time failures: sqlstate is None, match on the message. + ( + 'connection failed: FATAL: password authentication failed for user "app"', + ReadinessStatus.CONFIG_ERROR, + ), + ( + 'connection failed: FATAL: no pg_hba.conf entry for host "10.0.0.1"', + ReadinessStatus.CONFIG_ERROR, + ), + ( + 'connection failed: FATAL: database "appdb" does not exist', + ReadinessStatus.CONFIG_ERROR, + ), + ( + 'connection failed: FATAL: role "app" does not exist', + ReadinessStatus.CONFIG_ERROR, + ), + # libpq rejects bad values for recognized options client-side. + ( + 'connection is bad: invalid sslmode value: "bogus"', + ReadinessStatus.CONFIG_ERROR, + ), + ( + 'invalid integer value "abc" for connection option "connect_timeout"', + ReadinessStatus.CONFIG_ERROR, + ), + # A DB restart mid-deploy must never read as permanent. + ( + "connection failed: FATAL: the database system is starting up", + ReadinessStatus.UNREACHABLE, + ), + ( + "connection failed: FATAL: sorry, too many clients already", + ReadinessStatus.UNREACHABLE, + ), + ( + 'connection to server at "127.0.0.1", port 5432 failed: Connection refused', + ReadinessStatus.UNREACHABLE, + ), + ( + "[Errno 8] nodename nor servname provided, or not known", + ReadinessStatus.UNREACHABLE, + ), + ("connection timeout expired", ReadinessStatus.UNREACHABLE), + ], +) +def test_connect_time_message_classification(message, expected): + assert _classify_connection_failure(psycopg.OperationalError(message)) is expected + + +def test_query_time_sqlstate_classification(): + # Query-time failures carry a sqlstate on the error class. + assert ( + _classify_connection_failure(psycopg.errors.InvalidPassword("denied")) + is ReadinessStatus.CONFIG_ERROR + ) + # Server shutting down mid-check — retryable. + assert ( + _classify_connection_failure(psycopg.errors.AdminShutdown("terminating")) + is ReadinessStatus.UNREACHABLE + ) + + +def test_improperly_configured_url_is_config_error(monkeypatch): + def raise_improperly_configured(): + raise ImproperlyConfigured("POSTGRES_URL is not set.") + + monkeypatch.setattr(readiness, "_parse_runtime_url", raise_improperly_configured) + + result = check_database_ready() + + assert result.status is ReadinessStatus.CONFIG_ERROR + assert "POSTGRES_URL" in (result.connection_error or "") + + +def test_invalid_url_is_config_error(monkeypatch): + # parse_database_url raises ValueError for e.g. an unsupported scheme + # (mysql://...) — permanent, so it must classify as config error rather + # than crash with a traceback (whose exit 1 would read as retryable). + def raise_value_error(): + raise ValueError("No support for 'mysql'.") + + monkeypatch.setattr(readiness, "_parse_runtime_url", raise_value_error) + + result = check_database_ready() + + assert result.status is ReadinessStatus.CONFIG_ERROR + assert "mysql" in (result.connection_error or "") + + +def test_insufficient_privilege_is_config_error(db, monkeypatch): + # A runtime/management role split that misses the SELECT grant on + # plainmigrations raises InsufficientPrivilege (a ProgrammingError, not + # OperationalError) from the checks — a grant a human must add, so it + # classifies instead of crashing with a traceback. + def raise_insufficient_privilege(conn): + raise psycopg.errors.InsufficientPrivilege( + "permission denied for table plainmigrations" + ) + + monkeypatch.setattr(readiness, "_pending_migrations", raise_insufficient_privilege) + + result = check_database_ready(conn=get_connection()) + + assert result.status is ReadinessStatus.CONFIG_ERROR + assert "permission denied" in (result.connection_error or "") + + +def test_invalid_connection_option_is_config_error(db): + # psycopg rejects an unknown connection option (a typo'd URL query + # param) with ProgrammingError at connect time — config-shaped. + config: DatabaseConfig = { + **get_connection().settings_dict, + "OPTIONS": {"not_a_real_option": "1"}, + } + conn = DatabaseConnection(DirectSource(config)) + + result = check_database_ready(conn=conn) + + assert result.status is ReadinessStatus.CONFIG_ERROR + assert "not_a_real_option" in (result.connection_error or "") diff --git a/plain-postgres/tests/internal/test_wait_for_database.py b/plain-postgres/tests/internal/test_wait_for_database.py new file mode 100644 index 0000000000..714d9c5946 --- /dev/null +++ b/plain-postgres/tests/internal/test_wait_for_database.py @@ -0,0 +1,56 @@ +"""The connectivity wait that fronts the schema commands. + +`cli_wait_for_database()` retries while the database is unreachable and +fails immediately on configuration errors, using the same classification +as `plain postgres ready`. +""" + +from __future__ import annotations + +import click +import pytest + +from plain.postgres.cli.decorators import cli_wait_for_database +from plain.postgres.connection import DatabaseConnection +from plain.postgres.database_url import DatabaseConfig +from plain.postgres.db import _db_conn, get_connection +from plain.postgres.sources import DirectSource + + +def test_returns_immediately_when_already_connected(db): + # The test connection is open (inside the db fixture's transaction) — + # the wait must be a no-op, not a fresh probe. + cli_wait_for_database() + + +def _use_connection(config: DatabaseConfig): + conn = DatabaseConnection(DirectSource(config)) + return _db_conn.set(conn) + + +def test_config_error_fails_immediately(db): + config: DatabaseConfig = { + **get_connection().settings_dict, + "OPTIONS": {"not_a_real_option": "1"}, + } + token = _use_connection(config) + try: + with pytest.raises(click.ClickException, match="not_a_real_option"): + cli_wait_for_database() + finally: + _db_conn.reset(token) + + +def test_unreachable_retries_until_timeout(db, settings): + settings.POSTGRES_WAIT_TIMEOUT = 0.0 # one attempt, then give up + config: DatabaseConfig = { + **get_connection().settings_dict, + "HOST": "127.0.0.1", + "PORT": 59999, + } + token = _use_connection(config) + try: + with pytest.raises(click.ClickException, match="not reachable after 1 attempt"): + cli_wait_for_database() + finally: + _db_conn.reset(token) diff --git a/plain-postgres/tests/public/test_ready.py b/plain-postgres/tests/public/test_ready.py new file mode 100644 index 0000000000..e404637c41 --- /dev/null +++ b/plain-postgres/tests/public/test_ready.py @@ -0,0 +1,165 @@ +"""The readiness contract consumed by serving entrypoints. + +`check_database_ready()` truthfully classifies whether the database can +serve this code, and `plain postgres ready` maps that classification to +exit codes — 0 = ready, 1 = not ready but retryable, 2 = configuration +error a human must fix — with DEBUG downgrading schema gaps to warnings. + +All database staging here runs inside the `db` fixture's transaction — +Postgres DDL is transactional, so dropped tables/columns roll back. +""" + +from __future__ import annotations + +import json + +from app.examples.models.defaults import DefaultsExample +from click.testing import CliRunner + +from plain.postgres.cli import ready as ready_cli +from plain.postgres.db import get_connection +from plain.postgres.migrations.recorder import MigrationRecorder +from plain.postgres.readiness import ( + ReadinessResult, + ReadinessStatus, + check_database_ready, +) + +DATA_MIGRATION = "0019_readiness_data_migration" + + +def test_ready_when_synced(db): + result = check_database_ready(conn=get_connection()) + + assert result.status is ReadinessStatus.READY + assert result.pending_migrations == [] + assert result.pending_data_migrations == [] + assert result.missing_tables == [] + assert result.missing_columns == [] + + +def test_pending_schema_migration_gates(db): + recorder = MigrationRecorder(get_connection()) + # A schema-affecting migration (AddField) missing its applied record — + # what a pod sees when its image is ahead of the database. + recorder.record_unapplied("examples", "0017_random_string_token") + + result = check_database_ready(conn=get_connection()) + + assert result.status is ReadinessStatus.PENDING_MIGRATIONS + assert "examples.0017_random_string_token" in result.pending_migrations + + +def test_pending_data_migration_warns_not_gates(db): + recorder = MigrationRecorder(get_connection()) + # An all-RunPython migration pending — a long backfill mid-flight. + recorder.record_unapplied("examples", DATA_MIGRATION) + + result = check_database_ready(conn=get_connection()) + + assert result.status is ReadinessStatus.READY + assert result.pending_migrations == [] + assert f"examples.{DATA_MIGRATION}" in result.pending_data_migrations + + +def test_missing_column_not_satisfied(db): + conn = get_connection() + table = DefaultsExample.model_options.db_table + with conn.cursor() as cursor: + cursor.execute(f'ALTER TABLE "{table}" DROP COLUMN status') + + result = check_database_ready(conn=conn) + + assert result.status is ReadinessStatus.SCHEMA_NOT_SATISFIED + assert f"{table}.status" in result.missing_columns + assert result.pending_migrations == [] + + +def test_missing_table_not_satisfied(db): + conn = get_connection() + table = DefaultsExample.model_options.db_table + with conn.cursor() as cursor: + cursor.execute(f'DROP TABLE "{table}" CASCADE') + + result = check_database_ready(conn=conn) + + assert result.status is ReadinessStatus.SCHEMA_NOT_SATISFIED + assert table in result.missing_tables + + +def test_fresh_database_is_pending_migrations(db): + conn = get_connection() + # No plainmigrations table at all — a brand-new database a scheduled + # migrate hasn't touched yet. Classified as pending (retryable), not + # as an error. + with conn.cursor() as cursor: + cursor.execute("DROP TABLE plainmigrations") + + result = check_database_ready(conn=conn) + + assert result.status is ReadinessStatus.PENDING_MIGRATIONS + assert result.pending_migrations + + +def _invoke_ready_with_status(monkeypatch, status: ReadinessStatus): + monkeypatch.setattr( + ready_cli, + "check_database_ready", + lambda: ReadinessResult(status=status, connection_error="error detail"), + ) + return CliRunner().invoke(ready_cli.ready, []) + + +def test_cli_exit_codes(monkeypatch, settings): + settings.DEBUG = False + expected = { + ReadinessStatus.READY: 0, + ReadinessStatus.PENDING_MIGRATIONS: 1, + ReadinessStatus.SCHEMA_NOT_SATISFIED: 1, + ReadinessStatus.UNREACHABLE: 1, + ReadinessStatus.CONFIG_ERROR: 2, + } + for status, exit_code in expected.items(): + result = _invoke_ready_with_status(monkeypatch, status) + assert result.exit_code == exit_code, ( + f"{status} should exit {exit_code}, got {result.exit_code}\n{result.output}" + ) + + +def test_cli_exit_codes_debug_warns_past_schema_gaps(monkeypatch, settings): + settings.DEBUG = True + # Mid-development, gaps warn instead of gating — but an unreachable + # database or a config error still fails; nothing serves through those. + expected = { + ReadinessStatus.READY: 0, + ReadinessStatus.PENDING_MIGRATIONS: 0, + ReadinessStatus.SCHEMA_NOT_SATISFIED: 0, + ReadinessStatus.UNREACHABLE: 1, + ReadinessStatus.CONFIG_ERROR: 2, + } + for status, exit_code in expected.items(): + result = _invoke_ready_with_status(monkeypatch, status) + assert result.exit_code == exit_code, ( + f"{status} should exit {exit_code} in DEBUG, got {result.exit_code}\n{result.output}" + ) + + result = _invoke_ready_with_status(monkeypatch, ReadinessStatus.PENDING_MIGRATIONS) + assert "DEBUG" in result.output + + +def test_cli_json_includes_exit_code(monkeypatch, settings): + settings.DEBUG = True + # In DEBUG the process exits 0 while the classification stays truthful — + # the JSON payload carries the exit code so machine consumers can + # reconcile the two. + monkeypatch.setattr( + ready_cli, + "check_database_ready", + lambda: ReadinessResult(status=ReadinessStatus.PENDING_MIGRATIONS), + ) + result = CliRunner().invoke(ready_cli.ready, ["--json"]) + + payload = json.loads(result.output) + assert payload["status"] == "pending-migrations" + assert payload["exit_code"] == 0 + assert result.exit_code == 0 diff --git a/plain/plain/runtime/user_settings.py b/plain/plain/runtime/user_settings.py index 02b61abd30..9bb045be55 100644 --- a/plain/plain/runtime/user_settings.py +++ b/plain/plain/runtime/user_settings.py @@ -374,6 +374,16 @@ def display_value(self) -> str: return repr(self.value) def set_value(self, value: typing.Any, source: str) -> None: + # Accept an int where a float is annotated (PEP 484 numeric tower) — + # `PLAIN_X=30` and `X = 30` are natural spellings for a float + # setting. Stored as float so the value matches the annotation. + # bool is excluded: it's an int subclass but never a number here. + if ( + self.annotation is float + and isinstance(value, int) + and not isinstance(value, bool) + ): + value = float(value) self.check_type(value) self.value = value self.source = source diff --git a/plain/tests/internal/test_setting_int_to_float.py b/plain/tests/internal/test_setting_int_to_float.py new file mode 100644 index 0000000000..43a0f010f0 --- /dev/null +++ b/plain/tests/internal/test_setting_int_to_float.py @@ -0,0 +1,46 @@ +"""Float-annotated settings accept int values (PEP 484 numeric tower). + +`PLAIN_X=30` parses to an int via JSON, and `X = 30` in settings.py is the +natural spelling — both must satisfy a `float` annotation, stored as float. +bool stays rejected (an int subclass, but never a number here). +""" + +from __future__ import annotations + +import pytest + +from plain.exceptions import ImproperlyConfigured +from plain.runtime.user_settings import SettingDefinition + + +def _float_setting() -> SettingDefinition: + return SettingDefinition( + name="EXAMPLE_TIMEOUT", default_value=60.0, annotation=float + ) + + +def test_int_value_coerced_to_float(): + definition = _float_setting() + definition.set_value(30, source="env") + assert definition.value == 30.0 + assert isinstance(definition.value, float) + + +def test_float_value_unchanged(): + definition = _float_setting() + definition.set_value(2.5, source="explicit") + assert definition.value == 2.5 + + +def test_bool_value_still_rejected(): + definition = _float_setting() + with pytest.raises(ImproperlyConfigured): + definition.set_value(True, source="explicit") + + +def test_int_annotation_still_rejects_float(): + definition = SettingDefinition( + name="EXAMPLE_COUNT", default_value=4, annotation=int + ) + with pytest.raises(ImproperlyConfigured): + definition.set_value(4.5, source="explicit")