feat: harden multi-host runtime coordination

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;

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

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(
+    composition_hash = runtime_composition_hash(module_ids)
-        json.dumps(sorted(module_ids), separators=(",", ":")).encode("utf-8")
-    ).hexdigest()
     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,

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)

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")

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)]

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"))
+    )