fix(api): bound bootstrap lock during startup
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Failing after 34s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been skipped

This commit is contained in:
Your Name
2026-06-30 08:19:40 +08:00
parent 15b68a0767
commit 5f99459d1b
3 changed files with 134 additions and 13 deletions

View File

@@ -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]