diff --git a/apps/api/src/core/config.py b/apps/api/src/core/config.py index 8a310cbaf..2aec50780 100644 --- a/apps/api/src/core/config.py +++ b/apps/api/src/core/config.py @@ -298,6 +298,16 @@ class Settings(BaseSettings): "processes that share a constrained production role." ), ) + DATABASE_NULL_POOL_CONCURRENCY_LIMIT: int = Field( + default=0, + ge=0, + le=50, + description=( + "Per-process concurrent DB session cap when NullPool is enabled. " + "Zero leaves concurrency unchanged; bounded workers set this to " + "their verified role budget." + ), + ) DATABASE_BOOTSTRAP_DDL_ENABLED: bool = Field( default=True, description=( diff --git a/apps/api/src/db/base.py b/apps/api/src/db/base.py index 652f310b5..684405284 100644 --- a/apps/api/src/db/base.py +++ b/apps/api/src/db/base.py @@ -50,9 +50,54 @@ class Base(DeclarativeBase): _engine: AsyncEngine | None = None _session_factory: async_sessionmaker[AsyncSession] | None = None +_null_pool_session_limiter: asyncio.BoundedSemaphore | None = None +_null_pool_session_limiter_size = 0 logger = get_logger("awoooi.db") +def _get_null_pool_session_limiter() -> asyncio.BoundedSemaphore | None: + """Honor the configured connection budget even when NullPool is active.""" + + global _null_pool_session_limiter, _null_pool_session_limiter_size + limit = int(settings.DATABASE_NULL_POOL_CONCURRENCY_LIMIT) + if not settings.DATABASE_NULL_POOL or limit <= 0: + return None + if _null_pool_session_limiter is None or _null_pool_session_limiter_size != limit: + _null_pool_session_limiter = asyncio.BoundedSemaphore(limit) + _null_pool_session_limiter_size = limit + return _null_pool_session_limiter + + +@asynccontextmanager +async def _database_session_slot() -> AsyncGenerator[None, None]: + """Bound concurrent NullPool sessions without retaining DB connections.""" + + limiter = _get_null_pool_session_limiter() + if limiter is None: + yield + return + try: + await asyncio.wait_for( + limiter.acquire(), + timeout=settings.DATABASE_POOL_TIMEOUT_SECONDS, + ) + except TimeoutError as exc: + logger.warning( + "database_null_pool_session_concurrency_limit_reached", + concurrency_limit=int( + settings.DATABASE_NULL_POOL_CONCURRENCY_LIMIT + ), + timeout_seconds=settings.DATABASE_POOL_TIMEOUT_SECONDS, + ) + raise SQLAlchemyTimeoutError( + "NullPool database session concurrency limit reached" + ) from exc + try: + yield + finally: + limiter.release() + + def _raise_unauthorized_db_context(msg: str) -> None: context = get_current_project_context() logger.error( @@ -133,27 +178,28 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]: ... """ factory = get_session_factory() - async with factory() as session: - try: - from src.core.context import get_current_project_id + async with _database_session_slot(): + 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)" + # 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}, ) - - 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 + yield session + await session.commit() + except Exception: + await session.rollback() + raise @asynccontextmanager @@ -178,17 +224,18 @@ 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 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 _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 # ============================================================================= diff --git a/apps/api/src/jobs/agent99_controlled_dispatch_reconciler_job.py b/apps/api/src/jobs/agent99_controlled_dispatch_reconciler_job.py index 55ae37c7a..728d64997 100644 --- a/apps/api/src/jobs/agent99_controlled_dispatch_reconciler_job.py +++ b/apps/api/src/jobs/agent99_controlled_dispatch_reconciler_job.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import hashlib from collections.abc import Awaitable, Callable +from datetime import UTC, datetime from typing import Any import structlog @@ -54,6 +55,25 @@ LegacyIncidentFetcher = Callable[..., Awaitable[list[dict[str, Any]]]] Agent99Dispatcher = Callable[..., Awaitable[dict[str, Any]]] +def _dispatch_timeout_elapsed(value: Any) -> bool: + if not isinstance(value, datetime): + return False + normalized = value if value.tzinfo is not None else value.replace(tzinfo=UTC) + return normalized <= datetime.now(UTC) + + +async def _terminalize_expired_outcome( + ledger: Any, + item: dict[str, Any], +) -> bool: + if not _dispatch_timeout_elapsed(item.get("timeout_at")): + return False + terminal = await ledger.record_outcome_timeout_no_write_terminal( + identity=item["identity"], + ) + return terminal.get("receipt_persisted") is True + + def _is_bound_no_write_status_outcome( outcome: dict[str, Any] | None, *, @@ -594,6 +614,7 @@ async def reconcile_agent99_controlled_dispatches_once( "status_reconcile_requested": 0, "status_reconcile_pending": 0, "external_no_write_reconciled": 0, + "outcome_timeout_terminalized": 0, "verifier_written": 0, "closed": 0, } @@ -622,7 +643,10 @@ async def reconcile_agent99_controlled_dispatches_once( source_receipt.get("resolved") is not True or not source_receipt_ref ): - counters["source_not_resolved"] += 1 + if await _terminalize_expired_outcome(ledger, item): + counters["outcome_timeout_terminalized"] += 1 + else: + counters["source_not_resolved"] += 1 continue outcome = await asyncio.to_thread( read_agent99_sre_outcome, @@ -633,6 +657,9 @@ async def reconcile_agent99_controlled_dispatches_once( identity=identity, source_receipt_ref=source_receipt_ref, ): + if await _terminalize_expired_outcome(ledger, item): + counters["outcome_timeout_terminalized"] += 1 + continue request_receipt = await asyncio.to_thread( request_agent99_same_run_status_reconcile, identity, @@ -674,6 +701,8 @@ async def reconcile_agent99_controlled_dispatches_once( str(identity.run_id), ) if not isinstance(outcome, dict): + if await _terminalize_expired_outcome(ledger, item): + counters["outcome_timeout_terminalized"] += 1 continue mode = str(outcome.get("mode") or mode) outcome_contract = ( @@ -707,6 +736,8 @@ async def reconcile_agent99_controlled_dispatches_once( evidence_refs=evidence_refs, ) if verifier.get("receipt_persisted") is not True: + if await _terminalize_expired_outcome(ledger, item): + counters["outcome_timeout_terminalized"] += 1 continue counters["verifier_written"] += 1 receipt = verifier @@ -735,7 +766,20 @@ async def run_agent99_controlled_dispatch_reconciler_loop() -> None: error=str(exc), ) try: - await reconcile_agent99_controlled_dispatches_once() + reconciled = await reconcile_agent99_controlled_dispatches_once() + if any( + reconciled[key] > 0 + for key in ( + "external_no_write_reconciled", + "outcome_timeout_terminalized", + "verifier_written", + "closed", + ) + ): + logger.info( + "agent99_controlled_dispatch_reconciler_tick", + **reconciled, + ) except asyncio.CancelledError: raise except Exception as exc: diff --git a/apps/api/src/services/agent99_controlled_dispatch_ledger.py b/apps/api/src/services/agent99_controlled_dispatch_ledger.py index af334421b..e64ac799a 100644 --- a/apps/api/src/services/agent99_controlled_dispatch_ledger.py +++ b/apps/api/src/services/agent99_controlled_dispatch_ledger.py @@ -1752,6 +1752,200 @@ class PostgresAgent99DispatchLedger: "runtime_closure_verified": False, } + async def record_outcome_timeout_no_write_terminal( + self, + *, + identity: Agent99DispatchIdentity, + ) -> dict[str, Any]: + """Fail an expired run without replaying transport or claiming closure.""" + + now = _utc_now_naive() + error_code = "E-AGENT99-OUTCOME-TIMEOUT" + try: + async with get_db_context(identity.project_id) as db: + current_result = await db.execute( + select( + AwoooPRunState.state, + AwoooPRunState.error_detail, + AwoooPRunState.timeout_at, + ) + .where( + AwoooPRunState.run_id == identity.run_id, + AwoooPRunState.project_id == identity.project_id, + ) + .with_for_update() + ) + current = current_result.one_or_none() + envelope = _parse_envelope( + current.error_detail if current is not None else None + ) + identity_matches = bool( + isinstance(envelope, dict) + and _stage_identity_matches( + identity, + envelope.get("identity") + if isinstance(envelope.get("identity"), dict) + else {}, + ) + ) + if ( + current is not None + and current.state == "failed" + and identity_matches + and envelope.get("outcome_timeout_no_write_terminal") is True + ): + return { + **envelope, + "status": "outcome_timeout_no_write_terminal_idempotent", + "receipt_persisted": True, + "runtime_closure_verified": False, + } + if current is None or not identity_matches: + return { + "status": "outcome_timeout_run_identity_invalid_fail_closed", + "receipt_persisted": False, + "runtime_closure_verified": False, + } + timeout_at = current.timeout_at + if ( + current.state != "waiting_tool" + or timeout_at is None + or timeout_at > now + ): + return { + "status": "outcome_timeout_not_due_no_write", + "receipt_persisted": False, + "runtime_closure_verified": False, + } + + prior_controlled_apply_authorized = bool( + envelope.get("controlled_apply_authorized") is True + ) + prior_runtime_execution_authorized = bool( + envelope.get("runtime_execution_authorized") is True + ) + prior_runtime_execution_attempted = bool( + envelope.get("runtime_execution_attempted") is True + ) + terminal_receipt = { + "schema_version": "agent99_outcome_timeout_no_write_v1", + "status": "authenticated_outcome_timeout", + **_stage_identity(identity), + "timeout_at": timeout_at.isoformat(), + "terminal_at": now.isoformat(), + "terminalizer_runtime_write_performed": False, + "original_runtime_write_status": ( + "unknown_without_authenticated_outcome" + ), + "transport_replayed": False, + "incident_resolution_authorized": False, + "learning_writeback_authorized": False, + "telegram_authorized": False, + "stores_raw_evidence": False, + } + envelope.update({ + "status": "outcome_timeout_no_write_terminal", + "run_terminal_state": "failed", + "outcome_timeout_no_write_terminal": True, + "prior_controlled_apply_authorized": ( + prior_controlled_apply_authorized + ), + "prior_runtime_execution_authorized": ( + prior_runtime_execution_authorized + ), + "runtime_execution_attempted": ( + prior_runtime_execution_attempted + ), + "controlled_apply_authorized": False, + "runtime_execution_authorized": False, + "terminalizer_runtime_write_performed": False, + "outcome_runtime_write_status": ( + "unknown_without_authenticated_outcome" + ), + "transport_replayed": False, + "automation_execution_success": False, + "post_verifier_passed": False, + "runtime_closure_verified": False, + "closure_complete": False, + "incident_resolution_authorized": False, + "safe_next_action": ( + "await_new_source_recurrence_after_agent99_relay_recovery" + ), + "verifier": { + **_stage_identity(identity), + "step_seq": 2, + "status": "failed_authenticated_outcome_timeout", + "post_verifier_passed": False, + "receipt": terminal_receipt, + }, + "learning_writeback": { + **_stage_identity(identity), + "step_seq": 3, + "status": "not_applicable_outcome_timeout_no_write", + "required_receipts": [], + "receipt_refs": {}, + "stores_raw_evidence": False, + }, + }) + envelope_json = _stable_json(envelope) + updated = await db.execute( + update(AwoooPRunState) + .where( + AwoooPRunState.run_id == identity.run_id, + AwoooPRunState.project_id == identity.project_id, + AwoooPRunState.state == "waiting_tool", + AwoooPRunState.timeout_at <= now, + ) + .values( + state="failed", + output_sha256=_sha256(envelope_json), + error_code=error_code, + error_detail=envelope_json, + heartbeat_at=now, + completed_at=now, + lease_until=None, + next_attempt_at=None, + worker_id=None, + ) + .returning(AwoooPRunState.run_id) + ) + persisted = updated.scalar_one_or_none() is not None + if persisted: + await db.execute( + update(AwoooPRunStepJournal) + .where( + AwoooPRunStepJournal.run_id == identity.run_id, + AwoooPRunStepJournal.step_seq.in_([2, 3]), + AwoooPRunStepJournal.result_status == "pending", + ) + .values( + output_hash=_sha256(_stable_json(terminal_receipt)), + compensation_json=terminal_receipt, + result_status="failed", + error_code=error_code, + was_blocked=True, + block_reason="authenticated_outcome_timeout_no_write", + completed_at=now, + ) + ) + except Exception as exc: + logger.warning( + "agent99_outcome_timeout_terminal_failed_fail_closed", + incident_id=identity.incident_id, + run_id=str(identity.run_id), + error=str(exc), + ) + return { + "status": "outcome_timeout_terminal_persistence_failed", + "receipt_persisted": False, + "runtime_closure_verified": False, + } + return { + **envelope, + "receipt_persisted": persisted, + "runtime_closure_verified": False, + } + async def record_learning_writeback( self, *, @@ -2236,6 +2430,7 @@ class PostgresAgent99DispatchLedger: AwoooPRunState.run_id, AwoooPRunState.state, AwoooPRunState.error_detail, + AwoooPRunState.timeout_at, ) .where( AwoooPRunState.project_id == (project_id or "awoooi"), @@ -2309,6 +2504,7 @@ class PostgresAgent99DispatchLedger: "identity": identity, "run_state": str(row.state), "receipt": envelope, + "timeout_at": getattr(row, "timeout_at", None), }) return items diff --git a/apps/api/tests/test_agent99_controlled_dispatch_reconciler_job.py b/apps/api/tests/test_agent99_controlled_dispatch_reconciler_job.py index b230d7722..fc5b707a5 100644 --- a/apps/api/tests/test_agent99_controlled_dispatch_reconciler_job.py +++ b/apps/api/tests/test_agent99_controlled_dispatch_reconciler_job.py @@ -101,6 +101,7 @@ async def test_reconciler_consumes_authenticated_outcome_and_closes_same_run( "status_reconcile_requested": 0, "status_reconcile_pending": 0, "external_no_write_reconciled": 0, + "outcome_timeout_terminalized": 0, "verifier_written": 1, "closed": 1, } @@ -130,6 +131,7 @@ async def test_reconciler_does_not_finalize_without_outcome(monkeypatch) -> None "status_reconcile_requested": 0, "status_reconcile_pending": 0, "external_no_write_reconciled": 0, + "outcome_timeout_terminalized": 0, "verifier_written": 0, "closed": 0, } diff --git a/apps/api/tests/test_agent99_same_run_reconcile.py b/apps/api/tests/test_agent99_same_run_reconcile.py index 44d246dbb..d987aba8f 100644 --- a/apps/api/tests/test_agent99_same_run_reconcile.py +++ b/apps/api/tests/test_agent99_same_run_reconcile.py @@ -2,6 +2,7 @@ from __future__ import annotations import json from copy import deepcopy +from datetime import UTC, datetime, timedelta from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock @@ -812,6 +813,184 @@ async def test_external_no_write_terminal_rejects_malformed_source_counts( assert result["receipt_persisted"] is False +def test_dispatch_timeout_elapsed_handles_naive_and_aware_datetimes() -> None: + past = datetime.now(UTC) - timedelta(minutes=1) + future = datetime.now(UTC) + timedelta(minutes=1) + + assert job._dispatch_timeout_elapsed(past) is True + assert job._dispatch_timeout_elapsed(past.replace(tzinfo=None)) is True + assert job._dispatch_timeout_elapsed(future) is False + assert job._dispatch_timeout_elapsed(None) is False + + +@pytest.mark.asyncio +async def test_reconciler_terminalizes_expired_run_without_outcome( + monkeypatch, +) -> None: + identity = build_agent99_dispatch_identity( + project_id="awoooi", + incident_id="INC-BRR-EXPIRED", + source_fingerprint="backup-restore:" + "b" * 64, + route_id="agent99:backup_health:Status", + work_item_id="agent99-dispatch:awoooi:INC-BRR-EXPIRED", + ) + + class TimeoutLedger(_Ledger): + def __init__(self) -> None: + super().__init__() + self.timeout_calls: list[object] = [] + + async def list_reconcilable(self, **_kwargs): # type: ignore[no-untyped-def] + return [{ + "identity": identity, + "receipt": self.list_receipt, + "timeout_at": datetime.now(UTC) - timedelta(minutes=1), + }] + + async def record_outcome_timeout_no_write_terminal( + self, + **kwargs, + ): # type: ignore[no-untyped-def] + self.timeout_calls.append(kwargs["identity"]) + return { + "status": "outcome_timeout_no_write_terminal", + "receipt_persisted": True, + "runtime_closure_verified": False, + } + + ledger = TimeoutLedger() + monkeypatch.setattr(job, "get_agent99_dispatch_ledger", lambda: ledger) + monkeypatch.setattr(job, "read_agent99_sre_outcome", lambda _run: None) + + result = await job.reconcile_agent99_controlled_dispatches_once() + + assert result["outcome_timeout_terminalized"] == 1 + assert result["verifier_written"] == 0 + assert ledger.timeout_calls == [identity] + + +@pytest.mark.asyncio +async def test_outcome_timeout_terminal_does_not_resolve_incident_or_replay( + monkeypatch, +) -> None: + envelope = build_agent99_dispatch_receipt_envelope( + identity=IDENTITY, + dispatch_receipt={ + "accepted": True, + "inbox_triggered": True, + "queue_accepted": True, + "dispatch_identity_matched": True, + "delivery_certainty": "delivered", + }, + controlled_apply_authorized=True, + ) + + class TimeoutDB: + def __init__(self) -> None: + self.call = 0 + self.statements: list[str] = [] + + async def execute(self, statement): # type: ignore[no-untyped-def] + self.call += 1 + self.statements.append(str(statement)) + if self.call == 1: + return _ScalarResult( + row=SimpleNamespace( + state="waiting_tool", + error_detail=json.dumps(envelope), + timeout_at=( + datetime.now(UTC).replace(tzinfo=None) + - timedelta(minutes=1) + ), + ) + ) + if self.call == 2: + return _ScalarResult(value=IDENTITY.run_id) + return _ScalarResult() + + db = TimeoutDB() + monkeypatch.setattr( + ledger_module, + "get_db_context", + lambda _project_id: _Context(db), + ) + + result = ( + await PostgresAgent99DispatchLedger() + .record_outcome_timeout_no_write_terminal(identity=IDENTITY) + ) + + assert result["receipt_persisted"] is True + assert result["run_terminal_state"] == "failed" + assert result["prior_controlled_apply_authorized"] is True + assert result["controlled_apply_authorized"] is False + assert result["terminalizer_runtime_write_performed"] is False + assert result["outcome_runtime_write_status"] == ( + "unknown_without_authenticated_outcome" + ) + assert result["transport_replayed"] is False + assert result["automation_execution_success"] is False + assert result["post_verifier_passed"] is False + assert result["runtime_closure_verified"] is False + assert result["incident_resolution_authorized"] is False + assert result["learning_writeback"]["status"] == ( + "not_applicable_outcome_timeout_no_write" + ) + sql = "\n".join(db.statements).lower() + assert "incident_records" not in sql + assert "alert_operation_logs" not in sql + assert "telegram" not in sql + + +@pytest.mark.asyncio +async def test_outcome_timeout_terminal_rejects_unexpired_run( + monkeypatch, +) -> None: + envelope = build_agent99_dispatch_receipt_envelope( + identity=IDENTITY, + dispatch_receipt={ + "accepted": True, + "inbox_triggered": True, + "queue_accepted": True, + "dispatch_identity_matched": True, + }, + controlled_apply_authorized=False, + ) + + class FutureTimeoutDB: + def __init__(self) -> None: + self.call = 0 + + async def execute(self, _statement): # type: ignore[no-untyped-def] + self.call += 1 + return _ScalarResult( + row=SimpleNamespace( + state="waiting_tool", + error_detail=json.dumps(envelope), + timeout_at=( + datetime.now(UTC).replace(tzinfo=None) + + timedelta(minutes=1) + ), + ) + ) + + db = FutureTimeoutDB() + monkeypatch.setattr( + ledger_module, + "get_db_context", + lambda _project_id: _Context(db), + ) + + result = ( + await PostgresAgent99DispatchLedger() + .record_outcome_timeout_no_write_terminal(identity=IDENTITY) + ) + + assert result["status"] == "outcome_timeout_not_due_no_write" + assert result["receipt_persisted"] is False + assert db.call == 1 + + def test_windows_relay_and_control_plane_enforce_same_run_no_write_contract() -> None: root = Path(__file__).resolve().parents[3] relay = (root / "agent99-sre-alert-relay.ps1").read_text(encoding="utf-8") diff --git a/apps/api/tests/test_config_url_validation.py b/apps/api/tests/test_config_url_validation.py index 27489672d..86269675b 100644 --- a/apps/api/tests/test_config_url_validation.py +++ b/apps/api/tests/test_config_url_validation.py @@ -172,3 +172,4 @@ def test_database_pool_budget_defaults_to_production_safe_values(): assert s.DATABASE_MAX_OVERFLOW == 0 assert s.DATABASE_POOL_TIMEOUT_SECONDS == 5.0 assert s.DATABASE_NULL_POOL is False + assert s.DATABASE_NULL_POOL_CONCURRENCY_LIMIT == 0 diff --git a/apps/api/tests/test_runtime_bootstrap_guards.py b/apps/api/tests/test_runtime_bootstrap_guards.py index d42a70e7f..60b6f52df 100644 --- a/apps/api/tests/test_runtime_bootstrap_guards.py +++ b/apps/api/tests/test_runtime_bootstrap_guards.py @@ -8,6 +8,7 @@ Runtime bootstrap guard tests. from __future__ import annotations +import asyncio import inspect from collections.abc import Awaitable from pathlib import Path @@ -223,6 +224,52 @@ def test_get_engine_uses_null_pool_for_bounded_background_processes(monkeypatch) assert "pool_timeout" not in captured +@pytest.mark.asyncio +async def test_null_pool_session_limiter_serializes_worker_db_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, + "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) + first_entered = asyncio.Event() + release_first = asyncio.Event() + second_entered = asyncio.Event() + + async def first_session() -> None: + async with db_base._database_session_slot(): + first_entered.set() + await release_first.wait() + + async def second_session() -> None: + async with db_base._database_session_slot(): + second_entered.set() + + first_task = asyncio.create_task(first_session()) + await first_entered.wait() + second_task = asyncio.create_task(second_session()) + await asyncio.sleep(0) + + assert second_entered.is_set() is False + + release_first.set() + await asyncio.gather(first_task, second_task) + assert second_entered.is_set() is True + + @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 1d5802b05..7eaf00e8a 100644 --- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py +++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py @@ -91,6 +91,8 @@ def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None: assert worker_env["MCP_FEDERATION_STARTUP_SLEEP_SECONDS"] == "45" assert worker_env["MCP_FEDERATION_SOURCE_TIMEOUT_SECONDS"] == "10" assert worker_env["AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS"] == "168" + assert worker_env["DATABASE_NULL_POOL"] == "true" + assert worker_env["DATABASE_NULL_POOL_CONCURRENCY_LIMIT"] == "1" def test_execution_broker_is_only_workload_with_ssh_transport() -> None: diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml index f9510f202..842809ab7 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -31,9 +31,9 @@ spec: strategy: type: RollingUpdate rollingUpdate: - # 2026-07-01 Codex: DB role `awoooi` is capped at two connections and - # worker imports the same API DB stack. Do not run old+new worker pods - # during rollout until the DB role/pool budget is raised and verified. + # 2026-07-15 Codex: keep one worker owner during rollout. This prevents + # duplicate consumers and keeps the isolated worker DB role inside its + # verified connection budget while the old pod terminates. maxSurge: 0 maxUnavailable: 1 template: @@ -163,13 +163,17 @@ spec: fieldRef: fieldPath: metadata.name - name: DATABASE_POOL_SIZE - # 2026-07-01 Codex: keep worker DB usage inside the current - # production role connection budget during reboot rollouts. + # NullPool releases each session; the explicit concurrency cap + # below preserves the intended one-connection worker budget. value: "1" - name: DATABASE_MAX_OVERFLOW value: "0" - name: DATABASE_NULL_POOL value: "true" + - name: DATABASE_NULL_POOL_CONCURRENCY_LIMIT + # NullPool releases each connection after use, while this cap + # keeps concurrent background loops inside the worker role budget. + value: "1" - name: DATABASE_BOOTSTRAP_DDL_ENABLED value: "false" - name: AWOOOI_BUILD_COMMIT_SHA