fix(db): survive a database restart and a not-yet-ready database
Two connection-recovery gaps, both surfaced by a Postgres restart that left the app throwing tracebacks while the DB itself was healthy. pool_pre_ping + pool_recycle on the app engine [#2626]: when the database restarts, every connection already in the pool is dead at the socket level. SQLAlchemy only discovered that by failing a real query, so the first operation after a restart errored out on whatever triggered it. pre_ping checks liveness on checkout and swaps the dead connection transparently; pool_recycle caps connection age so a socket stranded by a NAT/conntrack timeout or a Docker network rebuild is retired on a timer instead. wait_for_database() gate at startup [#2627]: create_app touches the DB synchronously (migrations, secret re-encryption, settings load) and assumed it was both resolvable and accepting connections on the first try. Neither holds after a host reboot (Docker DNS not yet serving `db` -> gaierror -2) or an unclean shutdown (Postgres still replaying WAL -> "not yet accepting connections"). Both are transient, so retry with capped backoff behind one gate ahead of the first DB touch. Credential and missing-database errors are classified by SQLSTATE and still fail immediately -- waiting cannot fix those. Budget is bootstrap-configurable (STEWARD_DB_CONNECT_TIMEOUT / database.connect_timeout, default 60s) since it governs reaching the DB and so cannot live in the DB-backed settings. Tests drive the retry loop off a fake clock, so backoff and timeout behaviour are deterministic rather than wall-clock dependent. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -13,6 +13,14 @@
|
|||||||
database:
|
database:
|
||||||
url: "postgresql+asyncpg://steward:password@localhost/steward"
|
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.
|
# Optional: override the auto-generated secret key.
|
||||||
# If not set, a key is auto-generated on first run and saved to /data/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"
|
# secret_key: "change-me-to-a-random-string"
|
||||||
|
|||||||
+15
-1
@@ -5,7 +5,7 @@ import logging
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from quart import Quart, render_template
|
from quart import Quart, render_template
|
||||||
from .config import load_bootstrap
|
from .config import load_bootstrap
|
||||||
from .database import init_db
|
from .database import init_db, DB_CONNECT_TIMEOUT_SECONDS
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -23,6 +23,7 @@ def create_app(
|
|||||||
bootstrap = {
|
bootstrap = {
|
||||||
"database_url": "postgresql+asyncpg://test/test",
|
"database_url": "postgresql+asyncpg://test/test",
|
||||||
"secret_key": "test-secret-key",
|
"secret_key": "test-secret-key",
|
||||||
|
"db_connect_timeout": DB_CONNECT_TIMEOUT_SECONDS,
|
||||||
"plugin_dirs": ["plugins"],
|
"plugin_dirs": ["plugins"],
|
||||||
"plugin_install_dir": "plugins",
|
"plugin_install_dir": "plugins",
|
||||||
}
|
}
|
||||||
@@ -42,6 +43,19 @@ def create_app(
|
|||||||
from unittest.mock import MagicMock
|
from unittest.mock import MagicMock
|
||||||
app.db_sessionmaker = 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) ──────────────────
|
# ── 3. Core migrations only (creates app_settings table) ──────────────────
|
||||||
if not testing:
|
if not testing:
|
||||||
from .core.migration_runner import run_core_migrations
|
from .core.migration_runner import run_core_migrations
|
||||||
|
|||||||
@@ -49,6 +49,8 @@ def load_bootstrap(config_path: Path | str | None = None) -> dict[str, Any]:
|
|||||||
|
|
||||||
secret_key = _resolve_secret_key(raw)
|
secret_key = _resolve_secret_key(raw)
|
||||||
|
|
||||||
|
db_connect_timeout = _resolve_db_connect_timeout(raw)
|
||||||
|
|
||||||
# Plugin discovery spans two roots (see load_plugins / migration_runner):
|
# Plugin discovery spans two roots (see load_plugins / migration_runner):
|
||||||
# • bundled — first-party plugins shipped inside the image at repo-root
|
# • bundled — first-party plugins shipped inside the image at repo-root
|
||||||
# `plugins/`; they version atomically with core and are read-only at runtime.
|
# `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 {
|
return {
|
||||||
"database_url": database_url,
|
"database_url": database_url,
|
||||||
"secret_key": secret_key,
|
"secret_key": secret_key,
|
||||||
|
"db_connect_timeout": db_connect_timeout,
|
||||||
"plugin_dirs": plugin_dirs,
|
"plugin_dirs": plugin_dirs,
|
||||||
# Installs/downloads target the external (writable, persistent) dir.
|
# Installs/downloads target the external (writable, persistent) dir.
|
||||||
"plugin_install_dir": external_plugin_dir or bundled_plugin_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:
|
def _resolve_secret_key(raw: dict) -> str:
|
||||||
"""Resolve secret_key: env var → file → auto-generate.
|
"""Resolve secret_key: env var → file → auto-generate.
|
||||||
|
|
||||||
|
|||||||
+142
-1
@@ -1,10 +1,39 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
import asyncio
|
||||||
|
import logging
|
||||||
|
import time
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
from sqlalchemy import text
|
||||||
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
|
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
|
||||||
|
from sqlalchemy.pool import NullPool
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from quart import Quart
|
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:
|
def init_db(app: "Quart") -> None:
|
||||||
"""Create async engine and attach db_sessionmaker to app.
|
"""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.
|
Does not create tables — Alembic handles migrations.
|
||||||
"""
|
"""
|
||||||
db_url: str = app.config["DATABASE_URL"]
|
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(
|
app.db_sessionmaker: async_sessionmaker[AsyncSession] = async_sessionmaker(
|
||||||
engine, expire_on_commit=False
|
engine, expire_on_commit=False
|
||||||
)
|
)
|
||||||
app._db_engine = engine
|
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()
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -65,3 +65,50 @@ def test_secret_key_unpersistable_raises_instead_of_ephemeral(tmp_path, monkeypa
|
|||||||
monkeypatch.delenv("STEWARD_SECRET_KEY", raising=False)
|
monkeypatch.delenv("STEWARD_SECRET_KEY", raising=False)
|
||||||
with pytest.raises(RuntimeError, match="could not persist"):
|
with pytest.raises(RuntimeError, match="could not persist"):
|
||||||
_resolve_secret_key({})
|
_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
|
||||||
|
|||||||
Reference in New Issue
Block a user