From 21c1fa49b6f069a3139f07c895045ce3eeff56b7 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Mon, 3 Aug 2026 00:20:51 +0200 Subject: [PATCH] Add worker delivery acceptance probe --- docs/DEPLOYMENT_OPERATOR_GUIDE.md | 8 ++++ src/govoplan_core/celery_app.py | 72 +++++++++++++++++++++++++++++ src/govoplan_core/settings.py | 6 +++ tests/test_celery_queue_contract.py | 29 +++++++++++- 4 files changed, 113 insertions(+), 2 deletions(-) diff --git a/docs/DEPLOYMENT_OPERATOR_GUIDE.md b/docs/DEPLOYMENT_OPERATOR_GUIDE.md index 10d02d0..ea12cc7 100644 --- a/docs/DEPLOYMENT_OPERATOR_GUIDE.md +++ b/docs/DEPLOYMENT_OPERATOR_GUIDE.md @@ -158,6 +158,7 @@ release evidence. | `REDIS_URL` | `redis://redis:6379/0` | Celery broker/result backend when async workers are enabled. | | `CELERY_ENABLED` | `false` | Local/dev can send synchronously. Production campaign delivery should run workers and set this to `true`. | | `CELERY_QUEUES` | `send_email,append_sent,notifications,mail,calendar,dataflow,workflow,postbox,events,idm,default` | Queue list expected by worker/process manager definitions. Keep this aligned with every enabled module task route; Ops reports missing worker queue consumers. | +| `CELERY_VISIBILITY_TIMEOUT_SECONDS` | `3600` | Maximum time before Redis may redeliver work left unacknowledged by a lost worker. Set this above the longest supported task duration; changing it requires a worker-loss acceptance drill. | | `PLATFORM_EVENT_OUTBOX_MAX_ATTEMPTS` | `8` | Failed durable consumer deliveries are quarantined after this many attempts. | | `PLATFORM_EVENT_OUTBOX_TERMINAL_RETENTION_DAYS` | `90` | Successful event envelopes older than this are removed by the daily retention task. Quarantined evidence is retained. | @@ -177,6 +178,13 @@ crashes, and expired worker leases: python -m celery -A govoplan_core.celery_app:celery beat --loglevel INFO ``` +Before promoting a worker composition, run the repository worker-runtime drill +against the same Redis and Core build. It uses the bounded +`govoplan.worker.acceptance` task and records publish/consume, retry, warm +SIGTERM, and worker-loss redelivery evidence without accessing tenant data. +Production evidence must use the deployed queue configuration and a visibility +timeout that is longer than every supported business task. + ### Storage | Setting | Default | Notes | diff --git a/src/govoplan_core/celery_app.py b/src/govoplan_core/celery_app.py index c1d8e53..70463ac 100644 --- a/src/govoplan_core/celery_app.py +++ b/src/govoplan_core/celery_app.py @@ -4,6 +4,7 @@ from datetime import datetime, timedelta, timezone from importlib.metadata import PackageNotFoundError, version import logging import os +import time from celery import Celery from celery.signals import ( @@ -110,6 +111,14 @@ celery.conf.update( worker_prefetch_multiplier=1, task_acks_late=True, task_reject_on_worker_lost=True, + task_track_started=True, + broker_transport_options={ + "visibility_timeout": settings.celery_visibility_timeout_seconds, + }, + result_backend_transport_options={ + "visibility_timeout": settings.celery_visibility_timeout_seconds, + }, + visibility_timeout=settings.celery_visibility_timeout_seconds, beat_schedule={ "calendar-outbox-every-minute": { "task": "govoplan.calendar.dispatch_outbox", @@ -308,6 +317,69 @@ def ping(): return "pong" +@celery.task( + bind=True, + name="govoplan.worker.acceptance", + max_retries=1, + acks_late=True, + reject_on_worker_lost=True, + track_started=True, +) +def worker_acceptance_probe( + task, + probe_id: str, + *, + mode: str = "complete", + delay_seconds: float = 0.0, + track_delivery: bool = False, +): + """Exercise broker delivery semantics without touching business data.""" + + if mode not in {"complete", "retry_once"}: + raise ValueError(f"Unsupported worker acceptance mode: {mode}") + normalized_probe_id = str(probe_id).strip() + if not normalized_probe_id or len(normalized_probe_id) > 120: + raise ValueError("Worker acceptance probe_id must contain 1-120 characters") + bounded_delay = max(0.0, min(float(delay_seconds), 120.0)) + delivery_count = _worker_acceptance_delivery_count( + task, + normalized_probe_id, + ) if track_delivery else None + task.update_state( + state="PROGRESS", + meta={ + "phase": "started", + "probe_id": normalized_probe_id, + "retries": int(task.request.retries or 0), + "delivery_count": delivery_count, + }, + ) + if mode == "retry_once" and int(task.request.retries or 0) == 0: + raise task.retry(countdown=0.2) + if bounded_delay: + time.sleep(bounded_delay) + delivery_info = task.request.delivery_info or {} + return { + "probe_id": normalized_probe_id, + "mode": mode, + "retries": int(task.request.retries or 0), + "redelivered": bool(delivery_info.get("redelivered")), + "delivery_count": delivery_count, + } + + +def _worker_acceptance_delivery_count(task, probe_id: str) -> int: + client = getattr(task.backend, "client", None) + if client is None: + raise RuntimeError( + "Worker acceptance delivery tracking requires a Redis result backend" + ) + key = f"govoplan:worker-acceptance:{probe_id}" + count = int(client.incr(key)) + client.expire(key, 60 * 60) + return count + + def _platform_registry() -> PlatformRegistry: raw_enabled_modules = load_startup_enabled_modules(settings.enabled_modules) candidate_modules = startup_candidate_module_ids( diff --git a/src/govoplan_core/settings.py b/src/govoplan_core/settings.py index fb62b18..20730d8 100644 --- a/src/govoplan_core/settings.py +++ b/src/govoplan_core/settings.py @@ -115,6 +115,12 @@ class Settings(BaseSettings): migration_track: str = Field(default="release", alias="GOVOPLAN_MIGRATION_TRACK") redis_url: str = Field(default="redis://redis:6379/0", alias="REDIS_URL") celery_enabled: bool = Field(default=False, alias="CELERY_ENABLED") + celery_visibility_timeout_seconds: int = Field( + default=3600, + ge=30, + le=7 * 24 * 60 * 60, + alias="CELERY_VISIBILITY_TIMEOUT_SECONDS", + ) s3_endpoint_url: str = Field(default="http://garage:3900", alias="S3_ENDPOINT_URL") s3_region: str = Field(default="garage", alias="S3_REGION") diff --git a/tests/test_celery_queue_contract.py b/tests/test_celery_queue_contract.py index d9d7390..b75ccd8 100644 --- a/tests/test_celery_queue_contract.py +++ b/tests/test_celery_queue_contract.py @@ -1,8 +1,9 @@ from __future__ import annotations import unittest +from unittest.mock import patch -from govoplan_core.celery_app import celery +from govoplan_core.celery_app import celery, worker_acceptance_probe from govoplan_core.settings import settings @@ -26,9 +27,33 @@ class CeleryQueueContractTests(unittest.TestCase): def test_delivery_tasks_keep_worker_loss_protection(self) -> None: self.assertTrue(celery.conf.task_acks_late) self.assertTrue(celery.conf.task_reject_on_worker_lost) + self.assertTrue(celery.conf.task_track_started) self.assertEqual(1, celery.conf.worker_prefetch_multiplier) + self.assertEqual( + settings.celery_visibility_timeout_seconds, + celery.conf.broker_transport_options["visibility_timeout"], + ) + + def test_worker_acceptance_probe_is_bounded_and_side_effect_free(self) -> None: + with patch.object(worker_acceptance_probe, "update_state") as update_state: + result = worker_acceptance_probe.run( + "probe-1", + mode="complete", + delay_seconds=0, + ) + + self.assertEqual( + { + "probe_id": "probe-1", + "mode": "complete", + "retries": 0, + "redelivered": False, + "delivery_count": None, + }, + result, + ) + update_state.assert_called_once() if __name__ == "__main__": unittest.main() -