From 5f99459d1b02e83ad7f408ce42ada36a1e9b0049 Mon Sep 17 00:00:00 2001 From: Your Name Date: Tue, 30 Jun 2026 08:19:40 +0800 Subject: [PATCH] fix(api): bound bootstrap lock during startup --- apps/api/src/db/base.py | 58 ++++++++++++-- .../tests/test_runtime_bootstrap_guards.py | 76 +++++++++++++++++-- docs/LOGBOOK.md | 13 ++++ 3 files changed, 134 insertions(+), 13 deletions(-) diff --git a/apps/api/src/db/base.py b/apps/api/src/db/base.py index 4b7cdbb8c..85b7f38a5 100644 --- a/apps/api/src/db/base.py +++ b/apps/api/src/db/base.py @@ -13,6 +13,7 @@ Features: - 所有 Episodic Memory 必須使用 PostgreSQL """ +import asyncio from collections.abc import AsyncGenerator from contextlib import asynccontextmanager @@ -183,6 +184,9 @@ async def get_db_context(project_id: str | None = None) -> AsyncGenerator[AsyncS # ============================================================================= _DB_BOOTSTRAP_LOCK_NAME = "awoooi:init_db:ddl" +_DB_BOOTSTRAP_LOCK_WAIT_SECONDS = 45.0 +_DB_BOOTSTRAP_LOCK_POLL_SECONDS = 2.0 +_DB_BOOTSTRAP_DDL_WAIT_SECONDS = 120.0 async def init_db() -> None: @@ -196,13 +200,27 @@ async def init_db() -> None: async with engine.connect() as lock_conn: # 2026-05-24 ogt + Codex: 兩個 API replica 同時啟動時，PostgreSQL 會在 # ALTER TABLE ... IF NOT EXISTS 上互相等待並 deadlock。整段 bootstrap - # DDL 必須序列化，避免 rollout 因一個 pod CrashLoop 變成 1/2 ready。 - await lock_conn.execute( - text("SELECT pg_advisory_lock(hashtext(:lock_name))"), - {"lock_name": _DB_BOOTSTRAP_LOCK_NAME}, - ) + # DDL 必須序列化。 + # 2026-06-30 Codex: production rollout 觀察到新 API pod 卡在 init_db() + # 超過 startup probe 預算。bootstrap DDL 是冪等補欄流程，不能再用 + # 無界 pg_advisory_lock 把新版 API route 擋成永久舊版；改為有界等待， + # 逾時則 fail-visible skip，讓 serving path 先恢復，再由 verifier 回報。 + lock_acquired = await _try_acquire_bootstrap_lock(lock_conn) + if not lock_acquired: + return + try: - await _run_init_db_ddl(engine) + try: + await asyncio.wait_for( + _run_init_db_ddl(engine), + timeout=_DB_BOOTSTRAP_DDL_WAIT_SECONDS, + ) + except TimeoutError: + logger.warning( + "database_bootstrap_ddl_timeout_skipped", + timeout_seconds=_DB_BOOTSTRAP_DDL_WAIT_SECONDS, + lock_name=_DB_BOOTSTRAP_LOCK_NAME, + ) finally: await lock_conn.execute( text("SELECT pg_advisory_unlock(hashtext(:lock_name))"), @@ -210,6 +228,34 @@ async def init_db() -> None: ) +async def _try_acquire_bootstrap_lock(lock_conn: object) -> bool: + loop = asyncio.get_running_loop() + deadline = loop.time() + _DB_BOOTSTRAP_LOCK_WAIT_SECONDS + + while True: + result = await lock_conn.execute( + text("SELECT pg_try_advisory_lock(hashtext(:lock_name))"), + {"lock_name": _DB_BOOTSTRAP_LOCK_NAME}, + ) + if bool(result.scalar()): + logger.info( + "database_bootstrap_lock_acquired", + lock_name=_DB_BOOTSTRAP_LOCK_NAME, + ) + return True + + remaining = deadline - loop.time() + if remaining <= 0: + logger.warning( + "database_bootstrap_lock_timeout_skipped", + timeout_seconds=_DB_BOOTSTRAP_LOCK_WAIT_SECONDS, + lock_name=_DB_BOOTSTRAP_LOCK_NAME, + ) + return False + + await asyncio.sleep(min(_DB_BOOTSTRAP_LOCK_POLL_SECONDS, remaining)) + + async def _run_init_db_ddl(engine: AsyncEngine) -> None: """ Run idempotent DB bootstrap DDL while caller holds the bootstrap advisory lock. diff --git a/apps/api/tests/test_runtime_bootstrap_guards.py b/apps/api/tests/test_runtime_bootstrap_guards.py index 9e662c4a1..412101d6d 100644 --- a/apps/api/tests/test_runtime_bootstrap_guards.py +++ b/apps/api/tests/test_runtime_bootstrap_guards.py @@ -14,9 +14,18 @@ from typing import Any import pytest +class _FakeScalarResult: + def __init__(self, value: bool) -> None: + self.value = value + + def scalar(self) -> bool: + return self.value + + class _FakeLockConnection: - def __init__(self) -> None: + def __init__(self, lock_results: list[bool] | None = None) -> None: self.statements: list[str] = [] + self._lock_results = lock_results or [True] async def __aenter__(self) -> _FakeLockConnection: return self @@ -24,14 +33,22 @@ class _FakeLockConnection: async def __aexit__(self, *_exc: object) -> None: return None - async def execute(self, statement: object, params: dict[str, str] | None = None) -> None: - self.statements.append(str(statement)) + async def execute( + self, + statement: object, + params: dict[str, str] | None = None, + ) -> _FakeScalarResult: + sql = str(statement) + self.statements.append(sql) assert params == {"lock_name": "awoooi:init_db:ddl"} + if "pg_try_advisory_lock" in sql: + return _FakeScalarResult(self._lock_results.pop(0)) + return _FakeScalarResult(True) class _FakeEngine: - def __init__(self) -> None: - self.lock_conn = _FakeLockConnection() + def __init__(self, lock_results: list[bool] | None = None) -> None: + self.lock_conn = _FakeLockConnection(lock_results=lock_results) def connect(self) -> _FakeLockConnection: return self.lock_conn @@ -53,7 +70,7 @@ async def test_init_db_serializes_bootstrap_ddl_with_advisory_lock(monkeypatch): await db_base.init_db() assert calls == [fake_engine] - assert "pg_advisory_lock" in fake_engine.lock_conn.statements[0] + assert "pg_try_advisory_lock" in fake_engine.lock_conn.statements[0] assert "pg_advisory_unlock" in fake_engine.lock_conn.statements[-1] @@ -72,7 +89,52 @@ async def test_init_db_releases_bootstrap_lock_when_ddl_fails(monkeypatch): with pytest.raises(RuntimeError, match="ddl failed"): await db_base.init_db() - assert "pg_advisory_lock" in fake_engine.lock_conn.statements[0] + assert "pg_try_advisory_lock" in fake_engine.lock_conn.statements[0] + assert "pg_advisory_unlock" in fake_engine.lock_conn.statements[-1] + + +@pytest.mark.asyncio +async def test_init_db_skips_bootstrap_when_advisory_lock_times_out(monkeypatch): + from src.db import base as db_base + + fake_engine = _FakeEngine(lock_results=[False]) + calls: list[object] = [] + + async def fake_run_init_db_ddl(engine: object) -> None: + calls.append(engine) + + monkeypatch.setattr(db_base, "get_engine", lambda: fake_engine) + monkeypatch.setattr(db_base, "_run_init_db_ddl", fake_run_init_db_ddl) + monkeypatch.setattr(db_base, "_DB_BOOTSTRAP_LOCK_WAIT_SECONDS", 0.0) + + await db_base.init_db() + + assert calls == [] + assert "pg_try_advisory_lock" in fake_engine.lock_conn.statements[0] + assert all("pg_advisory_unlock" not in stmt for stmt in fake_engine.lock_conn.statements) + + +@pytest.mark.asyncio +async def test_init_db_releases_bootstrap_lock_when_ddl_times_out(monkeypatch): + from src.db import base as db_base + + fake_engine = _FakeEngine() + + async def fake_run_init_db_ddl(_engine: object) -> None: + raise AssertionError("wait_for should own ddl execution in this test") + + async def fake_wait_for(coro: Awaitable[Any], timeout: float) -> None: + assert timeout == 120.0 + coro.close() + raise TimeoutError + + monkeypatch.setattr(db_base, "get_engine", lambda: fake_engine) + monkeypatch.setattr(db_base, "_run_init_db_ddl", fake_run_init_db_ddl) + monkeypatch.setattr(db_base.asyncio, "wait_for", fake_wait_for) + + await db_base.init_db() + + assert "pg_try_advisory_lock" in fake_engine.lock_conn.statements[0] assert "pg_advisory_unlock" in fake_engine.lock_conn.statements[-1] diff --git a/docs/LOGBOOK.md b/docs/LOGBOOK.md index a7121e0a9..2cb45c848 100644 --- a/docs/LOGBOOK.md +++ b/docs/LOGBOOK.md @@ -1,3 +1,16 @@ +## 2026-06-30 — 08:19 API bootstrap controlled startup unblock + +**照主線完成的實作**： +- Production `awoooi-api` 新 image 已 rollout 到 pod，但新版 pod 卡在 `init_db()`，只剩舊 pod 對外服務，導致新 `agent-log-controlled-writeback-consumer-readback` route 仍未出現在正式 OpenAPI。 +- `apps/api/src/db/base.py` 已把無界 `pg_advisory_lock` 改成有界 `pg_try_advisory_lock`，bootstrap DDL 也加上 120 秒 timeout；逾時時記錄 fail-visible skip，避免冪等補欄流程把 API route rollout 永久卡成舊版。 +- 既有 DDL 序列化邊界仍保留：拿到 lock 才跑 bootstrap DDL，DDL 失敗仍釋放 lock；只是把非事故級 guard 改成 controlled apply 可讀回行為。 + +**驗證**： +- Focused pytest：runtime bootstrap guards / LOG consumer readback / autonomous runtime-control / CD profile `41 passed`。 +- `ruff`、`py_compile`：通過。 + +**邊界**：未重啟主機，未刪 pod，未 rollout restart，未操作 DB / K8s write / Docker / firewall，未讀 secret / token / raw sessions / SQLite / `.env`，未使用 GitHub / `gh` / GitHub API。 + ## 2026-06-30 — 02:28 P1-LOG KM/RAG/PlayBook/MCP consumer readback **照主線完成的實作**：