From f57845ba5e0414cd5c5a154cedc97d398c2cfb0a Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 15:49:18 +0800 Subject: [PATCH] fix(db): enforce global nullpool session budget --- .gitea/workflows/cd.yaml | 8 + apps/api/src/core/unit_of_work.py | 57 +-- apps/api/src/db/base.py | 222 +++++++++-- .../executor_trust_boundary_readback.py | 14 + .../test_executor_trust_boundary_readback.py | 21 ++ .../tests/test_runtime_bootstrap_guards.py | 356 +++++++++++++++++- ..._signal_worker_ansible_executor_binding.py | 1 + ...t_workload_database_identity_projection.py | 58 +++ k8s/awoooi-prod/06-deployment-api.yaml | 5 + .../test_cd_controlled_runtime_profile.py | 2 + ...erify_awoooi_api_liveness_stabilization.py | 24 +- ...erify_awoooi_api_liveness_stabilization.py | 8 +- 12 files changed, 709 insertions(+), 67 deletions(-) diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml index 3d682d36f..0d1283445 100644 --- a/.gitea/workflows/cd.yaml +++ b/.gitea/workflows/cd.yaml @@ -4123,6 +4123,12 @@ jobs: API_DB_NULL_POOL=$(normalize_bool "$($KUBECTL get deployment/awoooi-api \ -n awoooi-prod -o jsonpath='{.spec.template.spec.containers[?(@.name=="api")].env[?(@.name=="DATABASE_NULL_POOL")].value}')") + API_DB_NULL_POOL_CONCURRENCY_LIMIT=$($KUBECTL get deployment/awoooi-api \ + -n awoooi-prod -o jsonpath='{.spec.template.spec.containers[?(@.name=="api")].env[?(@.name=="DATABASE_NULL_POOL_CONCURRENCY_LIMIT")].value}' \ + 2>/dev/null || true) + case "$API_DB_NULL_POOL_CONCURRENCY_LIMIT" in + ''|*[!0-9]*) API_DB_NULL_POOL_CONCURRENCY_LIMIT=0 ;; + esac WORKER_DB_NULL_POOL=$(normalize_bool "$($KUBECTL get deployment/awoooi-worker \ -n awoooi-prod -o jsonpath='{.spec.template.spec.containers[?(@.name=="worker")].env[?(@.name=="DATABASE_NULL_POOL")].value}')") BROKER_DB_NULL_POOL=$(normalize_bool "$($KUBECTL get deployment/awoooi-ansible-executor-broker \ @@ -4173,6 +4179,7 @@ jobs: if [ "$DB_IDENTITY_ISOLATED" = "true" ] \ && [ "$REPRESENTATIVE_DB_PREFLIGHTS_VERIFIED" = "true" ] \ && [ "$API_DB_NULL_POOL" = "true" ] \ + && [ "$API_DB_NULL_POOL_CONCURRENCY_LIMIT" -eq 2 ] \ && [ "$WORKER_DB_NULL_POOL" = "true" ] \ && [ "$BROKER_DB_NULL_POOL" = "true" ] \ && [ "$API_DB_CONNECTION_LIMIT" -eq 12 ] \ @@ -4476,6 +4483,7 @@ jobs: --from-literal=api_ready_replica_count="$API_READY_REPLICA_COUNT" \ --from-literal=api_desired_replica_count="$API_DESIRED_REPLICA_COUNT" \ --from-literal=api_database_null_pool="$API_DB_NULL_POOL" \ + --from-literal=api_database_null_pool_concurrency_limit="$API_DB_NULL_POOL_CONCURRENCY_LIMIT" \ --from-literal=worker_database_null_pool="$WORKER_DB_NULL_POOL" \ --from-literal=broker_database_null_pool="$BROKER_DB_NULL_POOL" \ --from-literal=api_monolithic_secret_env_from_absent="$API_MONOLITHIC_SECRET_ENVFROM_ABSENT" \ diff --git a/apps/api/src/core/unit_of_work.py b/apps/api/src/core/unit_of_work.py index a192375e3..c55676821 100644 --- 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: """ diff --git a/apps/api/src/db/base.py b/apps/api/src/db/base.py index 684405284..90abd4f02 100644 --- 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 # ============================================================================= diff --git a/apps/api/src/services/executor_trust_boundary_readback.py b/apps/api/src/services/executor_trust_boundary_readback.py index 4810413ba..5f8a54318 100644 --- 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, diff --git a/apps/api/tests/test_executor_trust_boundary_readback.py b/apps/api/tests/test_executor_trust_boundary_readback.py index f45087216..d8ac3faae 100644 --- 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" diff --git a/apps/api/tests/test_runtime_bootstrap_guards.py b/apps/api/tests/test_runtime_bootstrap_guards.py index 60b6f52df..cc428b73b 100644 --- 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 diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py index 139e768d9..5f512c75a 100644 --- 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" diff --git a/apps/api/tests/test_workload_database_identity_projection.py b/apps/api/tests/test_workload_database_identity_projection.py index 63b197810..b1f84c324 100644 --- 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 diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index 92c3a8de9..6b217c4bb 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -177,6 +177,11 @@ spec: # 2026-07-11 Codex: release each API connection after use while # workload-specific DB identities are being split and verified. value: "true" + - name: DATABASE_NULL_POOL_CONCURRENCY_LIMIT + # The API role allows 12 connections and must retain 8 as verified + # headroom. Two single-process API replicas may therefore use at + # most 2 sessions each: 2 replicas * 2 = 4 runtime connections. + value: "2" - name: DATABASE_BOOTSTRAP_DDL_ENABLED # 2026-07-14 Codex: production schema DDL has one owner. CD # migration preflight applies/verifies it before API rollout; diff --git a/ops/runner/test_cd_controlled_runtime_profile.py b/ops/runner/test_cd_controlled_runtime_profile.py index e50ce4ce9..19d8b112d 100644 --- a/ops/runner/test_cd_controlled_runtime_profile.py +++ b/ops/runner/test_cd_controlled_runtime_profile.py @@ -124,6 +124,8 @@ def test_prod_db_pool_budget_and_non_overlap_rollouts_are_source_controlled() -> assert "value: \"1\"" in api assert "DATABASE_MAX_OVERFLOW" in api assert "value: \"0\"" in api + assert "DATABASE_NULL_POOL_CONCURRENCY_LIMIT" in api + assert "2 replicas * 2 = 4 runtime connections" in api assert "DATABASE_BOOTSTRAP_DDL_ENABLED" in api assert "CD migration preflight" in api assert "replicas: 2" in api diff --git a/ops/runner/test_verify_awoooi_api_liveness_stabilization.py b/ops/runner/test_verify_awoooi_api_liveness_stabilization.py index 5428f27c7..f2dee9987 100644 --- a/ops/runner/test_verify_awoooi_api_liveness_stabilization.py +++ b/ops/runner/test_verify_awoooi_api_liveness_stabilization.py @@ -3,7 +3,6 @@ from __future__ import annotations import importlib.util from pathlib import Path - ROOT = Path(__file__).resolve().parents[2] SCRIPT = ROOT / "ops" / "runner" / "verify_awoooi_api_liveness_stabilization.py" SPEC = importlib.util.spec_from_file_location("api_liveness_verifier", SCRIPT) @@ -27,6 +26,7 @@ def _receipt() -> dict[str, str]: return { "connection_budget_verified": "true", "api_database_null_pool": "true", + "api_database_null_pool_concurrency_limit": "2", "api_representative_db_preflight_passed": "true", "api_http_concurrency_probe_passed": "true", "api_db_connection_limit": "12", @@ -82,6 +82,28 @@ def test_live_contract_fails_closed_on_budget_restart_or_cpu_regression() -> Non assert "live_cpu_saturated:awoooi-api-a:1000m" in failures +def test_live_contract_rejects_missing_api_null_pool_concurrency_cap() -> None: + pods = _pods() + restarts = {pod["metadata"]["name"]: 0 for pod in pods} + receipt = _receipt() + receipt["api_database_null_pool_concurrency_limit"] = "0" + + failures = verifier.evaluate_live_state( + _live_deployment(), + receipt, + pods, + restarts, + restarts, + {pod["metadata"]["name"]: 410 for pod in pods}, + 900, + ) + + assert ( + "receipt_api_database_null_pool_concurrency_limit_not_verified" + in failures + ) + + def test_edge_receipt_requires_a_qualified_production_vantage() -> None: assert verifier.evaluate_edge_receipt( "192.168.0.120", diff --git a/ops/runner/verify_awoooi_api_liveness_stabilization.py b/ops/runner/verify_awoooi_api_liveness_stabilization.py index d3106e41e..3f725233e 100644 --- a/ops/runner/verify_awoooi_api_liveness_stabilization.py +++ b/ops/runner/verify_awoooi_api_liveness_stabilization.py @@ -12,7 +12,6 @@ import urllib.request from pathlib import Path from typing import Any - ROOT = Path(__file__).resolve().parents[2] MANIFEST = ROOT / "k8s" / "awoooi-prod" / "06-deployment-api.yaml" NAMESPACE = "awoooi-prod" @@ -81,11 +80,15 @@ def evaluate_source_contract(deployment: dict[str, Any]) -> list[str]: "replicas": (spec.get("replicas"), 2), "max_surge": (strategy.get("maxSurge"), 0), "max_unavailable": (strategy.get("maxUnavailable"), 1), - "cpu_limit": ((container.get("resources") or {}).get("limits", {}).get("cpu"), "1"), + "cpu_limit": ((container.get("resources") or {}).get("limits", {}).get("cpu"), "2"), "memory_limit": ((container.get("resources") or {}).get("limits", {}).get("memory"), "1Gi"), "database_pool_size": (env.get("DATABASE_POOL_SIZE"), "1"), "database_max_overflow": (env.get("DATABASE_MAX_OVERFLOW"), "0"), "database_null_pool": (env.get("DATABASE_NULL_POOL"), "true"), + "database_null_pool_concurrency_limit": ( + env.get("DATABASE_NULL_POOL_CONCURRENCY_LIMIT"), + "2", + ), "database_bootstrap_ddl_enabled": ( env.get("DATABASE_BOOTSTRAP_DDL_ENABLED"), "false", @@ -154,6 +157,7 @@ def evaluate_live_state( expected_receipt = { "connection_budget_verified": "true", "api_database_null_pool": "true", + "api_database_null_pool_concurrency_limit": "2", "api_representative_db_preflight_passed": "true", "api_http_concurrency_probe_passed": "true", }