diff --git a/config.example.yaml b/config.example.yaml index e165f7d..64b96e5 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -13,6 +13,14 @@ database: url: "postgresql+asyncpg://steward:password@localhost/steward" + # Optional: seconds to wait at startup for the database to become reachable + # before giving up (default: 60). Steward retries with backoff rather than + # crashing when the DB is merely slow to arrive — replaying WAL after an + # unclean shutdown, or container DNS not yet resolving after a host reboot. + # Raise it if your database is routinely slower than this to accept + # connections. Env var: STEWARD_DB_CONNECT_TIMEOUT + # connect_timeout: 60 + # Optional: override the auto-generated secret key. # If not set, a key is auto-generated on first run and saved to /data/secret.key. # secret_key: "change-me-to-a-random-string" diff --git a/steward/app.py b/steward/app.py index 3341651..14d822f 100644 --- a/steward/app.py +++ b/steward/app.py @@ -5,7 +5,7 @@ import logging from pathlib import Path from quart import Quart, render_template from .config import load_bootstrap -from .database import init_db +from .database import init_db, DB_CONNECT_TIMEOUT_SECONDS logger = logging.getLogger(__name__) @@ -23,6 +23,7 @@ def create_app( bootstrap = { "database_url": "postgresql+asyncpg://test/test", "secret_key": "test-secret-key", + "db_connect_timeout": DB_CONNECT_TIMEOUT_SECONDS, "plugin_dirs": ["plugins"], "plugin_install_dir": "plugins", } @@ -42,6 +43,19 @@ def create_app( from unittest.mock import MagicMock app.db_sessionmaker = MagicMock() + # ── 2b. Block until the database is actually reachable ──────────────────── + # Everything from here down touches the DB synchronously (migrations, + # secret re-encryption, settings load). Gate all of it behind one readiness + # check so a DB that is merely slow to come up — WAL recovery after an + # unclean shutdown, or Docker DNS not yet serving the `db` name after a host + # reboot — is waited out instead of crashing the container on a traceback. + if not testing: + from .database import wait_for_database + wait_for_database( + app.config["DATABASE_URL"], + timeout_seconds=bootstrap["db_connect_timeout"], + ) + # ── 3. Core migrations only (creates app_settings table) ────────────────── if not testing: from .core.migration_runner import run_core_migrations diff --git a/steward/config.py b/steward/config.py index 678359e..b177b7f 100644 --- a/steward/config.py +++ b/steward/config.py @@ -49,6 +49,8 @@ def load_bootstrap(config_path: Path | str | None = None) -> dict[str, Any]: secret_key = _resolve_secret_key(raw) + db_connect_timeout = _resolve_db_connect_timeout(raw) + # Plugin discovery spans two roots (see load_plugins / migration_runner): # • bundled — first-party plugins shipped inside the image at repo-root # `plugins/`; they version atomically with core and are read-only at runtime. @@ -68,12 +70,47 @@ def load_bootstrap(config_path: Path | str | None = None) -> dict[str, Any]: return { "database_url": database_url, "secret_key": secret_key, + "db_connect_timeout": db_connect_timeout, "plugin_dirs": plugin_dirs, # Installs/downloads target the external (writable, persistent) dir. "plugin_install_dir": external_plugin_dir or bundled_plugin_dir, } +def _resolve_db_connect_timeout(raw: dict) -> float: + """How long to wait for the database at startup, in seconds. + + Bootstrap-only by necessity: this governs reaching the DB, so it cannot + itself be read from the DB like the rest of Steward's settings. + + A deployment whose database is slower to come up than the default (a large + cluster replaying WAL, a remote DB behind a link that takes a while) can + raise it rather than crash-looping the container. + """ + from .database import DB_CONNECT_TIMEOUT_SECONDS + + value = _env("DB_CONNECT_TIMEOUT") or raw.get("database", {}).get("connect_timeout") + if value is None or value == "": + return DB_CONNECT_TIMEOUT_SECONDS + try: + parsed = float(value) + except (TypeError, ValueError): + logger.warning( + "Invalid database connect timeout %r — using default %.0fs", + value, DB_CONNECT_TIMEOUT_SECONDS, + ) + return DB_CONNECT_TIMEOUT_SECONDS + if parsed <= 0: + # 0/negative would mean "never wait", which is the broken behaviour this + # setting exists to fix — treat it as a mistake, not as an opt-out. + logger.warning( + "Database connect timeout %.0fs is not positive — using default %.0fs", + parsed, DB_CONNECT_TIMEOUT_SECONDS, + ) + return DB_CONNECT_TIMEOUT_SECONDS + return parsed + + def _resolve_secret_key(raw: dict) -> str: """Resolve secret_key: env var → file → auto-generate. diff --git a/steward/database.py b/steward/database.py index b04ce15..af24f8f 100644 --- a/steward/database.py +++ b/steward/database.py @@ -1,10 +1,39 @@ from __future__ import annotations +import asyncio +import logging +import time from typing import TYPE_CHECKING +from sqlalchemy import text from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession +from sqlalchemy.pool import NullPool if TYPE_CHECKING: from quart import Quart +logger = logging.getLogger(__name__) + +# Recycle pooled connections after 30 minutes. Nothing in Postgres closes an +# idle connection by default, but the path to it is not always durable: NAT and +# conntrack tables drop idle flows, and a Docker network rebuild silently +# strands existing sockets. Capping connection age means a stranded connection +# is retired on a timer instead of surfacing as a failed query later. +POOL_RECYCLE_SECONDS = 1800 + +# Total budget for the database to become reachable at startup, and the backoff +# ceiling between attempts. 60s comfortably covers Postgres WAL recovery after +# an unclean shutdown (observed ~7s) plus Docker DNS coming up on a host reboot. +DB_CONNECT_TIMEOUT_SECONDS = 60.0 +_BACKOFF_INITIAL_SECONDS = 0.5 +_BACKOFF_MAX_SECONDS = 5.0 + +# Postgres SQLSTATEs that will never resolve by waiting — retrying these just +# delays a clear error behind a full timeout budget. +_FATAL_SQLSTATES = { + "28P01", # invalid_password + "28000", # invalid_authorization_specification + "3D000", # invalid_catalog_name — database does not exist +} + def init_db(app: "Quart") -> None: """Create async engine and attach db_sessionmaker to app. @@ -13,8 +42,120 @@ def init_db(app: "Quart") -> None: Does not create tables — Alembic handles migrations. """ db_url: str = app.config["DATABASE_URL"] - engine = create_async_engine(db_url, echo=False) + engine = create_async_engine( + db_url, + echo=False, + # Check a pooled connection is still alive before handing it out. When + # the database restarts, every connection already in the pool is dead + # at the socket level; without this, SQLAlchemy only discovers that by + # failing a real query, so the first operation after a DB restart + # errors out on whatever triggered it (a request, a scheduled poll). + # The check is a cheap round-trip and it makes a DB restart invisible. + pool_pre_ping=True, + pool_recycle=POOL_RECYCLE_SECONDS, + ) app.db_sessionmaker: async_sessionmaker[AsyncSession] = async_sessionmaker( engine, expire_on_commit=False ) app._db_engine = engine + + +def _fatal_sqlstate(exc: BaseException) -> str | None: + """Return the SQLSTATE if this error chain carries a non-retryable one.""" + seen: set[int] = set() + cur: BaseException | None = exc + while cur is not None and id(cur) not in seen: + seen.add(id(cur)) + sqlstate = getattr(cur, "sqlstate", None) + if sqlstate in _FATAL_SQLSTATES: + return sqlstate + cur = cur.__cause__ or cur.__context__ + return None + + +def _describe(exc: BaseException) -> str: + """Innermost cause, which is the part that says what actually went wrong. + + SQLAlchemy wraps DBAPI errors several layers deep; the outer message is + boilerplate, so surface the root for the waiting-for-database log line. + """ + cur: BaseException = exc + seen: set[int] = {id(cur)} + while True: + nxt = cur.__cause__ or cur.__context__ + if nxt is None or id(nxt) in seen: + return f"{type(cur).__name__}: {cur}" + seen.add(id(nxt)) + cur = nxt + + +def wait_for_database( + db_url: str, + timeout_seconds: float = DB_CONNECT_TIMEOUT_SECONDS, +) -> None: + """Block until the database accepts a connection, or raise after the budget. + + Steward otherwise assumes the database is both resolvable and *ready* the + first time it asks, which is false in two ordinary situations: + + • Host reboot / full stack restart — the app container can try to resolve + the `db` service name before Docker's embedded DNS has the record, + giving `gaierror -2 Name or service not known`. + • Unclean shutdown — Postgres is listening but still replaying WAL, and + refuses connections with "the database system is not yet accepting + connections" until recovery reaches a consistent state. + + compose's `depends_on: service_healthy` covers ordering on a clean `up`, + but not either of the above. Both are transient and self-healing, so retry + rather than dumping a traceback and dying. Credential and missing-database + errors are NOT transient and fail immediately. + """ + deadline = time.monotonic() + timeout_seconds + delay = _BACKOFF_INITIAL_SECONDS + attempt = 0 + last_exc: BaseException | None = None + + while True: + attempt += 1 + try: + asyncio.run(_probe(db_url)) + except Exception as exc: # broad by design — classified just below + sqlstate = _fatal_sqlstate(exc) + if sqlstate is not None: + # Wrong password / missing database: waiting cannot fix it. + raise + last_exc = exc + remaining = deadline - time.monotonic() + if remaining <= 0: + break + logger.warning( + "Database not ready (attempt %d, %.0fs budget left): %s — " + "retrying in %.1fs", + attempt, remaining, _describe(exc), min(delay, remaining), + ) + time.sleep(min(delay, remaining)) + delay = min(delay * 2, _BACKOFF_MAX_SECONDS) + else: + if attempt > 1: + logger.info("Database ready after %d attempt(s)", attempt) + return + + detail = f" Last error: {_describe(last_exc)}" if last_exc is not None else "" + raise RuntimeError( + f"Database did not become available within {timeout_seconds:.0f}s " + f"({attempt} attempts).{detail}" + ) from last_exc + + +async def _probe(db_url: str) -> None: + """Open one throwaway connection and round-trip a trivial query. + + Uses its own engine with NullPool: this runs before the app engine exists, + and a probe connection must never be left in a pool for real work to reuse. + """ + engine = create_async_engine(db_url, echo=False, poolclass=NullPool) + try: + async with engine.connect() as conn: + await conn.execute(text("SELECT 1")) + finally: + await engine.dispose() diff --git a/tests/core/test_db_readiness.py b/tests/core/test_db_readiness.py new file mode 100644 index 0000000..cedb64c --- /dev/null +++ b/tests/core/test_db_readiness.py @@ -0,0 +1,215 @@ +"""Startup DB readiness retry + connection-pool resilience. + +Covers the two recovery gaps that let a database restart take the app down: +the engine handing out stale pooled connections, and startup assuming the DB +is reachable and ready the instant it asks. +""" +import types +import pytest + +from steward import database +from steward.database import ( + POOL_RECYCLE_SECONDS, + _describe, + _fatal_sqlstate, + init_db, + wait_for_database, +) + + +class _PGError(Exception): + """Stand-in for an asyncpg error, which carries a SQLSTATE attribute.""" + + def __init__(self, message: str, sqlstate: str | None = None): + super().__init__(message) + self.sqlstate = sqlstate + + +@pytest.fixture +def no_sleep(monkeypatch): + """Drive the retry budget off a fake clock instead of wall time. + + sleep() records its duration and advances the clock by exactly that much, + so backoff and timeout behaviour are deterministic rather than dependent on + how fast the test machine happens to run. + """ + slept: list[float] = [] + clock = {"now": 0.0} + + def fake_sleep(seconds: float) -> None: + slept.append(seconds) + clock["now"] += seconds + + monkeypatch.setattr(database.time, "sleep", fake_sleep) + monkeypatch.setattr(database.time, "monotonic", lambda: clock["now"]) + return slept + + +def _probe_raising(*errors, then_succeed: bool = True): + """Async probe stub raising the given errors in order. + + Once the list is exhausted it succeeds, unless then_succeed is False — in + which case it keeps raising the last error forever (a DB that never comes + back). + """ + calls = {"n": 0} + + async def probe(db_url): + i = calls["n"] + calls["n"] += 1 + if i < len(errors): + raise errors[i] + if not then_succeed: + raise errors[-1] + + probe.calls = calls + return probe + + +# ── engine pool configuration (issue #2626) ────────────────────────────────── + + +def test_engine_enables_pre_ping_and_recycle(): + app = types.SimpleNamespace( + config={"DATABASE_URL": "postgresql+asyncpg://u:p@localhost/db"} + ) + init_db(app) + + # _pre_ping / _recycle are SQLAlchemy pool internals; there is no public + # accessor, and these are exactly the settings a DB restart depends on. + pool = app._db_engine.sync_engine.pool + assert pool._pre_ping is True + assert pool._recycle == POOL_RECYCLE_SECONDS + + +def test_pool_recycle_is_positive(): + # A non-positive recycle disables age-based retirement entirely. + assert POOL_RECYCLE_SECONDS > 0 + + +# ── retry classification ───────────────────────────────────────────────────── + + +def test_fatal_sqlstate_detects_bad_password(): + exc = _PGError("password authentication failed", sqlstate="28P01") + assert _fatal_sqlstate(exc) == "28P01" + + +def test_fatal_sqlstate_finds_sqlstate_through_cause_chain(): + inner = _PGError("database does not exist", sqlstate="3D000") + outer = RuntimeError("wrapped by sqlalchemy") + outer.__cause__ = inner + assert _fatal_sqlstate(outer) == "3D000" + + +def test_transient_errors_are_not_fatal(): + assert _fatal_sqlstate(OSError(-2, "Name or service not known")) is None + # "the database system is not yet accepting connections" — resolves on its own. + assert _fatal_sqlstate(_PGError("not yet accepting", sqlstate="57P03")) is None + + +def test_describe_unwraps_to_innermost_cause(): + inner = OSError("Name or service not known") + outer = RuntimeError("sqlalchemy boilerplate") + outer.__cause__ = inner + assert "Name or service not known" in _describe(outer) + + +def test_describe_survives_self_referential_cause(): + exc = RuntimeError("loop") + exc.__cause__ = exc + assert "loop" in _describe(exc) + + +# ── wait_for_database (issue #2627) ────────────────────────────────────────── + + +def test_returns_immediately_when_db_is_up(monkeypatch, no_sleep): + probe = _probe_raising() + monkeypatch.setattr(database, "_probe", probe) + + wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=10) + + assert probe.calls["n"] == 1 + assert no_sleep == [] + + +def test_retries_dns_failure_then_succeeds(monkeypatch, no_sleep): + # The reported failure: gaierror -2 while Docker DNS is not yet serving `db`. + probe = _probe_raising( + OSError(-2, "Name or service not known"), + OSError(-2, "Name or service not known"), + ) + monkeypatch.setattr(database, "_probe", probe) + + wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=30) + + assert probe.calls["n"] == 3 + assert len(no_sleep) == 2 + + +def test_retries_while_postgres_is_still_recovering(monkeypatch, no_sleep): + probe = _probe_raising( + _PGError("the database system is not yet accepting connections", "57P03"), + ) + monkeypatch.setattr(database, "_probe", probe) + + wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=30) + + assert probe.calls["n"] == 2 + + +def test_backoff_grows_between_attempts(monkeypatch, no_sleep): + probe = _probe_raising(*[OSError("refused")] * 4) + monkeypatch.setattr(database, "_probe", probe) + + wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=60) + + assert no_sleep == sorted(no_sleep), "delays should be non-decreasing" + assert no_sleep[-1] > no_sleep[0], "backoff should grow, not stay flat" + + +def test_backoff_is_capped(monkeypatch, no_sleep): + probe = _probe_raising(*[OSError("refused")] * 12) + monkeypatch.setattr(database, "_probe", probe) + + wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=600) + + assert max(no_sleep) <= database._BACKOFF_MAX_SECONDS + + +def test_bad_credentials_fail_immediately_without_retrying(monkeypatch, no_sleep): + probe = _probe_raising(_PGError("password authentication failed", "28P01")) + monkeypatch.setattr(database, "_probe", probe) + + with pytest.raises(_PGError): + wait_for_database("postgresql+asyncpg://u:bad@db/steward", timeout_seconds=30) + + # Waiting cannot fix a wrong password — don't burn the whole budget on it. + assert probe.calls["n"] == 1 + assert no_sleep == [] + + +def test_missing_database_fails_immediately(monkeypatch, no_sleep): + probe = _probe_raising(_PGError("database does not exist", "3D000")) + monkeypatch.setattr(database, "_probe", probe) + + with pytest.raises(_PGError): + wait_for_database("postgresql+asyncpg://u:p@db/nope", timeout_seconds=30) + + assert probe.calls["n"] == 1 + + +def test_gives_up_after_budget_with_actionable_message(monkeypatch, no_sleep): + probe = _probe_raising( + OSError("Name or service not known"), then_succeed=False + ) + monkeypatch.setattr(database, "_probe", probe) + + with pytest.raises(RuntimeError) as excinfo: + wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=2) + + message = str(excinfo.value) + assert "did not become available" in message + # The operator needs the underlying cause, not just "timed out". + assert "Name or service not known" in message diff --git a/tests/test_config.py b/tests/test_config.py index 93d482a..6d98767 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -65,3 +65,50 @@ def test_secret_key_unpersistable_raises_instead_of_ephemeral(tmp_path, monkeypa monkeypatch.delenv("STEWARD_SECRET_KEY", raising=False) with pytest.raises(RuntimeError, match="could not persist"): _resolve_secret_key({}) + + +# ── database connect timeout (bootstrap-only: it governs reaching the DB) ──── + + +def test_db_connect_timeout_defaults_when_unset(tmp_path, monkeypatch): + from steward.database import DB_CONNECT_TIMEOUT_SECONDS + cfg_file = tmp_path / "config.yaml" + cfg_file.write_text("database:\n url: x\nsecret_key: s\n") + monkeypatch.delenv("STEWARD_DB_CONNECT_TIMEOUT", raising=False) + cfg = load_bootstrap(cfg_file) + assert cfg["db_connect_timeout"] == DB_CONNECT_TIMEOUT_SECONDS + + +def test_db_connect_timeout_from_yaml(tmp_path, monkeypatch): + cfg_file = tmp_path / "config.yaml" + cfg_file.write_text( + "database:\n url: x\n connect_timeout: 120\nsecret_key: s\n") + monkeypatch.delenv("STEWARD_DB_CONNECT_TIMEOUT", raising=False) + assert load_bootstrap(cfg_file)["db_connect_timeout"] == 120.0 + + +def test_db_connect_timeout_env_overrides_yaml(tmp_path, monkeypatch): + cfg_file = tmp_path / "config.yaml" + cfg_file.write_text( + "database:\n url: x\n connect_timeout: 120\nsecret_key: s\n") + monkeypatch.setenv("STEWARD_DB_CONNECT_TIMEOUT", "5") + assert load_bootstrap(cfg_file)["db_connect_timeout"] == 5.0 + + +def test_db_connect_timeout_garbage_falls_back_to_default(tmp_path, monkeypatch): + from steward.database import DB_CONNECT_TIMEOUT_SECONDS + cfg_file = tmp_path / "config.yaml" + cfg_file.write_text("database:\n url: x\nsecret_key: s\n") + monkeypatch.setenv("STEWARD_DB_CONNECT_TIMEOUT", "not-a-number") + cfg = load_bootstrap(cfg_file) + assert cfg["db_connect_timeout"] == DB_CONNECT_TIMEOUT_SECONDS + + +def test_db_connect_timeout_non_positive_falls_back_to_default(tmp_path, monkeypatch): + # 0 would mean "never wait", reinstating the crash this setting exists to fix. + from steward.database import DB_CONNECT_TIMEOUT_SECONDS + cfg_file = tmp_path / "config.yaml" + cfg_file.write_text("database:\n url: x\nsecret_key: s\n") + monkeypatch.setenv("STEWARD_DB_CONNECT_TIMEOUT", "0") + cfg = load_bootstrap(cfg_file) + assert cfg["db_connect_timeout"] == DB_CONNECT_TIMEOUT_SECONDS