diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 4b010629..c4873020 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -131,7 +131,15 @@ jobs: permissions: contents: write steps: + # fetch-depth: 0, not the default shallow clone. `git push` proves the + # fast-forward CLIENT-side: with depth 1 the runner does not have the + # commit `release` currently points at, so every push after the one that + # CREATED the ref was rejected with "fetch first" — release sat at the + # first published commit (#360) while three merges deployed past it, + # and the box's pull path had nothing new to converge on. - uses: actions/checkout@v5 + with: + fetch-depth: 0 - name: Fast-forward release to this commit run: | @@ -200,7 +208,14 @@ jobs: cp scripts/vps_apply.sh /tmp/deploy_remote.sh ssh_deploy() { - timeout 900 ssh \ + # 1800s, matched to the pull unit's TimeoutStartSec. 900 was fatal + # in a way verify() cannot see: the pre-migration backup alone + # takes ~11 minutes on the 4GB box, so the session was killed + # MID-APPLY — migrations never ran, services never restarted — + # and verify() then blessed the still-running OLD deployment as + # green. A timeout kill is the one failure mode where "the app is + # healthy" says nothing about "the deploy happened". + timeout 1800 ssh \ -i ~/.ssh/deploy_key -p "$SSH_PORT" \ -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \ -o ConnectTimeout=30 -o ServerAliveInterval=15 \ diff --git a/ai-company-brain/specs/deploy_delivery_path.md b/ai-company-brain/specs/deploy_delivery_path.md index 16921942..e5942efa 100644 --- a/ai-company-brain/specs/deploy_delivery_path.md +++ b/ai-company-brain/specs/deploy_delivery_path.md @@ -310,6 +310,35 @@ run this script as `ssh acb@host 'bash -s'`, and the applied script calls `sudo` passwordless sudo. So `User=acb` is not a workaround; running it as root was the deviation. +⚠️ **Corrections exist on paper until they are installed.** The corrected unit +below sat in this spec while the box kept running the `User=root` version — the +poller failed every five minutes from its first tick until 2026-08-06, when the +WS-26 activation found it. `.claude/` and `/etc/systemd/system/` share the same +failure mode: not in the repo, so no PR can fix them; every correction here is +also a hand-carried box change. + +**Defect 3 (2026-08-06) — the SHA can be current while the APPLY never +happened.** The poller skips when `HEAD == origin/release`. But the push path's +`git reset --hard` moves HEAD *before* migrations and restarts — so when the +push-path SSH session was killed mid-apply (Defect 4b), the tree said `8d83ca10` +while the DB had no migration 144/145 and both services still ran old code. The +poller then reported `already current` and stood down: both delivery paths +converged on believing a deploy that had not happened. `--force` is the manual +escape; the durable fix would be gating the skip on the `/var/lib/acb` +last-success marker (recorded at the END of an apply), not on git state. + +**Defect 4 (2026-08-06) — two `deploy.yml` failures, one green run.** +(a) `publish-release` used the default depth-1 checkout; `git push` proves +fast-forward client-side, so every publish after the ref-CREATING one was +rejected with "fetch first" — `release` sat at #360 for three merges. Fixed: +`fetch-depth: 0`. (b) `ssh_deploy`'s `timeout 900` is shorter than the +pre-migration backup alone (~11 min on the 4GB box); the session was killed +mid-apply and `verify()` blessed the still-healthy OLD deployment. Fixed: +`timeout 1800`, matched to the pull unit's `TimeoutStartSec`. Residual hole, +accepted and named: health-verify cannot distinguish "deploy succeeded" from +"deploy never finished but yesterday's app is healthy" — only an +identity-bearing health signal (commit SHA in `/health`) closes it. + ```ini # /etc/systemd/system/acb-pull.service [Unit] diff --git a/apps/services/email_ingestion/email_ingestion/inbound.py b/apps/services/email_ingestion/email_ingestion/inbound.py index 05779925..1168d08c 100644 --- a/apps/services/email_ingestion/email_ingestion/inbound.py +++ b/apps/services/email_ingestion/email_ingestion/inbound.py @@ -271,7 +271,7 @@ async def _persist_message(msg: EmailMessage) -> None: from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine engine = create_async_engine( - db_url, echo=False, connect_args={"timeout": _connect_timeout()} + db_url, echo=False, connect_args=_connect_args() ) session_factory = async_sessionmaker(engine, expire_on_commit=False) @@ -322,6 +322,22 @@ def _connect_timeout() -> int: return 10 +def _connect_args() -> dict: + """Connect args for this module's engines. + + :func:`_connect_timeout` bounds getting IN; the server-side + ``idle_in_transaction_session_timeout`` bounds staying in, so a wedged + persist cannot hold its locks for the life of the process. Mirrors + ``scheduler._connect_args`` and ``gateway.db.engine_connect_args`` — see the + 2026-08-06 write-up for why an unbounded one took the app down.""" + return { + "timeout": _connect_timeout(), + "server_settings": { + "idle_in_transaction_session_timeout": "600000", # 10 min + }, + } + + # -- Lifecycle management ----------------------------------------------------- diff --git a/apps/services/email_ingestion/email_ingestion/scheduler.py b/apps/services/email_ingestion/email_ingestion/scheduler.py index 481de7eb..46cc2223 100644 --- a/apps/services/email_ingestion/email_ingestion/scheduler.py +++ b/apps/services/email_ingestion/email_ingestion/scheduler.py @@ -87,6 +87,24 @@ def _connect_timeout() -> int: return 10 +def _connect_args() -> dict: + """Connect args for every engine in this service. + + :func:`_connect_timeout` bounds getting IN; the server-side + ``idle_in_transaction_session_timeout`` bounds staying in. A sync tick holds + an open session across provider calls, so a tick that wedges would otherwise + hold its row locks for as long as the process lives — which on 2026-08-06 + (gateway side, same shape) queued a migration's ALTER TABLE behind it and + took the whole app down. Mirrors ``gateway.db.engine_connect_args``; the two + are separate packages and cannot share the constant.""" + return { + "timeout": _connect_timeout(), + "server_settings": { + "idle_in_transaction_session_timeout": "600000", # 10 min + }, + } + + def _next_backoff(current: int, interval: int, *, failed: bool) -> int: """The next sleep length for a sync loop. @@ -141,7 +159,7 @@ async def _sync_account( # the session releases it, so the missing dispose no longer leaks. engine = create_async_engine( db_url, echo=False, poolclass=NullPool, - connect_args={"timeout": _connect_timeout()}, + connect_args=_connect_args(), ) session_factory = async_sessionmaker(engine, expire_on_commit=False) @@ -525,7 +543,7 @@ async def _get_account_sync_interval(account_id: str) -> int | None: """Read the current sync_interval_secs for an account.""" db_url = _get_db_url() engine = create_async_engine( - db_url, echo=False, connect_args={"timeout": _connect_timeout()} + db_url, echo=False, connect_args=_connect_args() ) session_factory = async_sessionmaker(engine, expire_on_commit=False) try: @@ -558,7 +576,7 @@ async def start_background_sync() -> dict[str, int]: db_url = _get_db_url() engine = create_async_engine( - db_url, echo=False, connect_args={"timeout": _connect_timeout()} + db_url, echo=False, connect_args=_connect_args() ) session_factory = async_sessionmaker(engine, expire_on_commit=False) diff --git a/apps/services/gateway/gateway/db.py b/apps/services/gateway/gateway/db.py index b2d92aeb..86be70d5 100644 --- a/apps/services/gateway/gateway/db.py +++ b/apps/services/gateway/gateway/db.py @@ -53,20 +53,60 @@ def async_database_url() -> str: return db_url +#: Ceiling on how long one of OUR sessions may sit `idle in transaction`, in ms. +#: +#: This is a lock-release deadline, not a performance knob. SQLAlchemy's +#: ``AsyncSession`` opens a transaction on first ``execute()`` and holds it until +#: commit/rollback/close, so a handler that reads a row and then awaits a slow +#: network call is `idle in transaction` — holding an ACCESS SHARE lock — for the +#: whole call. That is normal and fine. What is not fine is an await that never +#: returns: on 2026-08-06 a hung LLM call pinned one such transaction for 14h44m, +#: a migration's ``ALTER TABLE`` queued behind its lock, and because Postgres's +#: lock queue is FIFO every later reader of that table queued behind the *waiting* +#: ALTER. Sending mail stopped, and the pool drained behind the blocked readers. +#: +#: ⚠️ MUST stay comfortably above the LLM wall-clock worst case, because the email +#: automation package legitimately awaits completions with a session open. +#: ``acb_llm.client`` bounds one call at 3 attempts x 90s + 6s backoff ≈ 276s. At +#: 600s a genuine retrying completion can never trip this, while a hang is capped +#: at ten minutes instead of unbounded. Raise ``LLM_REQUEST_TIMEOUT_SECS`` and you +#: must raise this too — that coupling is the whole reason both numbers are +#: written down next to their reasoning. +_IDLE_IN_TXN_TIMEOUT_MS = "600000" # 10 minutes + + +def engine_connect_args() -> dict[str, Any]: + """Driver-level connect args every gateway engine should share. + + Two bounds, both about failing instead of hanging: + + * ``timeout`` — asyncpg's CONNECT-phase ceiling, so a slow or unreachable DB + fails fast rather than stalling request handlers. + * ``idle_in_transaction_session_timeout`` — the server-side deadline above. + Set through asyncpg's ``server_settings`` so it rides the connection's + startup packet and applies to every session from this pool, with no + migration and no ``ALTER ROLE``. Scoping it to the app's own connections is + deliberate: ``pg_dump`` and the migration runner connect as the same role + and must NOT inherit an app-tuned deadline. + """ + return { + "timeout": get_settings().db_connect_timeout, + "server_settings": { + "idle_in_transaction_session_timeout": _IDLE_IN_TXN_TIMEOUT_MS, + }, + } + + def get_engine() -> Any: """The shared pooled async engine, created on first use.""" global _ENGINE if _ENGINE is None: from sqlalchemy.ext.asyncio import create_async_engine - settings = get_settings() _ENGINE = create_async_engine( async_database_url(), echo=False, pool_pre_ping=True, pool_size=10, max_overflow=20, pool_recycle=1800, - # Bound the CONNECT phase (asyncpg's `timeout`) so a slow or - # unreachable DB fails fast instead of stalling request handlers — - # same ceiling as acb_graph's engine. - connect_args={"timeout": settings.db_connect_timeout}, + connect_args=engine_connect_args(), ) return _ENGINE diff --git a/apps/services/gateway/gateway/routes/admin/_common.py b/apps/services/gateway/gateway/routes/admin/_common.py index 0ad34d33..2f5c6b87 100644 --- a/apps/services/gateway/gateway/routes/admin/_common.py +++ b/apps/services/gateway/gateway/routes/admin/_common.py @@ -75,6 +75,7 @@ def _get_session_factory() -> Any: create_async_engine, ) + from gateway.db import engine_connect_args settings = get_settings() db_url = os.environ.get("DATABASE_URL", settings.database_url) if "postgresql+psycopg" in db_url: @@ -84,6 +85,7 @@ def _get_session_factory() -> Any: _ENGINE = create_async_engine( db_url, echo=False, pool_pre_ping=True, pool_size=5, max_overflow=10, pool_recycle=1800, + connect_args=engine_connect_args(), ) _SESSION_FACTORY = async_sessionmaker(_ENGINE, expire_on_commit=False) return _SESSION_FACTORY diff --git a/apps/services/gateway/gateway/routes/apps/_common.py b/apps/services/gateway/gateway/routes/apps/_common.py index 96da8aad..110b5be3 100644 --- a/apps/services/gateway/gateway/routes/apps/_common.py +++ b/apps/services/gateway/gateway/routes/apps/_common.py @@ -80,6 +80,8 @@ def _get_session_factory() -> Any: async_sessionmaker, create_async_engine, ) + + from gateway.db import engine_connect_args settings = get_settings() db_url = os.environ.get("DATABASE_URL", settings.database_url) if "postgresql+psycopg" in db_url: @@ -89,7 +91,7 @@ def _get_session_factory() -> Any: _ENGINE = create_async_engine( db_url, echo=False, pool_pre_ping=True, pool_size=10, max_overflow=20, pool_recycle=1800, - connect_args={"timeout": settings.db_connect_timeout}, + connect_args=engine_connect_args(), ) _SESSION_FACTORY = async_sessionmaker(_ENGINE, expire_on_commit=False) return _SESSION_FACTORY diff --git a/apps/services/gateway/gateway/routes/email/core.py b/apps/services/gateway/gateway/routes/email/core.py index 432de1aa..e2150b8f 100644 --- a/apps/services/gateway/gateway/routes/email/core.py +++ b/apps/services/gateway/gateway/routes/email/core.py @@ -393,6 +393,8 @@ def _get_session_factory(): async_sessionmaker, create_async_engine, ) + + from gateway.db import engine_connect_args settings = get_settings() db_url = os.environ.get("DATABASE_URL", settings.database_url) if "postgresql+psycopg" in db_url: @@ -404,10 +406,11 @@ def _get_session_factory(): _ENGINE = create_async_engine( db_url, echo=False, pool_pre_ping=True, pool_size=10, max_overflow=20, pool_recycle=1800, - # Bound the CONNECT phase (asyncpg's `timeout`) so a slow/unreachable - # DB fails fast instead of stalling request handlers — same ceiling - # as acb_graph's engine (settings.db_connect_timeout). - connect_args={"timeout": settings.db_connect_timeout}, + # Bounds the connect phase AND how long one of our sessions may sit + # `idle in transaction`. This engine is the one that drained on + # 2026-08-06 — see gateway/db.py::engine_connect_args for why the + # second bound exists and why it is 10 minutes and not 5. + connect_args=engine_connect_args(), ) _SESSION_FACTORY = async_sessionmaker(_ENGINE, expire_on_commit=False) return _SESSION_FACTORY diff --git a/apps/services/gateway/gateway/routes/notes/core.py b/apps/services/gateway/gateway/routes/notes/core.py index 32938112..362acbcf 100644 --- a/apps/services/gateway/gateway/routes/notes/core.py +++ b/apps/services/gateway/gateway/routes/notes/core.py @@ -152,6 +152,8 @@ def _get_session_factory(): async_sessionmaker, create_async_engine, ) + + from gateway.db import engine_connect_args settings = get_settings() db_url = os.environ.get("DATABASE_URL", settings.database_url) if "postgresql+psycopg" in db_url: @@ -161,7 +163,7 @@ def _get_session_factory(): _ENGINE = create_async_engine( db_url, echo=False, pool_pre_ping=True, pool_size=5, max_overflow=10, pool_recycle=1800, - connect_args={"timeout": settings.db_connect_timeout}, + connect_args=engine_connect_args(), ) _SESSION_FACTORY = async_sessionmaker(_ENGINE, expire_on_commit=False) return _SESSION_FACTORY diff --git a/apps/services/gateway/gateway/routes/whatsapp/core.py b/apps/services/gateway/gateway/routes/whatsapp/core.py index ba9a2290..79f87a42 100644 --- a/apps/services/gateway/gateway/routes/whatsapp/core.py +++ b/apps/services/gateway/gateway/routes/whatsapp/core.py @@ -126,6 +126,8 @@ def _get_session_factory(): async_sessionmaker, create_async_engine, ) + + from gateway.db import engine_connect_args settings = get_settings() db_url = os.environ.get("DATABASE_URL", settings.database_url) if "postgresql+psycopg" in db_url: @@ -135,7 +137,7 @@ def _get_session_factory(): _ENGINE = create_async_engine( db_url, echo=False, pool_pre_ping=True, pool_size=5, max_overflow=10, pool_recycle=1800, - connect_args={"timeout": settings.db_connect_timeout}, + connect_args=engine_connect_args(), ) _SESSION_FACTORY = async_sessionmaker(_ENGINE, expire_on_commit=False) return _SESSION_FACTORY diff --git a/apps/services/gateway/gateway/routes/workflows/core.py b/apps/services/gateway/gateway/routes/workflows/core.py index 010e7688..7e4d716e 100644 --- a/apps/services/gateway/gateway/routes/workflows/core.py +++ b/apps/services/gateway/gateway/routes/workflows/core.py @@ -55,6 +55,7 @@ def _get_session_factory() -> Any: create_async_engine, ) + from gateway.db import engine_connect_args settings = get_settings() db_url = os.environ.get("DATABASE_URL", settings.database_url) if "postgresql+psycopg" in db_url: @@ -68,7 +69,7 @@ def _get_session_factory() -> Any: pool_size=5, max_overflow=10, pool_recycle=1800, - connect_args={"timeout": settings.db_connect_timeout}, + connect_args=engine_connect_args(), ) _SESSION_FACTORY = async_sessionmaker(_ENGINE, expire_on_commit=False) return _SESSION_FACTORY diff --git a/packages/acb_llm/acb_llm/client.py b/packages/acb_llm/acb_llm/client.py index aee77c81..152519e7 100644 --- a/packages/acb_llm/acb_llm/client.py +++ b/packages/acb_llm/acb_llm/client.py @@ -30,6 +30,43 @@ "timeout", "connection", "retry", "service unavailable", ) +#: Hard ceiling on a single provider round-trip, in seconds. +#: +#: Every completion below runs under ``asyncio.wait_for`` at this bound. That is +#: not belt-and-braces over litellm's own ``timeout`` — it is the guarantee. A +#: provider-side ``timeout`` is only honoured to the extent the provider's +#: transport honours it, and on 2026-08-06 one that was not took production down: +#: a completion hung, and because callers across the email automation package hold +#: an open SQLAlchemy session across the call (``AsyncSession`` opens a +#: transaction on first ``execute()`` and holds it until commit/rollback/close), +#: the hung call parked a Postgres transaction `idle in transaction` for 14h44m. +#: A migration's ``ALTER TABLE`` then queued behind that transaction's ACCESS +#: SHARE lock, and because Postgres's lock queue is FIFO every later reader of the +#: table queued behind the *waiting* ALTER. Sending mail — which reads +#: ``email_assistant_settings`` for the signature — stopped working, and the +#: connection pool drained behind the blocked readers. +#: +#: So the bound that matters is on the WALL CLOCK of the await, not on the +#: provider's good intentions. Worst case per call is 3 attempts x this ceiling +#: plus 6s of backoff. +#: +#: Override with ``LLM_REQUEST_TIMEOUT_SECS``; a per-call ``timeout=`` kwarg still +#: wins for the provider-side bound, but never lifts the wall-clock ceiling. +_DEFAULT_REQUEST_TIMEOUT_SECS = 90.0 + + +def _request_timeout_secs() -> float: + """The per-attempt wall-clock ceiling, from env or the default.""" + raw = os.getenv("LLM_REQUEST_TIMEOUT_SECS", "").strip() + if raw: + try: + val = float(raw) + if val > 0: + return val + except ValueError: + _log.warning("acb_llm.bad_request_timeout", value=raw[:32]) + return _DEFAULT_REQUEST_TIMEOUT_SECS + # ── Tier → model mapping ────────────────────────────────────────────────── # Populated from config.yaml + tier_overrides.yaml at import time so the # runtime always matches the configured tiers. Falls back to these hardcoded @@ -664,17 +701,26 @@ async def complete( if enable_litellm_cache: extra.setdefault("cache", {"no-cache": False, "no-store": False}) + # Ask the provider to bound itself too, so a well-behaved transport fails + # cleanly rather than being cancelled mid-flight. `setdefault` — an explicit + # per-call `timeout=` still wins here. + ceiling = _request_timeout_secs() + extra.setdefault("timeout", ceiling) + last_exc: Exception | None = None for attempt in range(3): if attempt > 0: await asyncio.sleep(2 ** attempt) # 2 s, then 4 s try: - response = await acompletion( - model=model, - messages=messages, - temperature=temperature, - max_tokens=max_tokens, - **extra, + response = await asyncio.wait_for( + acompletion( + model=model, + messages=messages, + temperature=temperature, + max_tokens=max_tokens, + **extra, + ), + timeout=ceiling, ) # content can be None for thinking models (e.g. gemini-2.5-pro returns # reasoning tokens separately; the text content field is null until done). @@ -688,6 +734,15 @@ async def complete( _emit_usage(model, tier.value, response) content = choices[0]["message"]["content"] return content or "" # type: ignore[no-any-return,index] + except TimeoutError as exc: + # MUST precede the generic handler. `asyncio.wait_for`'s TimeoutError + # stringifies to '', so the substring test below reads it as + # NON-transient and re-raises — which would turn the one failure this + # ceiling exists to make survivable into a hard error on attempt 1. + _log.warning("acb_llm.request_timeout", model=model, + attempt=attempt + 1, ceiling_secs=ceiling) + last_exc = exc + continue except Exception as exc: if any(token in str(exc).lower() for token in _TRANSIENT_ERRORS): last_exc = exc @@ -735,19 +790,29 @@ async def complete_with_tools( if enable_litellm_cache: extra.setdefault("cache", {"no-cache": False, "no-store": False}) + # Same wall-clock ceiling as `complete` — see _DEFAULT_REQUEST_TIMEOUT_SECS. + # Tool-calling turns are the LONGEST-lived completions in the codebase and + # the ones most often awaited with a DB session open, so bounding this one + # matters more than bounding the plain-text path, not less. + ceiling = _request_timeout_secs() + extra.setdefault("timeout", ceiling) + last_exc: Exception | None = None for attempt in range(3): if attempt > 0: await asyncio.sleep(2 ** attempt) try: - response = await acompletion( - model=model, - messages=messages, - tools=tools, - tool_choice=tool_choice, - temperature=temperature, - max_tokens=max_tokens, - **extra, + response = await asyncio.wait_for( + acompletion( + model=model, + messages=messages, + tools=tools, + tool_choice=tool_choice, + temperature=temperature, + max_tokens=max_tokens, + **extra, + ), + timeout=ceiling, ) choices = response.get("choices") or [] if not choices: @@ -792,6 +857,12 @@ async def complete_with_tools( return result + except TimeoutError as exc: + # MUST precede the generic handler — see the note in `complete`. + _log.warning("acb_llm.request_timeout", model=model, + attempt=attempt + 1, ceiling_secs=ceiling) + last_exc = exc + continue except Exception as exc: if any(token in str(exc).lower() for token in _TRANSIENT_ERRORS): last_exc = exc diff --git a/scripts/apply_migrations.sh b/scripts/apply_migrations.sh index 8731a9f9..f6080514 100644 --- a/scripts/apply_migrations.sh +++ b/scripts/apply_migrations.sh @@ -16,12 +16,37 @@ # # Usage: scripts/apply_migrations.sh # Env: APP_DIR (default /opt/acb/app), PG_CONTAINER (default acb-postgres) +# MIGRATION_LOCK_TIMEOUT (default 5s), MIGRATION_LOCK_RETRIES (default 5) set -euo pipefail APP_DIR="${APP_DIR:-/opt/acb/app}" PG_CONTAINER="${PG_CONTAINER:-acb-postgres}" MIGRATIONS_DIR="$APP_DIR/infra/postgres" +# --- Never WAIT for a lock. This is the whole outage, in one setting. --------- +# +# 2026-08-06: a gateway session sat `idle in transaction` on +# email_assistant_settings for 14h44m (a hung LLM call, with a SQLAlchemy session +# open across it). This runner then asked for ACCESS EXCLUSIVE on that table to +# replay `39_email_learned_writing_style.sql` — and waited. +# +# Waiting is what made it an outage rather than a slow deploy. Postgres's lock +# queue is FIFO: once an ACCESS EXCLUSIVE request is QUEUED, every later reader +# queues behind it, even though the reader would not have conflicted with the +# stale transaction it is ultimately waiting on. So one idle session plus one +# patient ALTER froze the table for the entire application. Sending mail reads +# email_assistant_settings for the signature; it stopped. Blocked readers each +# pinned a pooled connection until the pool drained, and endpoints with nothing +# to do with email started answering 500. +# +# With a lock_timeout the ALTER gives up in seconds and never enters the queue, +# so a stale reader can delay a migration but can no longer freeze a table. +# Retries absorb the ordinary case (a long-running query holding the table for a +# moment); exhausting them fails the deploy LOUDLY, which is the correct outcome +# and the one that used to be indistinguishable from a hang. +MIGRATION_LOCK_TIMEOUT="${MIGRATION_LOCK_TIMEOUT:-5s}" +MIGRATION_LOCK_RETRIES="${MIGRATION_LOCK_RETRIES:-5}" + # Pull DB credentials from .env when present, else fall back to compose defaults. ENV_FILE="$APP_DIR/.env" PG_USER="acb" @@ -92,16 +117,43 @@ for f in $(ls "$MIGRATIONS_DIR"/[0-9][0-9]*_*.sql | sort -V); do 00_*|01_*) continue ;; # init-only, skip esac printf " - %s ... " "$base" - if docker exec -i "$PG_CONTAINER" \ - psql -v ON_ERROR_STOP=1 -U "$PG_USER" -d "$PG_DB" -q < "$f" >/dev/null 2>/tmp/migrate_err; then - echo "ok" - applied=$((applied + 1)) - else + attempt=1 + while :; do + # `SET lock_timeout` is prepended to the stream rather than passed as a psql + # flag so it lands in the SAME session as the migration, ahead of any BEGIN + # the file opens. Session-level, so one SET covers every statement in it. + if { printf 'SET lock_timeout = %s;\n' "$MIGRATION_LOCK_TIMEOUT"; cat "$f"; } \ + | docker exec -i "$PG_CONTAINER" \ + psql -v ON_ERROR_STOP=1 -U "$PG_USER" -d "$PG_DB" -q \ + >/dev/null 2>/tmp/migrate_err; then + echo "ok" + applied=$((applied + 1)) + break + fi + # Only a LOCK timeout is retryable. Any other psql error is a real migration + # failure and must not be papered over by trying it four more times. + if grep -qi 'lock timeout' /tmp/migrate_err \ + && [ "$attempt" -lt "$MIGRATION_LOCK_RETRIES" ]; then + printf "lock busy, retry %d/%d ... " "$attempt" "$MIGRATION_LOCK_RETRIES" + sleep $((attempt * 5)) + attempt=$((attempt + 1)) + continue + fi echo "FAILED" + if grep -qi 'lock timeout' /tmp/migrate_err; then + echo " Could not acquire a lock on this table after $MIGRATION_LOCK_RETRIES tries." >&2 + echo " Something is holding it. Find the holder before re-running:" >&2 + echo " SELECT pid, state, now()-state_change AS dur, query" >&2 + echo " FROM pg_stat_activity" >&2 + echo " WHERE datname = '$PG_DB' AND state <> 'idle'" >&2 + echo " ORDER BY state_change;" >&2 + echo " A session 'idle in transaction' for minutes is a wedged app" >&2 + echo " handler; pg_terminate_backend() releases it." >&2 + fi echo " ----- psql error -----" >&2 sed 's/^/ /' /tmp/migrate_err >&2 exit 1 - fi + done done say "Migrations complete ($applied file(s) applied idempotently)" diff --git a/tests/unit/test_stalled_session_cannot_freeze_the_db.py b/tests/unit/test_stalled_session_cannot_freeze_the_db.py new file mode 100644 index 00000000..2b4264c4 --- /dev/null +++ b/tests/unit/test_stalled_session_cannot_freeze_the_db.py @@ -0,0 +1,248 @@ +"""One wedged handler must not be able to take the database down. + +The 2026-08-06 outage, and the three independent bounds that now stop it +recurring. Worth stating the chain once, because every assertion below is a link +in it and each looks arbitrary on its own: + +1. An LLM completion hung. ``acompletion`` was awaited with no wall-clock bound. +2. The caller held a SQLAlchemy ``AsyncSession`` across that await. A session + opens a transaction on first ``execute()`` and holds it until + commit/rollback/close, so the hung call parked a Postgres transaction + `idle in transaction` — holding ACCESS SHARE on ``email_assistant_settings`` + — for 14h44m. +3. ``apply_migrations.sh`` asked for ACCESS EXCLUSIVE on that table to replay an + idempotent ``ALTER TABLE``, and waited. Postgres's lock queue is FIFO, so + every later reader queued behind the *waiting* ALTER — including the one the + send path makes to read the account's signature. Sending mail stopped, and + the connection pool drained behind the blocked readers. + +Any ONE of these bounds breaks the chain. All three are here because the chain +had no bound at all, and the cheapest way for it to come back is for someone to +restore one link while the other two look fine. +""" +from __future__ import annotations + +import asyncio +import re +import time +from pathlib import Path + +import pytest + +REPO = Path(__file__).resolve().parents[2] + + +# ── Link 1: the LLM call has a wall-clock ceiling ──────────────────────────── + +def test_request_timeout_default_is_positive_and_finite(): + from acb_llm.client import _request_timeout_secs + + ceiling = _request_timeout_secs() + assert 0 < ceiling < 600, ( + "an absent or enormous ceiling re-opens the unbounded await that " + "parked a transaction for 14h44m" + ) + + +def test_request_timeout_honours_env_and_ignores_junk(monkeypatch): + from acb_llm.client import _DEFAULT_REQUEST_TIMEOUT_SECS, _request_timeout_secs + + monkeypatch.setenv("LLM_REQUEST_TIMEOUT_SECS", "12.5") + assert _request_timeout_secs() == 12.5 + + # A typo'd or negative value must fall back, never disable the bound. + for junk in ("banana", "0", "-5", ""): + monkeypatch.setenv("LLM_REQUEST_TIMEOUT_SECS", junk) + assert _request_timeout_secs() == _DEFAULT_REQUEST_TIMEOUT_SECS + + +def _stub_llm(monkeypatch, acompletion): + """Neutralise everything `complete` touches except the call under test.""" + from acb_llm import client as m + + async def _no_keys() -> None: + return None + + monkeypatch.setattr(m, "_ensure_keys_loaded", _no_keys) + monkeypatch.setattr(m, "ensure_model_registered", lambda _model: None) + monkeypatch.setattr( + m, "apply_prompt_caching", + lambda *, model, messages, tools, cache_key, extra: (messages, tools, extra), + ) + monkeypatch.setattr(m, "_emit_usage", lambda *a, **k: None) + monkeypatch.setattr(m, "acompletion", acompletion) + # Skip the 2s/4s inter-attempt backoff so this stays a unit test. The + # ceiling itself is NOT skipped — that is what is under test. + + async def _no_sleep(_secs): + return None + + monkeypatch.setattr(m.asyncio, "sleep", _no_sleep) + monkeypatch.setenv("LLM_REQUEST_TIMEOUT_SECS", "0.05") + + +async def test_a_hanging_completion_raises_instead_of_hanging(monkeypatch): + """THE regression. A provider that never answers must not pin the caller. + + Before the fix this awaited forever, and because callers hold a DB session + across it, forever meant a Postgres transaction held open for as long as the + process lived. + """ + from acb_llm.client import LLMTier, complete + + never = asyncio.Event() + attempts = {"n": 0} + + async def _hang(**_kw): + attempts["n"] += 1 + await never.wait() # a provider that accepted the request and went quiet + + _stub_llm(monkeypatch, _hang) # ceiling = 0.05s + + started = time.monotonic() + with pytest.raises(TimeoutError): + await asyncio.wait_for( + complete(tier=LLMTier.TIER_1, messages=[{"role": "user", "content": "x"}]), + timeout=10, + ) + elapsed = time.monotonic() - started + + # NOT vacuous. The outer wait_for would ALSO raise TimeoutError if the inner + # ceiling were missing, so `pytest.raises` alone would pass on the broken + # code. These two are what actually distinguish the fix: 3 attempts at 0.05s + # each cannot look like one 10s hang, and a caller that never bounded itself + # would only ever have entered `acompletion` once. + assert elapsed < 2, f"took {elapsed:.2f}s — the inner ceiling did not fire" + assert attempts["n"] == 3, "each attempt must be bounded, not just the first" + + +async def test_a_timeout_is_retried_not_re_raised(monkeypatch): + """``except TimeoutError`` must sit ABOVE the generic handler. + + ``asyncio.wait_for``'s TimeoutError stringifies to '', so the generic + handler's ``any(token in str(exc).lower() ...)`` test reads it as + NON-transient and re-raises. Ordering the handlers wrongly would turn a + single slow provider response into a hard failure on attempt 1 — converting + the fix into a new, quieter outage. + """ + from acb_llm.client import LLMTier, complete + + calls = {"n": 0} + never = asyncio.Event() + + async def _hang_twice_then_answer(**_kw): + calls["n"] += 1 + if calls["n"] <= 2: + await never.wait() + return {"choices": [{"message": {"content": "recovered"}}]} + + _stub_llm(monkeypatch, _hang_twice_then_answer) + + # The outer bound is a CI guard, not the assertion. Without the ceiling this + # test would otherwise block forever on the first hang, and a suite that + # hangs is strictly worse feedback than one that fails — that is the whole + # lesson of the incident, applied to the test that pins it. + out = await asyncio.wait_for( + complete(tier=LLMTier.TIER_1, messages=[{"role": "user", "content": "x"}]), + timeout=10, + ) + assert out == "recovered" + assert calls["n"] == 3, "both timeouts should have been retried, not raised" + + +# ── Link 2: a session cannot sit `idle in transaction` forever ─────────────── + +def test_engine_connect_args_bound_both_getting_in_and_staying_in(): + from gateway.db import engine_connect_args + + args = engine_connect_args() + assert args["timeout"] > 0, "connect phase must stay bounded" + + ms = args["server_settings"]["idle_in_transaction_session_timeout"] + # asyncpg passes server_settings verbatim in the startup packet, so a + # non-string here fails at connect time, in production, not here. + assert isinstance(ms, str), "asyncpg requires server_settings values as str" + assert int(ms) > 0 + + +def test_idle_ceiling_clears_the_llm_worst_case(): + """The coupling that a future edit will otherwise silently break. + + A rules run legitimately awaits a completion with a session open, and + ``complete`` retries 3x with 2s+4s of backoff. If the DB's idle ceiling ever + drops below that, Postgres starts killing HEALTHY transactions mid-retry — + which reads as random, unattributable failures under load. + """ + from acb_llm.client import _request_timeout_secs + from gateway.db import engine_connect_args + + llm_worst_case = 3 * _request_timeout_secs() + 6 + idle_ceiling = int( + engine_connect_args()["server_settings"][ + "idle_in_transaction_session_timeout" + ] + ) / 1000 + assert idle_ceiling > llm_worst_case, ( + f"idle_in_transaction_session_timeout ({idle_ceiling}s) must exceed the " + f"LLM worst case ({llm_worst_case}s). Raise one and you must raise the " + f"other." + ) + + +def test_every_gateway_engine_carries_the_bounds(): + """A new app package must not be able to open engine number eight without it. + + Per-site opt-in is how the email engine ended up as the one that drained; + this is the guard that makes forgetting visible. + """ + roots = [REPO / "apps/services/gateway/gateway"] + offenders = [] + for root in roots: + for path in root.rglob("*.py"): + src = path.read_text(encoding="utf-8", errors="replace") + for match in re.finditer(r"create_async_engine\(", src): + tail = src[match.start():match.start() + 900] + if "engine_connect_args()" not in tail: + line = src[: match.start()].count("\n") + 1 + offenders.append(f"{path.relative_to(REPO)}:{line}") + assert not offenders, ( + "these engines do not bound idle-in-transaction sessions: " + + ", ".join(offenders) + ) + + +# ── Link 3: a migration never WAITS for a lock ─────────────────────────────── + +def _migration_runner() -> str: + return (REPO / "scripts/apply_migrations.sh").read_text(encoding="utf-8") + + +def test_migrations_set_a_lock_timeout(): + src = _migration_runner() + assert "SET lock_timeout" in src, ( + "without lock_timeout an ALTER TABLE queues for ACCESS EXCLUSIVE and, " + "because the lock queue is FIFO, freezes the table for every later " + "reader — which is what took sending mail down" + ) + assert "MIGRATION_LOCK_TIMEOUT" in src + + +def test_the_unbounded_psql_invocation_is_gone(): + """Pins the SHAPE, not just the presence of a setting. + + Re-adding a bare `psql ... < "$f"` alongside the bounded one would leave the + hazard in place while every other assertion here still passed. + """ + src = _migration_runner() + assert not re.search(r'psql[^\n|]*<\s*"\$f"', src), ( + "a migration is being piped to psql without the lock_timeout prelude" + ) + + +def test_only_lock_timeouts_are_retried(): + """A retry loop that swallows real SQL errors would be worse than no loop.""" + src = _migration_runner() + assert "MIGRATION_LOCK_RETRIES" in src + assert re.search(r"grep -qi 'lock timeout'", src), ( + "the retry must be gated on the error actually being a lock timeout" + )