diff --git a/docs/STATE_AND_RECOVERY_CONTRACT.md b/docs/STATE_AND_RECOVERY_CONTRACT.md index b99ab67..0f9b87e 100644 --- a/docs/STATE_AND_RECOVERY_CONTRACT.md +++ b/docs/STATE_AND_RECOVERY_CONTRACT.md @@ -52,6 +52,13 @@ software version, module-composition hash, queues, start time, and heartbeat. The registration identity includes a process incarnation so a stale process cannot update a replacement's row. +Worker metadata also records the orchestrator pool and declared concurrency. +Every Celery prefork child disposes the SQLAlchemy pool inherited from its +parent and creates a process-local pool before handling work. Deployment +rendering must therefore budget one database pool for the worker parent and +each child. Ops compares active queue ownership, software versions, and the +order-independent module-composition hash with the graph loaded by the API. + Drain is durable operator intent: - an API enters not-ready state after observing drain; diff --git a/src/govoplan_core/celery_app.py b/src/govoplan_core/celery_app.py index e6f5b36..c1d8e53 100644 --- a/src/govoplan_core/celery_app.py +++ b/src/govoplan_core/celery_app.py @@ -3,9 +3,15 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone from importlib.metadata import PackageNotFoundError, version import logging +import os from celery import Celery -from celery.signals import heartbeat_sent, worker_ready, worker_shutdown +from celery.signals import ( + heartbeat_sent, + worker_process_init, + worker_ready, + worker_shutdown, +) from govoplan_core.core.campaigns import ( CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, @@ -196,6 +202,21 @@ def _worker_runtime_identity(sender: object | None = None) -> RuntimeIdentity: return _worker_identity +def _worker_metadata() -> dict[str, object]: + raw_concurrency = str(os.getenv("CELERY_WORKER_CONCURRENCY") or "").strip() + return { + "process": "celery-worker", + "worker_pool": str(os.getenv("GOVOPLAN_WORKER_POOL") or "default"), + "concurrency": int(raw_concurrency) if raw_concurrency.isdigit() else None, + } + + +@worker_process_init.connect +def _reset_worker_process_database(**_kwargs) -> None: + # SQLAlchemy pools must not be shared across prefork child processes. + configure_database(settings.database_url, dispose_previous=True) + + @worker_ready.connect def _register_worker_runtime(sender=None, **_kwargs) -> None: global _worker_consumer @@ -206,7 +227,7 @@ def _register_worker_runtime(sender=None, **_kwargs) -> None: node = register_runtime_node( session, identity, - metadata={"process": "celery-worker"}, + metadata=_worker_metadata(), ) session.commit() except Exception: @@ -227,7 +248,7 @@ def _heartbeat_worker_runtime(sender=None, **_kwargs) -> None: node = heartbeat_runtime_node( session, identity, - metadata={"process": "celery-worker"}, + metadata=_worker_metadata(), ) session.commit() except Exception: # noqa: BLE001 - uncertain authority must fail closed diff --git a/src/govoplan_core/core/runtime_coordination.py b/src/govoplan_core/core/runtime_coordination.py index 629b09e..350de5f 100644 --- a/src/govoplan_core/core/runtime_coordination.py +++ b/src/govoplan_core/core/runtime_coordination.py @@ -166,9 +166,7 @@ def runtime_identity( for item in str(getattr(settings, "celery_queues", "") or "").split(",") if item.strip() ) - composition_hash = hashlib.sha256( - json.dumps(sorted(module_ids), separators=(",", ":")).encode("utf-8") - ).hexdigest() + composition_hash = runtime_composition_hash(module_ids) return RuntimeIdentity( installation_id=installation_id, node_id=effective_node_id, @@ -180,6 +178,12 @@ def runtime_identity( ) +def runtime_composition_hash(module_ids: tuple[str, ...]) -> str: + return hashlib.sha256( + json.dumps(sorted(set(module_ids)), separators=(",", ":")).encode("utf-8") + ).hexdigest() + + def register_runtime_node( session: Session, identity: RuntimeIdentity, diff --git a/src/govoplan_core/db/session.py b/src/govoplan_core/db/session.py index 6b08159..2c0b2a9 100644 --- a/src/govoplan_core/db/session.py +++ b/src/govoplan_core/db/session.py @@ -91,7 +91,12 @@ _default_database: DatabaseHandle | None = None def configure_database(database_url: str, *, engine: Engine | None = None, dispose_previous: bool = False) -> DatabaseHandle: global _default_database - if engine is None and _default_database is not None and _default_database.database_url == database_url: + if ( + engine is None + and not dispose_previous + and _default_database is not None + and _default_database.database_url == database_url + ): return _default_database previous_database = _default_database _default_database = DatabaseHandle(database_url, engine=engine) diff --git a/src/govoplan_core/settings.py b/src/govoplan_core/settings.py index 211dfcd..16566fa 100644 --- a/src/govoplan_core/settings.py +++ b/src/govoplan_core/settings.py @@ -73,6 +73,30 @@ class Settings(BaseSettings): ge=0, le=86_400, ) + database_connection_limit: int | None = Field( + default=None, + alias="GOVOPLAN_DB_CONNECTION_LIMIT", + ge=10, + le=1_000_000, + ) + database_connection_reserve: int = Field( + default=10, + alias="GOVOPLAN_DB_CONNECTION_RESERVE", + ge=1, + le=999_999, + ) + database_connection_peak: int | None = Field( + default=None, + alias="GOVOPLAN_DB_CONNECTION_PEAK", + ge=1, + le=1_000_000, + ) + database_connection_available: int | None = Field( + default=None, + alias="GOVOPLAN_DB_CONNECTION_AVAILABLE", + ge=1, + le=1_000_000, + ) access_database_url: str | None = Field(default=None, alias="ACCESS_DATABASE_URL") access_db_schema: str | None = Field(default=None, alias="ACCESS_DB_SCHEMA") access_table_prefix: str = Field(default="access_", alias="ACCESS_TABLE_PREFIX") diff --git a/tests/test_runtime_agents.py b/tests/test_runtime_agents.py index 9e34045..80e1082 100644 --- a/tests/test_runtime_agents.py +++ b/tests/test_runtime_agents.py @@ -123,3 +123,20 @@ def test_worker_disables_consumers_without_reclaiming_stale_identity( celery_app._worker_consumer, celery_app._worker_draining, ) = original_state + + +def test_worker_child_replaces_inherited_database_pool(monkeypatch) -> None: + from govoplan_core import celery_app + + calls: list[tuple[str, bool]] = [] + monkeypatch.setattr( + celery_app, + "configure_database", + lambda url, *, dispose_previous=False: calls.append( + (url, dispose_previous) + ), + ) + + celery_app._reset_worker_process_database() + + assert calls == [(celery_app.settings.database_url, True)] diff --git a/tests/test_runtime_coordination.py b/tests/test_runtime_coordination.py index 7143425..51596d1 100644 --- a/tests/test_runtime_coordination.py +++ b/tests/test_runtime_coordination.py @@ -18,6 +18,7 @@ from govoplan_core.core.runtime_coordination import ( list_runtime_nodes, register_runtime_node, request_runtime_node_drain, + runtime_composition_hash, ) from govoplan_core.db.base import Base @@ -128,3 +129,9 @@ def test_expired_lease_reassignment_increments_fence() -> None: finally: session.close() engine.dispose() + + +def test_runtime_composition_hash_is_order_and_duplicate_independent() -> None: + assert runtime_composition_hash(("files", "core", "files")) == ( + runtime_composition_hash(("core", "files")) + )