Automate worker runtime delivery drill

.gitea/workflows/module-matrix.yml

@@ -21,6 +21,13 @@ jobs:
           --health-interval 5s
           --health-timeout 5s
           --health-retries 20
+      redis:
+        image: redis:7-alpine
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 20
     steps:
       - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5
         with:
@@ -66,6 +73,13 @@ jobs:
             -s ../govoplan-search/tests \
             -p test_postgres_search.py \
             -v
+      - name: Prove worker delivery and shutdown guarantees
+        working-directory: govoplan
+        env:
+          GOVOPLAN_WORKER_DRILL_REDIS_URL: redis://redis:6379/15
+        run: |
+          .venv/bin/python tools/checks/worker-runtime-drill.py \
+            --output audit-reports/worker-runtime.json
       - name: Run module matrix and contract tests
         working-directory: govoplan
         run: GOVOPLAN_CORE_ROOT="$PWD/../govoplan-core" PYTHON="$PWD/.venv/bin/python" bash tools/checks/check-module-matrix.sh

docs/SCALING_AND_MULTI_HOST_DEPLOYMENT.md

@@ -203,6 +203,30 @@ redelivery, scheduler failover, migration exclusion, object-store outage, and a
 coordinated database/object/key restore. Recovery rules and evidence are
 defined in [Recovery And Rollback Guarantees](RECOVERY_AND_ROLLBACK_GUARANTEES.md).
 
+## Worker Delivery Evidence
+
+The module-matrix workflow runs `tools/checks/worker-runtime-drill.py` against a
+real isolated Redis database. The drill starts supervised Celery worker
+processes and records four guarantees without accessing tenant data:
+
+1. a task published through the broker is consumed exactly once;
+2. an application retry is delivered again and completes;
+3. warm `SIGTERM` lets an in-flight late-ack task complete before shutdown; and
+4. loss of a worker after task start causes the unacknowledged task to be
+   redelivered after the configured visibility timeout.
+
+Run the same drill with the release Python environment and target Redis before
+promoting a worker composition. Use a dedicated Redis database, retain the JSON
+evidence, and set `CELERY_VISIBILITY_TIMEOUT_SECONDS` above the longest supported
+business-task duration. The short visibility timeout used by CI is an isolated
+test setting, not a production recommendation.
+
+```bash
+GOVOPLAN_WORKER_DRILL_REDIS_URL=redis://redis.example.test:6379/15 \
+  .venv/bin/python tools/checks/worker-runtime-drill.py \
+  --output evidence/worker-runtime.json
+```
+
 ## Live Multi-Host Evidence
 
 After deploying a pinned release on at least two Kubernetes nodes, create an API

tests/test_worker_runtime_drill.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+import sys
+import unittest
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / "tools" / "checks" / "worker-runtime-drill.py"
+SPEC = importlib.util.spec_from_file_location("worker_runtime_drill", SCRIPT)
+assert SPEC is not None and SPEC.loader is not None
+MODULE = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = MODULE
+SPEC.loader.exec_module(MODULE)
+
+
+class WorkerRuntimeDrillTests(unittest.TestCase):
+    def test_redacts_redis_credentials_and_query(self) -> None:
+        self.assertEqual(
+            "rediss://redis.example.test:6380/9",
+            MODULE._redacted_redis_url(
+                "rediss://worker:secret@redis.example.test:6380/9?ssl=true"
+            ),
+        )
+
+    def test_worker_command_uses_solo_default_queue_for_deterministic_drill(self) -> None:
+        command = MODULE._worker_command("/usr/bin/python", "worker-a@%h")
+
+        self.assertEqual("/usr/bin/python", command[0])
+        self.assertIn("solo", command)
+        self.assertIn("default", command)
+        self.assertIn("worker-a@%h", command)
+
+    def test_rejects_implicit_or_non_redis_broker(self) -> None:
+        args = MODULE.build_parser().parse_args(["--redis-url", "memory://"])
+
+        with self.assertRaisesRegex(ValueError, "explicit redis"):
+            MODULE.run_drill(args)
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/checks/worker-runtime-drill.py

@@ -0,0 +1,397 @@
+#!/usr/bin/env python3
+"""Exercise GovOPlaN worker delivery guarantees against a real Redis broker."""
+
+from __future__ import annotations
+
+import argparse
+from datetime import UTC, datetime
+import json
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+from typing import Any
+from urllib.parse import urlsplit, urlunsplit
+import uuid
+
+
+TERMINAL_STATES = {"FAILURE", "REVOKED", "SUCCESS"}
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Prove Celery publish/consume, retry, warm shutdown, and worker-loss "
+            "redelivery against an isolated Redis database."
+        )
+    )
+    parser.add_argument(
+        "--redis-url",
+        default=os.environ.get("GOVOPLAN_WORKER_DRILL_REDIS_URL", ""),
+    )
+    parser.add_argument("--python", default=sys.executable)
+    parser.add_argument("--timeout-seconds", type=float, default=120.0)
+    parser.add_argument(
+        "--visibility-timeout-seconds",
+        type=int,
+        default=30,
+    )
+    parser.add_argument("--output", type=Path)
+    return parser
+
+
+def _redacted_redis_url(value: str) -> str:
+    parsed = urlsplit(value)
+    hostname = parsed.hostname or ""
+    port = f":{parsed.port}" if parsed.port else ""
+    return urlunsplit((parsed.scheme, f"{hostname}{port}", parsed.path, "", ""))
+
+
+def _wait_until(predicate, *, timeout_seconds: float, detail: str):
+    deadline = time.monotonic() + timeout_seconds
+    last_value: Any = None
+    while time.monotonic() < deadline:
+        last_value = predicate()
+        if last_value:
+            return last_value
+        time.sleep(0.2)
+    raise TimeoutError(f"Timed out waiting for {detail}; last value: {last_value!r}")
+
+
+def _worker_command(python: str, hostname: str) -> list[str]:
+    return [
+        python,
+        "-m",
+        "celery",
+        "-A",
+        "govoplan_core.celery_app:celery",
+        "worker",
+        "--pool",
+        "solo",
+        "--queues",
+        "default",
+        "--hostname",
+        hostname,
+        "--loglevel",
+        "WARNING",
+        "--without-mingle",
+    ]
+
+
+def _start_worker(
+    *,
+    python: str,
+    hostname: str,
+    environment: dict[str, str],
+    log_path: Path,
+) -> tuple[subprocess.Popen[bytes], Any]:
+    log = log_path.open("ab", buffering=0)
+    process = subprocess.Popen(
+        _worker_command(python, hostname),
+        env=environment,
+        stdin=subprocess.DEVNULL,
+        stdout=log,
+        stderr=subprocess.STDOUT,
+        start_new_session=True,
+    )
+    return process, log
+
+
+def _stop_worker(
+    process: subprocess.Popen[bytes] | None,
+    log: Any,
+    *,
+    timeout_seconds: float = 30.0,
+) -> None:
+    if process is not None and process.poll() is None:
+        process.send_signal(signal.SIGTERM)
+        try:
+            process.wait(timeout=timeout_seconds)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait(timeout=10)
+    if log is not None:
+        log.close()
+
+
+def _wait_for_worker(app, process: subprocess.Popen[bytes], prefix: str, timeout: float) -> str:
+    def ping() -> str | None:
+        if process.poll() is not None:
+            raise RuntimeError(f"Worker exited before readiness with code {process.returncode}")
+        replies = app.control.ping(timeout=1.0) or []
+        for reply in replies:
+            for hostname, payload in reply.items():
+                if hostname.startswith(prefix) and payload.get("ok") == "pong":
+                    return hostname
+        return None
+
+    return _wait_until(ping, timeout_seconds=timeout, detail=f"worker {prefix}")
+
+
+def _wait_for_started(result, *, timeout_seconds: float) -> dict[str, Any]:
+    def started() -> dict[str, Any] | None:
+        state = result.state
+        if state in TERMINAL_STATES and state != "SUCCESS":
+            raise RuntimeError(f"Probe {result.id} became {state}: {result.result!r}")
+        info = result.info
+        if state in {"PROGRESS", "STARTED"} and isinstance(info, dict):
+            return dict(info)
+        return None
+
+    return _wait_until(
+        started,
+        timeout_seconds=timeout_seconds,
+        detail=f"probe {result.id} to start",
+    )
+
+
+def _wait_for_result(result, *, timeout_seconds: float) -> dict[str, Any]:
+    value = result.get(timeout=timeout_seconds, propagate=True, disable_sync_subtasks=False)
+    if not isinstance(value, dict):
+        raise RuntimeError(f"Probe {result.id} returned an invalid result: {value!r}")
+    return dict(value)
+
+
+def _publish(probe_task, probe_id: str, *, mode: str, delay_seconds: float):
+    return probe_task.apply_async(
+        args=(probe_id,),
+        kwargs={
+            "mode": mode,
+            "delay_seconds": delay_seconds,
+            "track_delivery": True,
+        },
+        queue="default",
+    )
+
+
+def _assert_probe(value: dict[str, Any], *, probe_id: str) -> None:
+    if value.get("probe_id") != probe_id:
+        raise RuntimeError(f"Probe identity mismatch: {value!r}")
+
+
+def run_drill(args: argparse.Namespace) -> dict[str, Any]:
+    redis_url = str(args.redis_url or "").strip()
+    parsed_redis = urlsplit(redis_url)
+    if parsed_redis.scheme not in {"redis", "rediss"} or not parsed_redis.hostname:
+        raise ValueError("--redis-url must be an explicit redis:// or rediss:// URL")
+    if args.visibility_timeout_seconds < 30:
+        raise ValueError("--visibility-timeout-seconds must be at least 30")
+
+    run_id = uuid.uuid4().hex
+    started_at = datetime.now(UTC)
+    with tempfile.TemporaryDirectory(prefix="govoplan-worker-drill-") as directory:
+        root = Path(directory)
+        database_url = f"sqlite:///{root / 'runtime.db'}"
+        environment = dict(os.environ)
+        environment.update(
+            {
+                "REDIS_URL": redis_url,
+                "CELERY_ENABLED": "true",
+                "CELERY_QUEUES": "default",
+                "CELERY_VISIBILITY_TIMEOUT_SECONDS": str(
+                    args.visibility_timeout_seconds
+                ),
+                "DATABASE_URL": database_url,
+                "ENABLED_MODULES": "access",
+                "APP_ENV": "development",
+                "GOVOPLAN_EXPECTED_WORKER_REPLICAS": "1",
+                "GOVOPLAN_WORKER_POOL": "acceptance-drill",
+            }
+        )
+        os.environ.update(environment)
+
+        from govoplan_core.db.migrations import migrate_database
+
+        migrate_database(
+            database_url=database_url,
+            enabled_modules=("access",),
+        )
+        from govoplan_core.celery_app import celery, worker_acceptance_probe
+
+        celery.backend.client.ping()
+        celery.control.purge()
+        worker: subprocess.Popen[bytes] | None = None
+        worker_log: Any = None
+        evidence: dict[str, Any] = {
+            "schema_version": "1.0",
+            "run_id": run_id,
+            "started_at": started_at.isoformat(),
+            "redis": _redacted_redis_url(redis_url),
+            "visibility_timeout_seconds": args.visibility_timeout_seconds,
+            "checks": [],
+        }
+        try:
+            prefix = f"govoplan-drill-{run_id[:8]}-a@"
+            worker, worker_log = _start_worker(
+                python=args.python,
+                hostname=prefix + "%h",
+                environment=environment,
+                log_path=root / "worker-a.log",
+            )
+            hostname = _wait_for_worker(
+                celery,
+                worker,
+                prefix,
+                args.timeout_seconds,
+            )
+            evidence["checks"].append(
+                {"id": "worker_startup", "state": "passed", "worker": hostname}
+            )
+
+            probe_id = f"{run_id}-publish"
+            result = _publish(
+                worker_acceptance_probe,
+                probe_id,
+                mode="complete",
+                delay_seconds=0,
+            )
+            value = _wait_for_result(result, timeout_seconds=args.timeout_seconds)
+            _assert_probe(value, probe_id=probe_id)
+            if value.get("delivery_count") != 1:
+                raise RuntimeError(f"Publish probe was not delivered exactly once: {value!r}")
+            evidence["checks"].append(
+                {
+                    "id": "publish_consume",
+                    "state": "passed",
+                    "task_id": result.id,
+                    "delivery_count": value["delivery_count"],
+                }
+            )
+
+            probe_id = f"{run_id}-retry"
+            result = _publish(
+                worker_acceptance_probe,
+                probe_id,
+                mode="retry_once",
+                delay_seconds=0,
+            )
+            value = _wait_for_result(result, timeout_seconds=args.timeout_seconds)
+            _assert_probe(value, probe_id=probe_id)
+            if value.get("retries") != 1 or value.get("delivery_count") != 2:
+                raise RuntimeError(f"Retry probe did not execute twice: {value!r}")
+            evidence["checks"].append(
+                {
+                    "id": "application_retry",
+                    "state": "passed",
+                    "task_id": result.id,
+                    "delivery_count": value["delivery_count"],
+                }
+            )
+
+            probe_id = f"{run_id}-warm"
+            result = _publish(
+                worker_acceptance_probe,
+                probe_id,
+                mode="complete",
+                delay_seconds=2,
+            )
+            _wait_for_started(result, timeout_seconds=args.timeout_seconds)
+            worker.send_signal(signal.SIGTERM)
+            value = _wait_for_result(result, timeout_seconds=args.timeout_seconds)
+            _assert_probe(value, probe_id=probe_id)
+            worker.wait(timeout=args.timeout_seconds)
+            if worker.returncode != 0 or value.get("delivery_count") != 1:
+                raise RuntimeError(
+                    f"Warm worker shutdown did not finish in-flight work: {value!r}"
+                )
+            worker_log.close()
+            worker = None
+            worker_log = None
+            evidence["checks"].append(
+                {
+                    "id": "graceful_shutdown",
+                    "state": "passed",
+                    "task_id": result.id,
+                    "delivery_count": value["delivery_count"],
+                }
+            )
+
+            prefix = f"govoplan-drill-{run_id[:8]}-b@"
+            worker, worker_log = _start_worker(
+                python=args.python,
+                hostname=prefix + "%h",
+                environment=environment,
+                log_path=root / "worker-b.log",
+            )
+            _wait_for_worker(celery, worker, prefix, args.timeout_seconds)
+            probe_id = f"{run_id}-loss"
+            result = _publish(
+                worker_acceptance_probe,
+                probe_id,
+                mode="complete",
+                delay_seconds=min(120.0, args.visibility_timeout_seconds + 20.0),
+            )
+            first_started = _wait_for_started(
+                result,
+                timeout_seconds=args.timeout_seconds,
+            )
+            if first_started.get("delivery_count") != 1:
+                raise RuntimeError(f"Worker-loss probe did not start once: {first_started!r}")
+            worker.kill()
+            worker.wait(timeout=10)
+            worker_log.close()
+            worker = None
+            worker_log = None
+
+            prefix = f"govoplan-drill-{run_id[:8]}-c@"
+            worker, worker_log = _start_worker(
+                python=args.python,
+                hostname=prefix + "%h",
+                environment=environment,
+                log_path=root / "worker-c.log",
+            )
+            _wait_for_worker(celery, worker, prefix, args.timeout_seconds)
+            value = _wait_for_result(
+                result,
+                timeout_seconds=args.timeout_seconds
+                + args.visibility_timeout_seconds
+                + 30,
+            )
+            _assert_probe(value, probe_id=probe_id)
+            if value.get("delivery_count") != 2:
+                raise RuntimeError(f"Worker-loss probe was not redelivered: {value!r}")
+            evidence["checks"].append(
+                {
+                    "id": "worker_loss_redelivery",
+                    "state": "passed",
+                    "task_id": result.id,
+                    "delivery_count": value["delivery_count"],
+                    "broker_redelivered": bool(value.get("redelivered")),
+                }
+            )
+        except BaseException as exc:
+            evidence["error"] = f"{type(exc).__name__}: {exc}"
+            evidence["result"] = {"state": "failed"}
+            raise
+        finally:
+            _stop_worker(worker, worker_log)
+            celery.control.purge()
+        evidence["completed_at"] = datetime.now(UTC).isoformat()
+        evidence["result"] = {"state": "passed"}
+        return evidence
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    try:
+        evidence = run_drill(args)
+    except (OSError, RuntimeError, TimeoutError, ValueError) as exc:
+        print(f"worker runtime drill failed: {exc}", file=sys.stderr)
+        return 1
+    payload = json.dumps(evidence, indent=2, sort_keys=True) + "\n"
+    if args.output:
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        temporary = args.output.with_suffix(args.output.suffix + ".tmp")
+        temporary.write_text(payload, encoding="utf-8")
+        temporary.replace(args.output)
+        print(f"Worker runtime evidence written to {args.output}")
+    else:
+        print(payload, end="")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())