#!/usr/bin/env python3 """Prove Redis/Celery redelivery does not repeat an ambiguous SMTP effect. The default run starts an isolated Redis Compose service, two successive real Celery worker processes, and a controlled loopback SMTP endpoint. It kills the first worker after complete DATA but before a final SMTP response. The same unacknowledged broker task must be delivered to the replacement worker, which must freeze the unfinished durable attempt as ``outcome_unknown`` without a second SMTP connection or DATA transaction. """ from __future__ import annotations import argparse from collections import Counter from contextlib import contextmanager from dataclasses import dataclass, replace from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import re import shutil import socket import subprocess import sys import tempfile import time from typing import Any, Callable, Iterator, Mapping from uuid import uuid4 from redis import Redis from redis.exceptions import RedisError SCRIPT_ROOT = Path(__file__).resolve().parent REPOSITORY_ROOT = SCRIPT_ROOT.parents[1] if str(SCRIPT_ROOT) not in sys.path: sys.path.insert(0, str(SCRIPT_ROOT)) from run_campaign_acceptance import ( # noqa: E402 AcceptanceError, DEFAULT_FIXTURE, EXPECTED_AUDIT_ACTIONS, TestbedSettings, _assert_evidence_safe, _core_package_version, _durable_state_evidence, _expect, _report_evidence, create_mail_profile, prepare_campaign_scenario, required_composition_versions, smtp_fault_endpoint, ) EVIDENCE_SCHEMA = "govoplan.campaign.celery-redelivery-acceptance.v1" MAX_EVIDENCE_BYTES = 128 * 1024 DEFAULT_COMPOSE_FILE = SCRIPT_ROOT / "docker-compose.yml" TASK_RECEIVED_PATTERN = re.compile( r"Task govoplan[.]campaigns[.]send_email\[([0-9a-f-]{36})\] received", re.IGNORECASE, ) TASK_SUCCEEDED_PATTERN = re.compile( r"Task govoplan[.]campaigns[.]send_email\[([0-9a-f-]{36})\] succeeded", re.IGNORECASE, ) WORKER_BOOTSTRAP = r""" import os import sys from govoplan_core.celery_app import celery visibility_timeout = int(os.environ["GOVOPLAN_CAMPAIGN_TEST_REDIS_VISIBILITY_TIMEOUT_SECONDS"]) celery.conf.broker_transport_options = { **dict(celery.conf.broker_transport_options or {}), "polling_interval": 0.25, "visibility_timeout": visibility_timeout, } celery.worker_main( [ "worker", "--loglevel=INFO", "--pool=solo", "--concurrency=1", "--queues=send_email", f"--hostname={sys.argv[1]}@%h", "--without-gossip", "--without-mingle", "--without-heartbeat", ] ) """ @dataclass(slots=True) class WorkerProcess: process: subprocess.Popen[bytes] log_path: Path log_handle: Any def text(self) -> str: self.log_handle.flush() try: return self.log_path.read_text(encoding="utf-8", errors="replace") except OSError as exc: raise AcceptanceError("Celery worker evidence log could not be read") from exc def received_task_ids(self) -> tuple[str, ...]: return tuple(TASK_RECEIVED_PATTERN.findall(self.text())) def succeeded_task_ids(self) -> tuple[str, ...]: return tuple(TASK_SUCCEEDED_PATTERN.findall(self.text())) @dataclass(frozen=True, slots=True) class RedisBrokerState: queue_depth: int unacked_hash_count: int unacked_index_count: int def as_dict(self) -> dict[str, int]: return { "queue_depth": self.queue_depth, "unacked_hash_count": self.unacked_hash_count, "unacked_index_count": self.unacked_index_count, } def _positive_int(value: str, *, label: str) -> int: try: parsed = int(value) except ValueError as exc: raise argparse.ArgumentTypeError(f"{label} must be a positive integer") from exc if parsed <= 0: raise argparse.ArgumentTypeError(f"{label} must be a positive integer") return parsed def _unused_loopback_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: probe.bind(("127.0.0.1", 0)) return int(probe.getsockname()[1]) def _compose_command( *, compose_file: Path, project_name: str, operation: str ) -> list[str]: prefix = [ "docker", "compose", "--file", str(compose_file), "--project-name", project_name, ] if operation == "up": return [*prefix, "up", "--detach", "redis"] if operation == "down": return [*prefix, "down", "--volumes", "--remove-orphans"] raise AcceptanceError("Unsupported Redis Compose operation") def _run_compose( command: list[str], *, environment: Mapping[str, str], timeout_seconds: int, ) -> None: try: completed = subprocess.run( command, env=dict(environment), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout_seconds, check=False, ) except (OSError, subprocess.TimeoutExpired) as exc: raise AcceptanceError("Redis Compose lifecycle command failed") from exc if completed.returncode != 0: raise AcceptanceError("Redis Compose lifecycle command failed") def _wait_for_redis(redis_url: str, *, timeout_seconds: int) -> None: deadline = time.monotonic() + timeout_seconds client = Redis.from_url( redis_url, socket_connect_timeout=1, socket_timeout=1, decode_responses=False, ) try: while time.monotonic() < deadline: try: if client.ping() is True: return except RedisError: pass time.sleep(0.25) finally: client.close() raise AcceptanceError("Isolated Redis broker did not become ready") @contextmanager def isolated_redis_broker( *, compose_file: Path, timeout_seconds: int, requested_port: int | None = None, ) -> Iterator[str]: if shutil.which("docker") is None: raise AcceptanceError("Docker CLI is required to start the isolated Redis broker") if not compose_file.is_file(): raise AcceptanceError("Redis Compose definition is unavailable") port = requested_port or _unused_loopback_port() if port <= 0 or port > 65_535: raise AcceptanceError("Redis test port is invalid") project_name = f"govoplan-campaign-redelivery-{uuid4().hex[:12]}" environment = { **os.environ, "GOVOPLAN_CAMPAIGN_TEST_REDIS_PORT": str(port), } lifecycle_attempted = False try: lifecycle_attempted = True _run_compose( _compose_command( compose_file=compose_file, project_name=project_name, operation="up", ), environment=environment, timeout_seconds=timeout_seconds, ) redis_url = f"redis://127.0.0.1:{port}/0" _wait_for_redis(redis_url, timeout_seconds=timeout_seconds) yield redis_url finally: if lifecycle_attempted: _run_compose( _compose_command( compose_file=compose_file, project_name=project_name, operation="down", ), environment=environment, timeout_seconds=timeout_seconds, ) def _start_worker(runtime_root: Path, *, label: str) -> WorkerProcess: log_path = runtime_root / f"{label}.log" log_handle = log_path.open("wb") environment = {**os.environ, "PYTHONUNBUFFERED": "1"} try: process = subprocess.Popen( [sys.executable, "-c", WORKER_BOOTSTRAP, label], env=environment, stdin=subprocess.DEVNULL, stdout=log_handle, stderr=subprocess.STDOUT, close_fds=True, ) except Exception: log_handle.close() raise return WorkerProcess(process=process, log_path=log_path, log_handle=log_handle) def _wait_for_worker_ready(worker: WorkerProcess, *, timeout_seconds: int) -> None: deadline = time.monotonic() + timeout_seconds while time.monotonic() < deadline: if worker.process.poll() is not None: raise AcceptanceError("Celery worker exited before becoming ready") if " ready." in worker.text(): return time.sleep(0.2) raise AcceptanceError("Celery worker did not become ready") def _wait_for_received_task( worker: WorkerProcess, *, timeout_seconds: int, expected_task_id: str | None = None, ) -> str: deadline = time.monotonic() + timeout_seconds while time.monotonic() < deadline: received = worker.received_task_ids() if received: if len(set(received)) != 1: raise AcceptanceError("Celery worker received more than one task identity") task_id = received[0] if expected_task_id is not None and task_id != expected_task_id: raise AcceptanceError("Replacement worker received a different broker task") return task_id if worker.process.poll() is not None: raise AcceptanceError("Celery worker exited before receiving the task") time.sleep(0.2) raise AcceptanceError("Celery worker did not receive the broker task") def _wait_for_task_success( worker: WorkerProcess, *, task_id: str, timeout_seconds: int, ) -> None: deadline = time.monotonic() + timeout_seconds while time.monotonic() < deadline: succeeded = worker.succeeded_task_ids() if task_id in succeeded: return if worker.process.poll() is not None: raise AcceptanceError("Replacement Celery worker exited before task success") time.sleep(0.2) raise AcceptanceError("Redelivered Celery task did not complete") def _kill_worker(worker: WorkerProcess, *, timeout_seconds: int) -> int: if worker.process.poll() is not None: raise AcceptanceError("Celery worker exited before controlled termination") worker.process.kill() try: return_code = worker.process.wait(timeout=timeout_seconds) except subprocess.TimeoutExpired as exc: raise AcceptanceError("Celery worker could not be killed") from exc if return_code == 0: raise AcceptanceError("Celery worker termination was not forced") return return_code def _stop_worker(worker: WorkerProcess, *, timeout_seconds: int) -> None: if worker.process.poll() is None: worker.process.terminate() try: worker.process.wait(timeout=timeout_seconds) except subprocess.TimeoutExpired: worker.process.kill() worker.process.wait(timeout=timeout_seconds) worker.log_handle.close() def _broker_state(redis_url: str) -> RedisBrokerState: client = Redis.from_url( redis_url, socket_connect_timeout=2, socket_timeout=2, decode_responses=False, ) try: return RedisBrokerState( queue_depth=int(client.llen("send_email")), unacked_hash_count=int(client.hlen("unacked")), unacked_index_count=int(client.zcard("unacked_index")), ) except RedisError as exc: raise AcceptanceError("Redis broker state could not be inspected") from exc finally: client.close() def _wait_for_broker_drained( redis_url: str, *, timeout_seconds: int, ) -> RedisBrokerState: deadline = time.monotonic() + timeout_seconds last = RedisBrokerState(0, 0, 0) while time.monotonic() < deadline: last = _broker_state(redis_url) if last == RedisBrokerState(0, 0, 0): return last time.sleep(0.2) raise AcceptanceError("Redis broker retained delivery state after recovery") def _queue_evidence(payload: Mapping[str, Any]) -> dict[str, Any]: expected = { "queued_count": 1, "skipped_count": 0, "blocked_count": 0, "enqueued_count": 1, "delivery_mode": "worker_queue", "worker_queue_available": True, "dry_run": False, } evidence = {key: payload.get(key) for key in expected} if evidence != expected: raise AcceptanceError("Campaign was not durably queued to one Celery task") return evidence def execute_redelivery_scenario( client: Any, headers: Mapping[str, str], *, fixture_path: Path, settings: TestbedSettings, endpoint: Any, redis_url: str, runtime_root: Path, snapshot_probe: Callable[[str], tuple[Mapping[str, Any], Mapping[str, Any]]], audit_probe: Callable[[str, str], Mapping[str, int]], delivery_probe: Callable[[str, str], Mapping[str, Any]], ) -> dict[str, Any]: profile_id = create_mail_profile( client, headers, settings, name="Campaign Redis Celery redelivery drill", smtp_host=endpoint.host, smtp_port=endpoint.port, ) prepared = prepare_campaign_scenario( client, headers, fixture_path=fixture_path, profile_id=profile_id, settings=settings, scenario="celery_broker_redelivery", snapshot_probe=snapshot_probe, ) first_worker = _start_worker(runtime_root, label="first-worker") replacement_worker: WorkerProcess | None = None try: _wait_for_worker_ready( first_worker, timeout_seconds=settings.provider_timeout_seconds, ) queued = _expect( client.post( f"/api/v1/campaigns/{prepared.campaign_id}/queue", headers=dict(headers), json={ "version_id": prepared.version_id, "include_warnings": True, "enqueue_celery": True, "dry_run": False, }, ), 200, "Celery-redelivery Campaign queue", ) queue_evidence = _queue_evidence(queued) first_task_id = _wait_for_received_task( first_worker, timeout_seconds=settings.provider_timeout_seconds, ) if not endpoint.wait_for_data(settings.provider_timeout_seconds): raise AcceptanceError("Celery worker did not reach complete SMTP DATA") first_exit_code = _kill_worker( first_worker, timeout_seconds=settings.provider_timeout_seconds, ) endpoint.release_held_connection() interrupted_state = _durable_state_evidence( delivery_probe(prepared.campaign_id, prepared.version_id) ) expected_interrupted = { "job_count": 1, "send_status_counts": {"sending": 1}, "attempt_status_counts": {"smtp_in_progress": 1}, "unfinished_attempt_count": 1, } if interrupted_state != expected_interrupted: raise AcceptanceError("Killed worker state was not durably SMTP-in-progress") replacement_worker = _start_worker(runtime_root, label="replacement-worker") _wait_for_worker_ready( replacement_worker, timeout_seconds=settings.provider_timeout_seconds, ) redelivered_task_id = _wait_for_received_task( replacement_worker, timeout_seconds=settings.provider_timeout_seconds, expected_task_id=first_task_id, ) _wait_for_task_success( replacement_worker, task_id=redelivered_task_id, timeout_seconds=settings.provider_timeout_seconds, ) recovered_state = _durable_state_evidence( delivery_probe(prepared.campaign_id, prepared.version_id) ) expected_recovered = { "job_count": 1, "send_status_counts": {"outcome_unknown": 1}, "attempt_status_counts": {"outcome_unknown": 1}, "unfinished_attempt_count": 0, } if recovered_state != expected_recovered: raise AcceptanceError("Redelivered task did not freeze the unfinished attempt") protocol = endpoint.evidence() expected_protocol = { "connection_count": 1, "accepted_rcpt_commands": 1, "refused_rcpt_commands": 0, "data_transactions": 1, } if protocol != expected_protocol: raise AcceptanceError("Broker redelivery caused an unexpected SMTP transaction") broker_after = _wait_for_broker_drained( redis_url, timeout_seconds=settings.provider_timeout_seconds, ) first_received = first_worker.received_task_ids() replacement_received = replacement_worker.received_task_ids() if first_received != (first_task_id,) or replacement_received != ( redelivered_task_id, ): raise AcceptanceError( "Celery workers did not each receive the broker task exactly once" ) report = _report_evidence( _expect( client.get( f"/api/v1/campaigns/{prepared.campaign_id}/report", headers=dict(headers), params={"version_id": prepared.version_id}, ), 200, "Celery-redelivery Campaign report", ) ) if report["send_status_counts"] != {"outcome_unknown": 1}: raise AcceptanceError("Campaign report did not retain outcome_unknown") audit_actions = dict( sorted(audit_probe(prepared.campaign_id, prepared.version_id).items()) ) if not { "campaign.created", "campaign.validated", "campaign.messages_built", "campaign.queued", }.issubset(audit_actions): raise AcceptanceError("Celery-redelivery Campaign audit evidence is incomplete") return { **prepared.public_evidence(), "queue": queue_evidence, "interrupted_durable_state": interrupted_state, "recovered_durable_state": recovered_state, "protocol": protocol, "report": report, "audit_actions": audit_actions, "broker": { "transport": "redis", "same_task_identity_redelivered": first_task_id == redelivered_task_id, "first_worker_received_count": len(first_received), "replacement_worker_received_count": len(replacement_received), **broker_after.as_dict(), }, "supervision": { "first_worker_killed_after_complete_data": True, "first_worker_forced_exit": first_exit_code != 0, "replacement_worker_started": True, "replacement_worker_completed_redelivery": True, }, } finally: endpoint.release_held_connection() _stop_worker(first_worker, timeout_seconds=5) if replacement_worker is not None: _stop_worker(replacement_worker, timeout_seconds=5) def _runtime_module_versions(registry: Any) -> dict[str, str]: versions = {"core": _core_package_version()} versions.update( { manifest.id: manifest.version for manifest in registry.manifests() if manifest.id in {"access", "audit", "campaigns", "mail"} } ) return versions def _create_runtime_root() -> Path: """Create the isolated runtime under the platform-selected temp root.""" return Path(tempfile.mkdtemp(prefix="govoplan-campaign-celery-redelivery-")) def _bootstrap_and_run( *, settings: TestbedSettings, fixture_path: Path, redis_url: str, visibility_timeout_seconds: int, ) -> dict[str, Any]: runtime_root = _create_runtime_root() database = None try: os.environ.update( { "APP_ENV": "test", "DATABASE_URL": f"sqlite:///{runtime_root / 'acceptance.db'}", "FILE_STORAGE_BACKEND": "local", "FILE_STORAGE_LOCAL_ROOT": str(runtime_root / "files"), "MOCK_MAILBOX_DIR": str(runtime_root / "mock-mailbox"), "DEV_BOOTSTRAP_ENABLED": "false", "CELERY_ENABLED": "true", "REDIS_URL": redis_url, "GOVOPLAN_CAMPAIGN_TEST_REDIS_VISIBILITY_TIMEOUT_SECONDS": str( visibility_timeout_seconds ), "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true", } ) from fastapi.testclient import TestClient from govoplan_core.db.base import Base from govoplan_core.db.bootstrap import bootstrap_dev_data from govoplan_core.db.session import configure_database, set_database from govoplan_core.settings import Settings, settings as core_settings from govoplan_core.tenancy.scope import create_scope_tables isolated_settings = Settings() for field_name in Settings.model_fields: setattr(core_settings, field_name, getattr(isolated_settings, field_name)) database = configure_database(os.environ["DATABASE_URL"]) set_database(database) from govoplan_core.server.app import app create_scope_tables(database.engine) Base.metadata.create_all(bind=database.engine) with database.SessionLocal() as session: bootstrap_dev_data( session, api_key_secret="celery-redelivery-unused-api-key", user_password="celery-redelivery-admin", ) def snapshot_probe( version_id: str, ) -> tuple[Mapping[str, Any], Mapping[str, Any]]: from govoplan_campaign.backend.db.models import CampaignVersion with database.SessionLocal() as session: version = session.get(CampaignVersion, version_id) if version is None: raise AcceptanceError("Campaign execution snapshot is unavailable") raw = version.raw_json if isinstance(version.raw_json, dict) else {} snapshot = ( version.execution_snapshot if isinstance(version.execution_snapshot, dict) else {} ) return raw, snapshot def audit_probe(campaign_id: str, version_id: str) -> Mapping[str, int]: from govoplan_audit.backend.db.models import AuditLog with database.SessionLocal() as session: actions = [ row[0] for row in session.query(AuditLog.action) .filter(AuditLog.object_id.in_([campaign_id, version_id])) .all() if row[0] in EXPECTED_AUDIT_ACTIONS ] return dict(Counter(actions)) def delivery_probe(campaign_id: str, version_id: str) -> Mapping[str, Any]: from govoplan_campaign.backend.db.models import CampaignJob, SendAttempt with database.SessionLocal() as session: jobs = ( session.query(CampaignJob) .filter( CampaignJob.campaign_id == campaign_id, CampaignJob.campaign_version_id == version_id, ) .all() ) job_ids = [job.id for job in jobs] attempts = ( session.query(SendAttempt) .filter(SendAttempt.job_id.in_(job_ids)) .all() if job_ids else [] ) return { "job_count": len(jobs), "send_status_counts": dict( Counter(job.send_status for job in jobs) ), "attempt_status_counts": dict( Counter(attempt.status for attempt in attempts) ), "unfinished_attempt_count": sum( 1 for attempt in attempts if attempt.finished_at is None ), } with TestClient(app) as client: login = _expect( client.post( "/api/v1/auth/login", json={ "email": "admin@example.local", "password": "celery-redelivery-admin", }, ), 200, "Acceptance login", ) access_token = str(login.get("access_token") or "") if not access_token: raise AcceptanceError("Acceptance login returned no access token") from govoplan_core.core.runtime import get_registry registry = get_registry() if registry is None: raise AcceptanceError("The GovOPlaN module registry is unavailable") composition_versions = required_composition_versions( fixture_path, _runtime_module_versions(registry), ) with smtp_fault_endpoint("post_data_hold") as endpoint: scenario = execute_redelivery_scenario( client, {"Authorization": f"Bearer {access_token}"}, fixture_path=fixture_path, settings=settings, endpoint=endpoint, redis_url=redis_url, runtime_root=runtime_root, snapshot_probe=snapshot_probe, audit_probe=audit_probe, delivery_probe=delivery_probe, ) evidence = { "schema_version": EVIDENCE_SCHEMA, "generated_at": datetime.now(timezone.utc).isoformat(), "fixture_sha256": hashlib.sha256(fixture_path.read_bytes()).hexdigest(), "declared_module_versions": composition_versions, "target": { "kind": "local_redis_celery_controlled_smtp", "isolated_temporary_database": True, "redis_started_by_runner": True, "worker_pool": "solo", "worker_prefetch_multiplier": 1, "task_acks_late": True, "task_reject_on_worker_lost": True, "visibility_timeout_seconds": visibility_timeout_seconds, }, "scenario": scenario, "coverage": { "redis_broker_delivery": True, "celery_worker_processes": True, "forced_worker_loss_after_complete_data": True, "same_task_broker_redelivery": True, "durable_outcome_unknown_recovery": True, "duplicate_smtp_transaction_prevented": True, "production_daemon_supervisor": False, "target_provider": False, "source_artifact_provenance": False, }, } _assert_evidence_safe(evidence, settings=settings) rendered = json.dumps( evidence, ensure_ascii=False, indent=2, sort_keys=True, ).encode("utf-8") + b"\n" if len(rendered) > MAX_EVIDENCE_BYTES: raise AcceptanceError("Celery-redelivery evidence exceeds its size limit") return evidence finally: if database is not None: database.engine.dispose() shutil.rmtree(runtime_root, ignore_errors=True) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE) parser.add_argument("--compose-file", type=Path, default=DEFAULT_COMPOSE_FILE) parser.add_argument("--redis-port", type=int) parser.add_argument( "--visibility-timeout-seconds", type=lambda value: _positive_int(value, label="visibility timeout"), default=os.environ.get( "GOVOPLAN_CAMPAIGN_TEST_REDIS_VISIBILITY_TIMEOUT_SECONDS", "3", ), ) parser.add_argument( "--timeout-seconds", type=lambda value: _positive_int(value, label="timeout"), default=60, ) parser.add_argument("--evidence", type=Path) args = parser.parse_args(argv) try: settings = TestbedSettings.from_environment() settings.assert_local_testbed() settings = replace( settings, provider_timeout_seconds=args.timeout_seconds, ) with isolated_redis_broker( compose_file=args.compose_file.resolve(), timeout_seconds=args.timeout_seconds, requested_port=args.redis_port, ) as redis_url: evidence = _bootstrap_and_run( settings=settings, fixture_path=args.fixture.resolve(), redis_url=redis_url, visibility_timeout_seconds=args.visibility_timeout_seconds, ) rendered = json.dumps( evidence, ensure_ascii=False, indent=2, sort_keys=True, ) + "\n" if args.evidence: args.evidence.parent.mkdir(parents=True, exist_ok=True) args.evidence.write_text(rendered, encoding="utf-8") else: sys.stdout.write(rendered) return 0 except AcceptanceError as exc: print(f"Campaign Celery-redelivery acceptance failed: {exc}", file=sys.stderr) return 1 except Exception as exc: print( "Campaign Celery-redelivery acceptance failed unexpectedly " f"({type(exc).__name__}); inspect local service logs.", file=sys.stderr, ) return 1 if __name__ == "__main__": raise SystemExit(main())