fix(api): cap db pool during prod rollout

apps/api/tests/test_config_url_validation.py

@@ -164,3 +164,11 @@ def test_empty_string_fallback_url_accepted():
     """OLLAMA_FALLBACK_URL 預設空字串（未設定），應通過"""
     s = _make_settings(OLLAMA_FALLBACK_URL="")
     assert s.OLLAMA_FALLBACK_URL == ""
+
+
+def test_database_pool_budget_defaults_to_production_safe_values():
+    """Production DB role has a tiny connection limit; defaults must not fan out."""
+    s = _make_settings()
+    assert s.DATABASE_POOL_SIZE == 1
+    assert s.DATABASE_MAX_OVERFLOW == 0
+    assert s.DATABASE_POOL_TIMEOUT_SECONDS == 5.0

apps/api/tests/test_runtime_bootstrap_guards.py

@@ -33,6 +33,9 @@ class _FakeLockConnection:
     async def __aexit__(self, *_exc: object) -> None:
         return None
 
+    async def commit(self) -> None:
+        self.statements.append("COMMIT")
+
     async def execute(
         self,
         statement: object,
@@ -54,6 +57,37 @@ class _FakeEngine:
         return self.lock_conn
 
 
+class _ConnectionBudgetEngine:
+    def connect(self) -> _FakeLockConnection:
+        raise RuntimeError('too many connections for role "awoooi"')
+
+
+def test_get_engine_uses_database_pool_budget(monkeypatch):
+    from src.db import base as db_base
+
+    captured: dict[str, object] = {}
+    fake_engine = object()
+
+    def fake_create_async_engine(database_url: str, **kwargs: object) -> object:
+        captured["database_url"] = database_url
+        captured.update(kwargs)
+        return fake_engine
+
+    monkeypatch.setattr(db_base, "_engine", None)
+    monkeypatch.setattr(db_base, "_session_factory", None)
+    monkeypatch.setattr(db_base.settings, "DATABASE_URL", "postgresql+asyncpg://u:p@localhost/db")
+    monkeypatch.setattr(db_base.settings, "DATABASE_POOL_SIZE", 1)
+    monkeypatch.setattr(db_base.settings, "DATABASE_MAX_OVERFLOW", 0)
+    monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 5.0)
+    monkeypatch.setattr(db_base, "create_async_engine", fake_create_async_engine)
+
+    assert db_base.get_engine() is fake_engine
+    assert captured["database_url"] == "postgresql+asyncpg://u:p@localhost/db"
+    assert captured["pool_size"] == 1
+    assert captured["max_overflow"] == 0
+    assert captured["pool_timeout"] == 5.0
+
+
 @pytest.mark.asyncio
 async def test_init_db_serializes_bootstrap_ddl_with_advisory_lock(monkeypatch):
     from src.db import base as db_base
@@ -69,9 +103,27 @@ async def test_init_db_serializes_bootstrap_ddl_with_advisory_lock(monkeypatch):
 
     await db_base.init_db()
 
-    assert calls == [fake_engine]
+    assert calls == [fake_engine.lock_conn]
     assert "pg_try_advisory_lock" in fake_engine.lock_conn.statements[0]
-    assert "pg_advisory_unlock" in fake_engine.lock_conn.statements[-1]
+    assert any("pg_advisory_unlock" in stmt for stmt in fake_engine.lock_conn.statements)
+    assert "COMMIT" in fake_engine.lock_conn.statements
+
+
+@pytest.mark.asyncio
+async def test_init_db_skips_bootstrap_when_connection_budget_exhausted(monkeypatch):
+    from src.db import base as db_base
+
+    calls: list[object] = []
+
+    async def fake_run_init_db_ddl(engine: object) -> None:
+        calls.append(engine)
+
+    monkeypatch.setattr(db_base, "get_engine", lambda: _ConnectionBudgetEngine())
+    monkeypatch.setattr(db_base, "_run_init_db_ddl", fake_run_init_db_ddl)
+
+    await db_base.init_db()
+
+    assert calls == []
 
 
 @pytest.mark.asyncio
@@ -90,7 +142,7 @@ async def test_init_db_releases_bootstrap_lock_when_ddl_fails(monkeypatch):
         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]
+    assert any("pg_advisory_unlock" in stmt for stmt in fake_engine.lock_conn.statements)
 
 
 @pytest.mark.asyncio
@@ -135,7 +187,7 @@ async def test_init_db_releases_bootstrap_lock_when_ddl_times_out(monkeypatch):
     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]
+    assert any("pg_advisory_unlock" in stmt for stmt in fake_engine.lock_conn.statements)
 
 
 @pytest.mark.asyncio