fix(db): enforce global nullpool session budget

2026-07-15 15:49:18 +08:00
parent 19d7db93e5
commit f57845ba5e
12 changed files with 709 additions and 67 deletions
--- a/apps/api/src/core/unit_of_work.py
+++ b/apps/api/src/core/unit_of_work.py
@@ -83,14 +83,27 @@ class UnitOfWork:
        """進入事務"""
        from src.core.context import get_current_project_id

-        self._session = self._session_factory()
+        if self._session is not None:
+            raise RuntimeError("UnitOfWork cannot be entered more than once")
+
+        session = self._session_factory()
+        self._session = session
        effective_pid = (
-            self._project_id if self._project_id is not None else get_current_project_id()
-        )
-        await self._session.execute(
-            text("SELECT set_config('app.project_id', :pid, TRUE)"),
-            {"pid": effective_pid},
+            self._project_id
+            if self._project_id is not None
+            else get_current_project_id()
        )
+        try:
+            await self._session.execute(
+                text("SELECT set_config('app.project_id', :pid, TRUE)"),
+                {"pid": effective_pid},
+            )
+        except BaseException:
+            # ``__aexit__`` is not invoked when ``__aenter__`` fails.  Close
+            # here so a budgeted session cannot leak its global DB slot.
+            self._session = None
+            await session.close()
+            raise
        self._committed = False
        logger.debug("uow_started", project_id=effective_pid)
        return self
@@ -109,22 +122,24 @@ class UnitOfWork:
        - 無例外且未手動 commit: 自動 commit
        - 已手動 commit: 不做任何事
        """
-        if exc_type is not None:
-            # 有例外，rollback
-            await self.rollback()
-            logger.warning(
-                "uow_rollback_on_exception",
-                exc_type=exc_type.__name__ if exc_type else None,
-                exc_val=str(exc_val) if exc_val else None,
-            )
-        elif not self._committed:
-            # 無例外且未手動 commit，自動 commit
-            await self.commit()
-
-        # 關閉 session
-        if self._session:
-            await self._session.close()
+        try:
+            if exc_type is not None:
+                # 有例外，rollback
+                await self.rollback()
+                logger.warning(
+                    "uow_rollback_on_exception",
+                    exc_type=exc_type.__name__ if exc_type else None,
+                    exc_val=str(exc_val) if exc_val else None,
+                )
+            elif not self._committed:
+                # 無例外且未手動 commit，自動 commit
+                await self.commit()
+        finally:
+            # Commit/rollback failures must also release a budgeted session.
+            session = self._session
            self._session = None
+            if session:
+                await session.close()

    async def commit(self) -> None:
        """
--- a/apps/api/src/db/base.py
+++ b/apps/api/src/db/base.py
@@ -16,6 +16,7 @@ Features:
 import asyncio
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
+from typing import Any

 from fastapi import HTTPException
 from sqlalchemy import text
@@ -25,6 +26,7 @@ from sqlalchemy.ext.asyncio import (
    AsyncConnection,
    AsyncEngine,
    AsyncSession,
+    AsyncSessionTransaction,
    async_sessionmaker,
    create_async_engine,
 )
@@ -98,6 +100,160 @@ async def _database_session_slot() -> AsyncGenerator[None, None]:
        limiter.release()


+class _BudgetedAsyncSession(AsyncSession):
+    """Apply one NullPool concurrency lease to every session access path.
+
+    ``get_db`` and ``get_db_context`` used to own the semaphore lease outside
+    the SQLAlchemy session.  Callers that consumed ``get_session_factory()``
+    directly (notably ``UnitOfWork``) bypassed that boundary entirely.  Keep
+    the lease on the session instead so dependency, context-manager, raw
+    factory, and transaction APIs all share the same per-process budget.
+
+    The lease is lazy for sessions constructed but never used, and is held
+    until close/reset/invalidate so a single session cannot churn through the
+    semaphore while retaining transactional state.
+    """
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self._database_slot_context: Any | None = None
+        self._database_slot_lock = asyncio.Lock()
+
+    async def _acquire_database_slot(self) -> None:
+        if self._database_slot_context is not None:
+            return
+        async with self._database_slot_lock:
+            if self._database_slot_context is not None:
+                return
+            slot_context = _database_session_slot()
+            await slot_context.__aenter__()
+            self._database_slot_context = slot_context
+
+    async def _release_database_slot(self) -> None:
+        async with self._database_slot_lock:
+            slot_context = self._database_slot_context
+            self._database_slot_context = None
+            if slot_context is not None:
+                await slot_context.__aexit__(None, None, None)
+
+    async def __aenter__(self) -> "_BudgetedAsyncSession":
+        slot_was_held = self._database_slot_context is not None
+        await self._acquire_database_slot()
+        try:
+            return await super().__aenter__()
+        except BaseException:
+            # A failed context entry has no matching ``__aexit__``.  Close a
+            # lease acquired by this entry so cancellation/future SQLAlchemy
+            # entry failures cannot strand one of the process-wide slots.
+            if not slot_was_held:
+                await self.close()
+            raise
+
+    def begin(self) -> AsyncSessionTransaction:
+        return _BudgetedAsyncSessionTransaction(self)
+
+    def begin_nested(self) -> AsyncSessionTransaction:
+        return _BudgetedAsyncSessionTransaction(self, nested=True)
+
+    async def connection(self, *args: Any, **kwargs: Any) -> Any:
+        await self._acquire_database_slot()
+        return await super().connection(*args, **kwargs)
+
+    async def delete(self, *args: Any, **kwargs: Any) -> None:
+        await self._acquire_database_slot()
+        await super().delete(*args, **kwargs)
+
+    async def execute(self, *args: Any, **kwargs: Any) -> Any:
+        await self._acquire_database_slot()
+        return await super().execute(*args, **kwargs)
+
+    async def flush(self, *args: Any, **kwargs: Any) -> None:
+        await self._acquire_database_slot()
+        await super().flush(*args, **kwargs)
+
+    async def get(self, *args: Any, **kwargs: Any) -> Any:
+        await self._acquire_database_slot()
+        return await super().get(*args, **kwargs)
+
+    async def get_one(self, *args: Any, **kwargs: Any) -> Any:
+        await self._acquire_database_slot()
+        return await super().get_one(*args, **kwargs)
+
+    async def merge(self, *args: Any, **kwargs: Any) -> Any:
+        await self._acquire_database_slot()
+        return await super().merge(*args, **kwargs)
+
+    async def refresh(self, *args: Any, **kwargs: Any) -> None:
+        await self._acquire_database_slot()
+        await super().refresh(*args, **kwargs)
+
+    async def run_sync(self, *args: Any, **kwargs: Any) -> Any:
+        await self._acquire_database_slot()
+        return await super().run_sync(*args, **kwargs)
+
+    async def scalar(self, *args: Any, **kwargs: Any) -> Any:
+        await self._acquire_database_slot()
+        return await super().scalar(*args, **kwargs)
+
+    async def scalars(self, *args: Any, **kwargs: Any) -> Any:
+        await self._acquire_database_slot()
+        return await super().scalars(*args, **kwargs)
+
+    async def stream(self, *args: Any, **kwargs: Any) -> Any:
+        await self._acquire_database_slot()
+        return await super().stream(*args, **kwargs)
+
+    async def stream_scalars(self, *args: Any, **kwargs: Any) -> Any:
+        await self._acquire_database_slot()
+        return await super().stream_scalars(*args, **kwargs)
+
+    async def commit(self) -> None:
+        await self._acquire_database_slot()
+        await super().commit()
+
+    async def close(self) -> None:
+        try:
+            await super().close()
+        finally:
+            await self._release_database_slot()
+
+    async def reset(self) -> None:
+        try:
+            await super().reset()
+        finally:
+            await self._release_database_slot()
+
+    async def invalidate(self) -> None:
+        try:
+            await super().invalidate()
+        finally:
+            await self._release_database_slot()
+
+
+class _BudgetedAsyncSessionTransaction(AsyncSessionTransaction):
+    """Acquire the session lease for explicit ``begin`` entry paths."""
+
+    async def start(
+        self,
+        is_ctxmanager: bool = False,
+    ) -> AsyncSessionTransaction:
+        session = self.session
+        if isinstance(session, _BudgetedAsyncSession):
+            slot_was_held = session._database_slot_context is not None
+            await session._acquire_database_slot()
+            try:
+                return await super().start(is_ctxmanager=is_ctxmanager)
+            except BaseException:
+                # ``async_sessionmaker.begin()`` cannot call its exit path
+                # when transaction entry fails.  Close only when this start
+                # acquired the lease; an already-entered session still owns
+                # its pre-existing lease and lifecycle.
+                if not slot_was_held:
+                    await session.close()
+                raise
+        return await super().start(is_ctxmanager=is_ctxmanager)
+
+
 def _raise_unauthorized_db_context(msg: str) -> None:
    context = get_current_project_context()
    logger.error(
@@ -152,12 +308,12 @@ def get_engine() -> AsyncEngine:


 def get_session_factory() -> async_sessionmaker[AsyncSession]:
-    """Get or create session factory"""
+    """Get the globally budgeted async session factory."""
    global _session_factory
    if _session_factory is None:
        _session_factory = async_sessionmaker(
            bind=get_engine(),
-            class_=AsyncSession,
+            class_=_BudgetedAsyncSession,
            expire_on_commit=False,
            autoflush=False,
        )
@@ -178,28 +334,27 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]:
            ...
    """
    factory = get_session_factory()
-    async with _database_session_slot():
-        async with factory() as session:
-            try:
-                from src.core.context import get_current_project_id
+    async with factory() as session:
+        try:
+            from src.core.context import get_current_project_id

-                # AwoooP Phase 2.3 (2026-05-04 ogt): SET LOCAL app.project_id 讓 RLS Policy 生效
-                # Fail-Closed RLS: 遇到未授權情境拋出錯誤而非回退到 "awoooi"
-                pid = get_current_project_id()
-                if not pid:
-                    _raise_unauthorized_db_context(
-                        "Unauthorized: project_id is missing in context (Fail-Closed RLS)"
-                    )
-
-                await session.execute(
-                    text("SELECT set_config('app.project_id', :pid, TRUE)"),
-                    {"pid": pid},
+            # AwoooP Phase 2.3 (2026-05-04 ogt): SET LOCAL app.project_id 讓 RLS Policy 生效
+            # Fail-Closed RLS: 遇到未授權情境拋出錯誤而非回退到 "awoooi"
+            pid = get_current_project_id()
+            if not pid:
+                _raise_unauthorized_db_context(
+                    "Unauthorized: project_id is missing in context (Fail-Closed RLS)"
                )
-                yield session
-                await session.commit()
-            except Exception:
-                await session.rollback()
-                raise
+
+            await session.execute(
+                text("SELECT set_config('app.project_id', :pid, TRUE)"),
+                {"pid": pid},
+            )
+            yield session
+            await session.commit()
+        except Exception:
+            await session.rollback()
+            raise


@asynccontextmanager
@@ -224,18 +379,17 @@ async def get_db_context(project_id: str | None = None) -> AsyncGenerator[AsyncS
        _raise_unauthorized_db_context("Unauthorized: project_id is missing in context (Fail-Closed RLS)")

    factory = get_session_factory()
-    async with _database_session_slot():
-        async with factory() as session:
-            try:
-                await session.execute(
-                    text("SELECT set_config('app.project_id', :pid, TRUE)"),
-                    {"pid": effective_pid},
-                )
-                yield session
-                await session.commit()
-            except Exception:
-                await session.rollback()
-                raise
+    async with factory() as session:
+        try:
+            await session.execute(
+                text("SELECT set_config('app.project_id', :pid, TRUE)"),
+                {"pid": effective_pid},
+            )
+            yield session
+            await session.commit()
+        except Exception:
+            await session.rollback()
+            raise


 # =============================================================================
--- a/apps/api/src/services/executor_trust_boundary_readback.py
+++ b/apps/api/src/services/executor_trust_boundary_readback.py
@@ -154,6 +154,11 @@ def _build_database_identity_boundary(
            "database_null_pool": _truthy(
                receipt.get(f"{workload_id}_database_null_pool")
            ),
+            "null_pool_concurrency_limit": _int_or_none(
+                receipt.get(
+                    f"{workload_id}_database_null_pool_concurrency_limit"
+                )
+            ),
            "monolithic_secret_env_from_absent": _truthy(
                receipt.get(
                    f"{workload_id}_monolithic_secret_env_from_absent"
@@ -198,6 +203,9 @@ def _build_database_identity_boundary(
        workloads[workload_id]["database_null_pool"] is True
        for workload_id in _DB_WORKLOAD_IDS
    )
+    api_null_pool_concurrency_limit_verified = (
+        workloads["api"]["null_pool_concurrency_limit"] == 2
+    )
    workload_budget_verified = all(
        workloads[workload_id]["connection_limit"]
        == _DB_WORKLOAD_BUDGETS[workload_id]["connection_limit"]
@@ -256,6 +264,7 @@ def _build_database_identity_boundary(
    connection_budget_verified = bool(
        db_identity_isolated
        and null_pool_verified
+        and api_null_pool_concurrency_limit_verified
        and workload_budget_verified
        and representative_preflights_verified
        and verifier_verified
@@ -276,6 +285,8 @@ def _build_database_identity_boundary(
        blockers.append("database_role_fingerprints_not_distinct")
    if not api_http_concurrency_verified:
        blockers.append("api_http_concurrency_probe_not_verified")
+    if not api_null_pool_concurrency_limit_verified:
+        blockers.append("api_null_pool_concurrency_limit_not_verified")
    if not verifier_verified:
        blockers.append("database_identity_verifier_not_verified")
    if verified_at_utc is None:
@@ -360,6 +371,9 @@ def _build_database_identity_boundary(
            "distinct_database_secret_refs": distinct_secret_refs,
            "distinct_db_role_fingerprints": distinct_role_fingerprints,
            "database_null_pool_verified": null_pool_verified,
+            "api_null_pool_concurrency_limit_verified": (
+                api_null_pool_concurrency_limit_verified
+            ),
            "workload_connection_budgets_verified": workload_budget_verified,
            "database_identity_verifier_verified": verifier_verified,
            "database_capacity_receipt_fresh": capacity_receipt_fresh,
--- a/apps/api/tests/test_executor_trust_boundary_readback.py
+++ b/apps/api/tests/test_executor_trust_boundary_readback.py
@@ -92,6 +92,7 @@ def _verified_receipt(source_sha: str = "abc123") -> dict[str, str]:
        "representative_db_preflights_verified": "true",
        "api_http_concurrency_probe_passed": "true",
        "api_database_null_pool": "true",
+        "api_database_null_pool_concurrency_limit": "2",
        "worker_database_null_pool": "true",
        "broker_database_null_pool": "true",
        "api_monolithic_secret_env_from_absent": "true",
@@ -128,6 +129,9 @@ def test_executor_trust_boundary_requires_live_controls_and_matching_source() ->
    assert database_boundary["controls"][
        "representative_db_preflights_verified"
    ] is True
+    assert database_boundary["controls"][
+        "api_null_pool_concurrency_limit_verified"
+    ] is True
    assert database_boundary["secret_values_read"] is False
    assert database_boundary["role_names_exposed"] is False
    assert database_boundary["signal_freshness"]["fresh"] is True
@@ -361,6 +365,23 @@ def test_executor_trust_boundary_requires_live_connection_headroom() -> None:
    )


+def test_executor_trust_boundary_requires_api_null_pool_concurrency_cap() -> None:
+    receipt = _verified_receipt()
+    receipt["api_database_null_pool_concurrency_limit"] = "0"
+
+    readback = build_executor_trust_boundary_readback(
+        receipt,
+        deployed_source_sha="abc123",
+    )
+
+    database_boundary = readback["database_identity_boundary"]
+    assert readback["full_workload_boundary_verified"] is False
+    assert database_boundary["connection_budget_verified"] is False
+    assert "api_null_pool_concurrency_limit_not_verified" in (
+        database_boundary["active_blockers"]
+    )
+
+
 def test_executor_trust_boundary_binds_budget_and_verifier_contract() -> None:
    wrong_budget = _verified_receipt()
    wrong_budget["worker_required_connection_budget"] = "4"
--- a/apps/api/tests/test_runtime_bootstrap_guards.py
+++ b/apps/api/tests/test_runtime_bootstrap_guards.py
@@ -225,17 +225,11 @@ def test_get_engine_uses_null_pool_for_bounded_background_processes(monkeypatch)


@pytest.mark.asyncio
-async def test_null_pool_session_limiter_serializes_worker_db_sessions(
+async def test_null_pool_session_limiter_serializes_raw_factory_sessions(
    monkeypatch,
 ) -> None:
    from src.db import base as db_base

-    assert "async with _database_session_slot()" in inspect.getsource(
-        db_base.get_db
-    )
-    assert "async with _database_session_slot()" in inspect.getsource(
-        db_base.get_db_context
-    )
    monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", True)
    monkeypatch.setattr(
        db_base.settings,
@@ -245,17 +239,25 @@ async def test_null_pool_session_limiter_serializes_worker_db_sessions(
    monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 1.0)
    monkeypatch.setattr(db_base, "_null_pool_session_limiter", None)
    monkeypatch.setattr(db_base, "_null_pool_session_limiter_size", 0)
+    monkeypatch.setattr(db_base, "_session_factory", None)
+    monkeypatch.setattr(db_base, "get_engine", lambda: None)
+    factory = db_base.get_session_factory()
+
+    assert factory.class_ is db_base._BudgetedAsyncSession
+    assert "_database_session_slot" not in inspect.getsource(db_base.get_db)
+    assert "_database_session_slot" not in inspect.getsource(db_base.get_db_context)
+
    first_entered = asyncio.Event()
    release_first = asyncio.Event()
    second_entered = asyncio.Event()

    async def first_session() -> None:
-        async with db_base._database_session_slot():
+        async with factory():
            first_entered.set()
            await release_first.wait()

    async def second_session() -> None:
-        async with db_base._database_session_slot():
+        async with factory():
            second_entered.set()

    first_task = asyncio.create_task(first_session())
@@ -270,6 +272,342 @@ async def test_null_pool_session_limiter_serializes_worker_db_sessions(
    assert second_entered.is_set() is True


+@pytest.mark.asyncio
+async def test_unit_of_work_shares_raw_factory_cap_without_double_acquire(
+    monkeypatch,
+) -> None:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    from src.core.unit_of_work import UnitOfWork
+    from src.db import base as db_base
+
+    monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", True)
+    monkeypatch.setattr(
+        db_base.settings,
+        "DATABASE_NULL_POOL_CONCURRENCY_LIMIT",
+        1,
+    )
+    monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 1.0)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter", None)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter_size", 0)
+    monkeypatch.setattr(db_base, "_session_factory", None)
+    monkeypatch.setattr(db_base, "get_engine", lambda: None)
+
+    executed_session_ids: list[int] = []
+
+    async def fake_execute(
+        session: AsyncSession,
+        *_args: object,
+        **_kwargs: object,
+    ) -> object:
+        executed_session_ids.append(id(session))
+        return object()
+
+    async def fake_commit(_session: AsyncSession) -> None:
+        return None
+
+    async def fake_close(_session: AsyncSession) -> None:
+        return None
+
+    monkeypatch.setattr(AsyncSession, "execute", fake_execute)
+    monkeypatch.setattr(AsyncSession, "commit", fake_commit)
+    monkeypatch.setattr(AsyncSession, "close", fake_close)
+
+    factory = db_base.get_session_factory()
+    raw_entered = asyncio.Event()
+    release_raw = asyncio.Event()
+    uow_entered = asyncio.Event()
+
+    async def raw_factory_session() -> None:
+        async with factory():
+            raw_entered.set()
+            await release_raw.wait()
+
+    async def unit_of_work_session() -> None:
+        async with UnitOfWork(factory, project_id="awoooi"):
+            uow_entered.set()
+
+    raw_task = asyncio.create_task(raw_factory_session())
+    await raw_entered.wait()
+    uow_task = asyncio.create_task(unit_of_work_session())
+    await asyncio.sleep(0)
+
+    assert uow_entered.is_set() is False
+    assert executed_session_ids == []
+
+    release_raw.set()
+    await asyncio.gather(raw_task, uow_task)
+
+    assert uow_entered.is_set() is True
+    assert len(executed_session_ids) == 1
+
+    async with asyncio.timeout(0.5):
+        async with db_base.get_db_context("awoooi"):
+            pass
+
+    assert len(executed_session_ids) == 2
+
+
+@pytest.mark.asyncio
+async def test_unit_of_work_enter_failure_releases_global_session_cap(
+    monkeypatch,
+) -> None:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    from src.core.unit_of_work import UnitOfWork
+    from src.db import base as db_base
+
+    monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", True)
+    monkeypatch.setattr(
+        db_base.settings,
+        "DATABASE_NULL_POOL_CONCURRENCY_LIMIT",
+        1,
+    )
+    monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 0.2)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter", None)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter_size", 0)
+    monkeypatch.setattr(db_base, "_session_factory", None)
+    monkeypatch.setattr(db_base, "get_engine", lambda: None)
+
+    execute_attempts = 0
+
+    async def fake_execute(
+        _session: AsyncSession,
+        *_args: object,
+        **_kwargs: object,
+    ) -> object:
+        nonlocal execute_attempts
+        execute_attempts += 1
+        if execute_attempts == 1:
+            raise RuntimeError("synthetic_uow_enter_failure")
+        return object()
+
+    async def fake_commit(_session: AsyncSession) -> None:
+        return None
+
+    async def fake_close(_session: AsyncSession) -> None:
+        return None
+
+    monkeypatch.setattr(AsyncSession, "execute", fake_execute)
+    monkeypatch.setattr(AsyncSession, "commit", fake_commit)
+    monkeypatch.setattr(AsyncSession, "close", fake_close)
+
+    factory = db_base.get_session_factory()
+    with pytest.raises(RuntimeError, match="synthetic_uow_enter_failure"):
+        async with UnitOfWork(factory, project_id="awoooi"):
+            pass
+
+    async with asyncio.timeout(0.5):
+        async with UnitOfWork(factory, project_id="awoooi"):
+            pass
+
+    assert execute_attempts == 2
+
+
+@pytest.mark.asyncio
+async def test_budgeted_transaction_enter_failure_releases_global_session_cap(
+    monkeypatch,
+) -> None:
+    from sqlalchemy.ext.asyncio import AsyncSession, AsyncSessionTransaction
+
+    from src.db import base as db_base
+
+    monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", True)
+    monkeypatch.setattr(
+        db_base.settings,
+        "DATABASE_NULL_POOL_CONCURRENCY_LIMIT",
+        1,
+    )
+    monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 0.2)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter", None)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter_size", 0)
+    monkeypatch.setattr(db_base, "_session_factory", None)
+    monkeypatch.setattr(db_base, "get_engine", lambda: None)
+
+    async def fail_start(
+        _transaction: AsyncSessionTransaction,
+        is_ctxmanager: bool = False,
+    ) -> AsyncSessionTransaction:
+        del is_ctxmanager
+        raise RuntimeError("synthetic_transaction_enter_failure")
+
+    async def fake_close(_session: AsyncSession) -> None:
+        return None
+
+    monkeypatch.setattr(AsyncSessionTransaction, "start", fail_start)
+    monkeypatch.setattr(AsyncSession, "close", fake_close)
+
+    factory = db_base.get_session_factory()
+    with pytest.raises(RuntimeError, match="synthetic_transaction_enter_failure"):
+        async with factory.begin():
+            pass
+
+    async with asyncio.timeout(0.5):
+        async with factory():
+            pass
+
+
+@pytest.mark.asyncio
+async def test_cancelled_transaction_entry_releases_global_session_cap(
+    monkeypatch,
+) -> None:
+    from sqlalchemy.ext.asyncio import AsyncSession, AsyncSessionTransaction
+
+    from src.db import base as db_base
+
+    monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", True)
+    monkeypatch.setattr(
+        db_base.settings,
+        "DATABASE_NULL_POOL_CONCURRENCY_LIMIT",
+        1,
+    )
+    monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 0.2)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter", None)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter_size", 0)
+    monkeypatch.setattr(db_base, "_session_factory", None)
+    monkeypatch.setattr(db_base, "get_engine", lambda: None)
+
+    transaction_start_entered = asyncio.Event()
+    never_complete = asyncio.Event()
+
+    async def block_start(
+        _transaction: AsyncSessionTransaction,
+        is_ctxmanager: bool = False,
+    ) -> AsyncSessionTransaction:
+        del is_ctxmanager
+        transaction_start_entered.set()
+        await never_complete.wait()
+        raise AssertionError("unreachable")
+
+    async def fake_close(_session: AsyncSession) -> None:
+        return None
+
+    monkeypatch.setattr(AsyncSessionTransaction, "start", block_start)
+    monkeypatch.setattr(AsyncSession, "close", fake_close)
+    factory = db_base.get_session_factory()
+
+    async def enter_transaction() -> None:
+        async with factory.begin():
+            pass
+
+    transaction_task = asyncio.create_task(enter_transaction())
+    await transaction_start_entered.wait()
+    transaction_task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await transaction_task
+
+    async with asyncio.timeout(0.5):
+        async with factory():
+            pass
+
+
+@pytest.mark.asyncio
+async def test_cancelled_slot_waiter_does_not_consume_or_over_release_capacity(
+    monkeypatch,
+) -> None:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    from src.db import base as db_base
+
+    monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", True)
+    monkeypatch.setattr(
+        db_base.settings,
+        "DATABASE_NULL_POOL_CONCURRENCY_LIMIT",
+        1,
+    )
+    monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 1.0)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter", None)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter_size", 0)
+    monkeypatch.setattr(db_base, "_session_factory", None)
+    monkeypatch.setattr(db_base, "get_engine", lambda: None)
+
+    async def fake_close(_session: AsyncSession) -> None:
+        return None
+
+    monkeypatch.setattr(AsyncSession, "close", fake_close)
+    factory = db_base.get_session_factory()
+    first_entered = asyncio.Event()
+    release_first = asyncio.Event()
+
+    async def hold_first_slot() -> None:
+        async with factory():
+            first_entered.set()
+            await release_first.wait()
+
+    async def wait_for_slot() -> None:
+        async with factory():
+            pass
+
+    first_task = asyncio.create_task(hold_first_slot())
+    await first_entered.wait()
+    cancelled_waiter = asyncio.create_task(wait_for_slot())
+    await asyncio.sleep(0)
+    cancelled_waiter.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await cancelled_waiter
+
+    release_first.set()
+    await first_task
+
+    async with asyncio.timeout(0.5):
+        async with factory():
+            pass
+
+
+@pytest.mark.asyncio
+async def test_unit_of_work_reentry_fails_without_replacing_active_session(
+    monkeypatch,
+) -> None:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    from src.core.unit_of_work import UnitOfWork
+    from src.db import base as db_base
+
+    monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", True)
+    monkeypatch.setattr(
+        db_base.settings,
+        "DATABASE_NULL_POOL_CONCURRENCY_LIMIT",
+        1,
+    )
+    monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 0.2)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter", None)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter_size", 0)
+    monkeypatch.setattr(db_base, "_session_factory", None)
+    monkeypatch.setattr(db_base, "get_engine", lambda: None)
+
+    async def fake_execute(
+        _session: AsyncSession,
+        *_args: object,
+        **_kwargs: object,
+    ) -> object:
+        return object()
+
+    async def fake_commit(_session: AsyncSession) -> None:
+        return None
+
+    async def fake_close(_session: AsyncSession) -> None:
+        return None
+
+    monkeypatch.setattr(AsyncSession, "execute", fake_execute)
+    monkeypatch.setattr(AsyncSession, "commit", fake_commit)
+    monkeypatch.setattr(AsyncSession, "close", fake_close)
+
+    factory = db_base.get_session_factory()
+    unit_of_work = UnitOfWork(factory, project_id="awoooi")
+    async with unit_of_work:
+        active_session = unit_of_work.session
+        with pytest.raises(
+            RuntimeError,
+            match="UnitOfWork cannot be entered more than once",
+        ):
+            await unit_of_work.__aenter__()
+        assert unit_of_work.session is active_session
+
+    async with asyncio.timeout(0.5):
+        async with factory():
+            pass
+
+
@pytest.mark.asyncio
 async def test_init_db_serializes_bootstrap_ddl_with_advisory_lock(monkeypatch):
    from src.db import base as db_base
--- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py
+++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py
@@ -144,6 +144,7 @@ def test_api_manifest_has_read_only_identity_and_no_ssh_executor_mounts() -> Non
    assert pod_spec["serviceAccountName"] == "awoooi-api"
    assert env["SSH_MCP_ENABLED"] == "false"
    assert env["DATABASE_NULL_POOL"] == "true"
+    assert env["DATABASE_NULL_POOL_CONCURRENCY_LIMIT"] == "2"
    assert env["AWOOOI_API_BACKGROUND_TASKS_ENABLED"] == "false"
    assert env["ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER"] == "false"
    assert env["ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER"] == "false"
--- a/apps/api/tests/test_workload_database_identity_projection.py
+++ b/apps/api/tests/test_workload_database_identity_projection.py
@@ -196,6 +196,11 @@ def test_runtime_workloads_use_distinct_database_secret_projections() -> None:
        "awoooi-broker-db",
    }

+    api_env = _env_by_name(
+        _deployment_container("k8s/awoooi-prod/06-deployment-api.yaml", "api")
+    )
+    assert api_env["DATABASE_NULL_POOL_CONCURRENCY_LIMIT"]["value"] == "2"
+

 def test_api_and_worker_project_shared_secrets_explicitly() -> None:
    for path, container_name in (
@@ -217,6 +222,36 @@ def test_api_and_worker_project_shared_secrets_explicitly() -> None:
            assert secret_ref["optional"] is True


+def test_api_null_pool_cap_preserves_the_verified_role_headroom() -> None:
+    documents = yaml.safe_load_all(
+        (
+            REPO_ROOT / "k8s/awoooi-prod/06-deployment-api.yaml"
+        ).read_text()
+    )
+    deployment = next(
+        document
+        for document in documents
+        if document and document.get("kind") == "Deployment"
+    )
+    container = next(
+        item
+        for item in deployment["spec"]["template"]["spec"]["containers"]
+        if item["name"] == "api"
+    )
+    env = _env_by_name(container)
+    replica_count = int(deployment["spec"]["replicas"])
+    per_process_cap = int(
+        env["DATABASE_NULL_POOL_CONCURRENCY_LIMIT"]["value"]
+    )
+
+    role_connection_limit = 12
+    required_free_headroom = 8
+    assert replica_count == 2
+    assert replica_count * per_process_cap == (
+        role_connection_limit - required_free_headroom
+    )
+
+
 def test_workload_database_bootstrap_keeps_secret_values_off_output() -> None:
    script = (
        REPO_ROOT / "scripts/ops/awoooi-workload-db-identity-bootstrap.sh"
@@ -261,6 +296,7 @@ def test_workload_database_bootstrap_keeps_secret_values_off_output() -> None:

 def test_cd_uses_live_spec_rollback_and_non_mutating_post_verifier() -> None:
    workflow = (REPO_ROOT / ".gitea/workflows/cd.yaml").read_text()
+    normalized_workflow = " ".join(workflow.split())

    assert "WORKLOAD_DB_ROLLBACK_FILE" in workflow
    assert "WORKLOAD_DB_SECRET_PREEXISTENCE_FILE" in workflow
@@ -285,6 +321,28 @@ def test_cd_uses_live_spec_rollback_and_non_mutating_post_verifier() -> None:
    assert "API_PUBLIC_READINESS_ATTEMPTS" in workflow
    assert "API_HTTP_CONCURRENCY_ATTEMPTS" in workflow
    assert "API_HTTP_PROBE_SAFETY_RESERVE=4" in workflow
+    assert '"$API_DB_NULL_POOL_CONCURRENCY_LIMIT" -eq 2' in workflow
+    assert (
+        '"$API_DB_AVAILABLE_CONNECTION_HEADROOM" -ge '
+        '"$API_DB_REQUIRED_CONNECTION_BUDGET"'
+    ) in workflow
+    assert (
+        '"$WORKER_DB_AVAILABLE_CONNECTION_HEADROOM" -ge '
+        '"$WORKER_DB_REQUIRED_CONNECTION_BUDGET"'
+    ) in workflow
+    assert (
+        '"$BROKER_DB_AVAILABLE_CONNECTION_HEADROOM" -ge '
+        '"$BROKER_DB_REQUIRED_CONNECTION_BUDGET"'
+    ) in workflow
+    assert (
+        "available_connection_headroom >= required_connection_budget"
+        in normalized_workflow
+    )
+    assert (
+        '"$API_DB_AVAILABLE_CONNECTION_HEADROOM" -ge '
+        '"$API_DB_REPRESENTATIVE_PROBE_COUNT"'
+    ) not in workflow
+    assert "api_database_null_pool_concurrency_limit" in workflow
    assert "API_HTTP_CONCURRENCY_WORKERS" in workflow
    assert "API sequential readiness permanent HTTP status" in workflow
    assert "api/v1/health/ready" in workflow