From 9da03090a7487dbec27aa2e820b9d7a3e6e68b03 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Mon, 3 Aug 2026 03:48:03 +0200 Subject: [PATCH] Fence Campaign delivery effects in recovery ledger --- docs/CAMPAIGN_DELIVERY_RUNBOOK.md | 11 + src/govoplan_campaign/backend/manifest.py | 2 +- src/govoplan_campaign/backend/sending/jobs.py | 681 +++++++++++++++++- tests/conftest.py | 28 + tests/test_campaign_delivery_recovery.py | 119 +++ tests/test_imap_append_idempotency.py | 5 + tests/test_single_message_actions.py | 14 + 7 files changed, 821 insertions(+), 39 deletions(-) create mode 100644 tests/conftest.py create mode 100644 tests/test_campaign_delivery_recovery.py diff --git a/docs/CAMPAIGN_DELIVERY_RUNBOOK.md b/docs/CAMPAIGN_DELIVERY_RUNBOOK.md index c2c3bd5..b69ee80 100644 --- a/docs/CAMPAIGN_DELIVERY_RUNBOOK.md +++ b/docs/CAMPAIGN_DELIVERY_RUNBOOK.md @@ -65,6 +65,17 @@ Before the first live send for a sender domain or mail-server profile: ## Outcome Handling +Each real Campaign job delivery is represented by a Core recovery-ledger +operation before the worker claims the job or invokes Mail, Postbox, or print. +Synchronous batches use the same boundary after their batch-wide preflight. +Explicit test/resend actions and post-acceptance IMAP appends use separate +action-fenced operations. Operations store only opaque IDs, digests, channel +policy, and bounded status evidence. A verified acceptance becomes +`succeeded`, a definitive pre-effect or provider rejection becomes `rejected`, +and uncertain or stranded effects remain `outcome_unknown` or +`recovery_required` in Ops. Campaign jobs, message actions, and channel-attempt +records remain the business source of truth. + - `smtp_accepted`: Do not retry. If IMAP append is enabled and pending, run or enqueue the append action. - `failed_temporary`: Retry explicitly after checking the error and retry count. diff --git a/src/govoplan_campaign/backend/manifest.py b/src/govoplan_campaign/backend/manifest.py index 202a58d..3fc0279 100644 --- a/src/govoplan_campaign/backend/manifest.py +++ b/src/govoplan_campaign/backend/manifest.py @@ -1052,7 +1052,7 @@ manifest = ModuleManifest( id="campaigns.reference.shared-build-artifacts", title="Operate Campaign build artifacts across workers", summary="Generated messages use shared object storage and are verified before delivery.", - body="Campaign stores generated EML under opaque shared object keys and records expected size, SHA-256 digest, and Message-ID in each job. A worker may run on another node and verifies that evidence before delivery. Before object or Files-owned output effects, a fenced Core recovery operation records the canonical build request, validated-version evidence, and reserved object prefix. Object-only failures prove compensation; managed Files output and uncertain cleanup remain explicit forward-recovery or recovery-required work in Ops. Retention retains metadata when deletion fails. Never copy or edit runtime object keys as business data.", + body="Campaign stores generated EML under opaque shared object keys and records expected size, SHA-256 digest, and Message-ID in each job. A worker may run on another node and verifies that evidence before delivery. Before object or Files-owned output effects, a fenced Core recovery operation records the canonical build request, validated-version evidence, and reserved object prefix. Before a real Mail, Postbox, or print effect, a separate job-fenced operation records immutable message and recipient digests and later verifies the authoritative Campaign/channel attempt state. Definitive rejection is distinct from accepted, outcome-unknown, and recovery-required work in Ops. Object-only failures prove compensation; managed Files output and uncertain cleanup remain explicit forward-recovery work. Retention retains metadata when deletion fails. Never copy or edit runtime object keys as business data.", layer="evidence", documentation_types=("admin",), audience=("campaign_operator", "platform_operator", "release_reviewer"), diff --git a/src/govoplan_campaign/backend/sending/jobs.py b/src/govoplan_campaign/backend/sending/jobs.py index f9ca370..ca1bbca 100644 --- a/src/govoplan_campaign/backend/sending/jobs.py +++ b/src/govoplan_campaign/backend/sending/jobs.py @@ -15,6 +15,13 @@ from sqlalchemy import func from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session +from govoplan_core.core.recovery import RecoveryMode, RecoveryPlan, RecoveryStatus +from govoplan_core.core.recovery_runtime import ( + DurableRecoveryOperation, + RecoveryOperationBusy, + RecoveryOperationStateConflict, + begin_durable_recovery_operation, +) from govoplan_core.core.notifications import ( NotificationDispatchRequest, notification_dispatch_provider, @@ -23,7 +30,9 @@ from govoplan_core.core.object_storage import ( StorageBackendError, configured_storage_backend, ) +from govoplan_core.core.runtime_coordination import process_runtime_identity from govoplan_core.audit.logging import audit_event +from govoplan_core.db.session import get_database from govoplan_core.security.redaction import redact_secret_values from govoplan_core.settings import settings as core_settings from govoplan_campaign.backend.db.models import ( @@ -1074,19 +1083,13 @@ def send_campaign_now( skipped_after_queue = 0 for job in jobs: try: - claimed = _claimed_campaign_job_for_delivery(session, job) - if isinstance(claimed, SendJobResult): - result = claimed - else: - claimed_job, claim_token = claimed - result = _send_claimed_campaign_job( - session, - job=claimed_job, - claim_token=claim_token, - context=delivery_contexts[job.id], - use_rate_limit=use_rate_limit, - enqueue_imap_task=enqueue_imap_task, - ) + result = _deliver_job_with_recovery( + session, + job=job, + context=delivery_contexts[job.id], + use_rate_limit=use_rate_limit, + enqueue_imap_task=enqueue_imap_task, + ) result_dict = result.as_dict() results.append(result_dict) if result.status in DELIVERY_ACCEPTED_STATUSES | {"already_accepted"}: @@ -2096,6 +2099,134 @@ def _single_message_action_response( } +def _begin_single_action_delivery_recovery( + *, + action: CampaignMessageAction, + job: CampaignJob, + delivery_context: _SendJobDeliveryContext, +): + return begin_durable_recovery_operation( + get_database().SessionLocal, + identity=process_runtime_identity(), + module_id="campaigns", + operation_type="single-message-external-delivery", + idempotency_key=f"campaign-message-action:{action.id}", + request={ + "tenant_id": action.tenant_id, + "campaign_id": action.campaign_id, + "version_id": action.campaign_version_id, + "job_id": action.job_id, + "action_id": action.id, + "action_kind": action.kind, + "message_sha256": action.message_sha256, + "recipient_manifest_sha256": action.recipient_manifest_sha256, + "smtp_transport_revision_sha256": hashlib.sha256( + str(delivery_context.snapshot.smtp_transport_revision).encode("utf-8") + ).hexdigest(), + }, + recovery_plan=RecoveryPlan( + mode=RecoveryMode.FORWARD_RECOVERY, + preconditions=( + "the immutable single-message action is durably recorded", + "the message and SMTP transport revision passed preflight", + ), + forward_recovery_steps=( + "inspect the Campaign message action and SMTP provider evidence", + "reconcile acceptance, rejection, or an unknown outcome", + ), + verification_steps=( + "reload the message action through an independent session", + "verify its terminal outcome and attempt evidence", + ), + ), + precondition_evidence={ + "action_id": action.id, + "action_status": action.status, + "action_kind": action.kind, + "job_id": job.id, + "message_sha256": action.message_sha256, + "recipient_manifest_sha256": action.recipient_manifest_sha256, + }, + lease_resource_key=f"campaign:message-action:{action.tenant_id}:{action.id}", + lease_ttl_seconds=15 * 60, + resource_type="campaign_message_action", + resource_id=action.id, + metadata={"resources": ["postgresql", "smtp"]}, + ) + + +def _single_action_recovery_evidence( + action_id: str, +) -> tuple[str, dict[str, Any]]: + with get_database().SessionLocal() as evidence_session: + action = evidence_session.get(CampaignMessageAction, action_id) + if action is None: + return "recovery_required", { + "verified": False, + "action_present": False, + } + latest = ( + evidence_session.query(CampaignMessageActionAttempt) + .filter(CampaignMessageActionAttempt.action_id == action.id) + .order_by(CampaignMessageActionAttempt.attempt_number.desc()) + .first() + ) + evidence = { + "verified": True, + "checks": { + "action_state_reloaded": True, + "action_attempt_compared": True, + }, + "action_present": True, + "action_id": action.id, + "action_kind": action.kind, + "action_status": action.status, + "attempt_status": latest.status if latest else None, + "accepted_count": action.accepted_count, + "refused_count": action.refused_count, + } + if action.status in {"accepted", "accepted_with_refusals"}: + return "succeeded", evidence + if action.status == "outcome_unknown": + return "outcome_unknown", evidence + if action.status in { + "failed_temporary", + "failed_permanent", + "initiation_failed", + }: + return "failed", evidence + return "recovery_required", evidence + + +def _finish_single_action_delivery_recovery( + operation: DurableRecoveryOperation, + *, + action_id: str, +) -> None: + outcome, evidence = _single_action_recovery_evidence(action_id) + if outcome == "succeeded": + operation.succeed(evidence=evidence) + elif outcome == "failed": + operation.reject( + summary="The single-message SMTP effect was definitively rejected", + evidence=evidence, + ) + elif outcome == "outcome_unknown": + operation.unresolved( + status=RecoveryStatus.OUTCOME_UNKNOWN, + summary="The single-message SMTP outcome requires reconciliation", + evidence=evidence, + failure_summary="The message may have been accepted by SMTP", + ) + else: + operation.unresolved( + status=RecoveryStatus.RECOVERY_REQUIRED, + summary="The single-message action did not reach a verified terminal state", + evidence=evidence, + failure_summary="The message action requires forward recovery", + ) + + def _send_single_message_direct( session: Session, *, @@ -2127,6 +2258,53 @@ def _send_single_message_direct( messages_per_minute=delivery_context.snapshot.delivery.rate_limit.messages_per_minute, enabled=use_rate_limit, ) + try: + recovery_start = _begin_single_action_delivery_recovery( + action=action, + job=job, + delivery_context=delivery_context, + ) + except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc: + _finish_single_message_action( + session, + action=action, + status="initiation_failed", + error_type=exc.__class__.__name__, + error_message=str(exc), + final_send_status=job.send_status, + ) + raise QueueingError(str(exc)) from exc + if recovery_start.replayed or recovery_start.operation is None: + session.refresh(action) + return _single_message_action_response(action, duplicate=True) + operation = recovery_start.operation + try: + result = _perform_single_message_direct_effect( + session, + campaign=campaign, + version=version, + job=job, + action=action, + delivery_context=delivery_context, + enqueue_imap_task=enqueue_imap_task, + ) + except Exception: + _finish_single_action_delivery_recovery(operation, action_id=action.id) + raise + _finish_single_action_delivery_recovery(operation, action_id=action.id) + return result + + +def _perform_single_message_direct_effect( + session: Session, + *, + campaign: Campaign, + version: CampaignVersion, + job: CampaignJob, + action: CampaignMessageAction, + delivery_context: _SendJobDeliveryContext, + enqueue_imap_task: bool, +) -> dict[str, Any]: attempt = _start_single_message_action_attempt(session, action) try: result = mail_integration().send_campaign_email_bytes( @@ -2726,7 +2904,12 @@ def _recipients_from_job(job: CampaignJob) -> list[str]: return list(dict.fromkeys(recipients)) -def _claim_job_for_sending(session: Session, job: CampaignJob) -> str | None: +def _claim_job_for_sending( + session: Session, + job: CampaignJob, + *, + claim_token: str | None = None, +) -> str | None: """Atomically claim a queued job and return the claim token. A duplicate task can observe CLAIMED/SENDING but cannot acquire a second @@ -2735,7 +2918,7 @@ def _claim_job_for_sending(session: Session, job: CampaignJob) -> str | None: could race a slow worker. """ - claim_token = str(uuid4()) + effective_claim_token = claim_token or str(uuid4()) changed = ( session.query(CampaignJob) .filter( @@ -2748,7 +2931,7 @@ def _claim_job_for_sending(session: Session, job: CampaignJob) -> str | None: CampaignJob.queue_status: JobQueueStatus.SENDING.value, CampaignJob.send_status: JobSendStatus.CLAIMED.value, CampaignJob.claimed_at: _utcnow(), - CampaignJob.claim_token: claim_token, + CampaignJob.claim_token: effective_claim_token, CampaignJob.last_error: None, }, synchronize_session=False, @@ -2756,7 +2939,7 @@ def _claim_job_for_sending(session: Session, job: CampaignJob) -> str | None: ) session.commit() session.expire_all() - return claim_token if changed == 1 else None + return effective_claim_token if changed == 1 else None def _record_attempt_start( @@ -2946,6 +3129,184 @@ def _campaign_delivery_status_pair( return None +def _canonical_delivery_sha256(value: object) -> str: + return hashlib.sha256( + json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + default=str, + ).encode("utf-8") + ).hexdigest() + + +def _begin_job_delivery_recovery( + *, + job: CampaignJob, + context: _SendJobDeliveryContext, + claim_token: str, +): + channel_policy = DeliveryChannelPolicy(job.delivery_channel_policy) + resources = ["postgresql"] + if channel_policy.uses_mail: + resources.append("smtp") + if channel_policy.uses_postbox: + resources.append("postbox") + if channel_policy.uses_print: + resources.append("print-provider") + claim_sha256 = hashlib.sha256(claim_token.encode("utf-8")).hexdigest() + recipient_manifest_sha256 = _canonical_delivery_sha256( + _recipients_from_job(job) + ) + return begin_durable_recovery_operation( + get_database().SessionLocal, + identity=process_runtime_identity(), + module_id="campaigns", + operation_type="external-channel-delivery", + idempotency_key=f"campaign-delivery:{job.id}:{claim_sha256[:32]}", + request={ + "tenant_id": job.tenant_id, + "campaign_id": job.campaign_id, + "version_id": job.campaign_version_id, + "job_id": job.id, + "channel_policy": channel_policy.value, + "message_sha256": job.eml_sha256, + "recipient_manifest_sha256": recipient_manifest_sha256, + "execution_snapshot_sha256": context.version.execution_snapshot_hash, + }, + recovery_plan=RecoveryPlan( + mode=RecoveryMode.FORWARD_RECOVERY, + preconditions=( + "the immutable built job is explicitly queued", + "message and transport revisions passed preflight", + ), + forward_recovery_steps=( + "inspect provider and module attempt evidence", + "reconcile accepted, rejected, or outcome-unknown channel state", + ), + verification_steps=( + "reload the Campaign job through an independent session", + "compare final channel and attempt states", + ), + ), + precondition_evidence={ + "job_id": job.id, + "queue_status": job.queue_status, + "send_status": job.send_status, + "build_status": job.build_status, + "validation_status": job.validation_status, + "message_sha256": job.eml_sha256, + "recipient_manifest_sha256": recipient_manifest_sha256, + "claim_sha256": claim_sha256, + }, + lease_resource_key=f"campaign:delivery:{job.tenant_id}:{job.id}", + lease_ttl_seconds=30 * 60, + resource_type="campaign_job", + resource_id=job.id, + metadata={ + "resources": resources, + "channel_policy": channel_policy.value, + }, + ) + + +def _job_delivery_recovery_evidence(job_id: str) -> tuple[str, dict[str, Any]]: + with get_database().SessionLocal() as evidence_session: + job = evidence_session.get(CampaignJob, job_id) + if job is None: + return "recovery_required", { + "verified": False, + "job_present": False, + } + latest_smtp = ( + evidence_session.query(SendAttempt) + .filter(SendAttempt.job_id == job.id) + .order_by(SendAttempt.attempt_number.desc()) + .first() + ) + latest_postbox = ( + evidence_session.query(PostboxDeliveryAttempt) + .filter(PostboxDeliveryAttempt.job_id == job.id) + .order_by(PostboxDeliveryAttempt.attempt_number.desc()) + .first() + ) + latest_print = ( + evidence_session.query(PrintOutputAttempt) + .filter(PrintOutputAttempt.job_id == job.id) + .order_by(PrintOutputAttempt.attempt_number.desc()) + .first() + ) + evidence = { + "verified": True, + "checks": { + "job_state_reloaded": True, + "channel_attempts_compared": True, + }, + "job_present": True, + "job_id": job.id, + "queue_status": job.queue_status, + "send_status": job.send_status, + "postbox_status": job.postbox_status, + "print_status": job.print_status, + "attempt_count": job.attempt_count, + "postbox_attempt_count": job.postbox_attempt_count, + "print_attempt_count": job.print_attempt_count, + "latest_attempts": { + "smtp": latest_smtp.status if latest_smtp else None, + "postbox": latest_postbox.status if latest_postbox else None, + "print": latest_print.status if latest_print else None, + }, + } + if ( + job.send_status == JobSendStatus.OUTCOME_UNKNOWN.value + or job.postbox_status == JobPostboxStatus.OUTCOME_UNKNOWN.value + ): + return "outcome_unknown", evidence + if ( + job.send_status + in {JobSendStatus.CLAIMED.value, JobSendStatus.SENDING.value} + or job.postbox_status == JobPostboxStatus.DELIVERING.value + or job.print_status == JobPrintStatus.ACCEPTING.value + ): + return "recovery_required", evidence + if job.send_status in { + *FULLY_ACCEPTED_STATUSES, + JobSendStatus.PARTIALLY_ACCEPTED.value, + }: + return "succeeded", evidence + return "failed", evidence + + +def _finish_job_delivery_recovery( + operation: DurableRecoveryOperation, + *, + job_id: str, +) -> None: + outcome, evidence = _job_delivery_recovery_evidence(job_id) + if outcome == "succeeded": + operation.succeed(evidence=evidence) + elif outcome == "failed": + operation.reject( + summary="Campaign delivery completed with a verified provider rejection", + evidence=evidence, + ) + elif outcome == "outcome_unknown": + operation.unresolved( + status=RecoveryStatus.OUTCOME_UNKNOWN, + summary="Campaign delivery provider outcome requires reconciliation", + evidence=evidence, + failure_summary="A Campaign delivery effect may have been accepted", + ) + else: + operation.unresolved( + status=RecoveryStatus.RECOVERY_REQUIRED, + summary="Campaign delivery did not reach a verified terminal state", + evidence=evidence, + failure_summary="Campaign delivery state requires forward recovery", + ) + + def send_campaign_job( session: Session, *, @@ -2986,20 +3347,82 @@ def send_campaign_job( message=f"Would deliver via {'; '.join(descriptions)}", ) - claimed = _claimed_campaign_job_for_delivery(session, job) - if isinstance(claimed, SendJobResult): - return claimed - claimed_job, claim_token = claimed - return _send_claimed_campaign_job( + return _deliver_job_with_recovery( session, - job=claimed_job, - claim_token=claim_token, + job=job, context=context, use_rate_limit=use_rate_limit, enqueue_imap_task=enqueue_imap_task, ) +def _deliver_job_with_recovery( + session: Session, + *, + job: CampaignJob, + context: _SendJobDeliveryContext, + use_rate_limit: bool, + enqueue_imap_task: bool, +) -> SendJobResult: + claim_token = str(uuid4()) + try: + recovery_start = _begin_job_delivery_recovery( + job=job, + context=context, + claim_token=claim_token, + ) + except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc: + return SendJobResult( + job_id=job.id, + status="recovery_blocked", + attempt_number=job.attempt_count, + message=str(exc), + ) + if recovery_start.replayed or recovery_start.operation is None: + return SendJobResult( + job_id=job.id, + status="already_completed", + attempt_number=job.attempt_count, + message="The durable delivery operation already completed.", + ) + recovery_operation = recovery_start.operation + claimed = _claimed_campaign_job_for_delivery( + session, + job, + claim_token=claim_token, + ) + if isinstance(claimed, SendJobResult): + recovery_operation.reject( + summary="Campaign job could not be claimed before any provider effect", + evidence={ + "verified": True, + "checks": { + "provider_effect_started": False, + "claim_rejected": True, + }, + "provider_effect_started": False, + "job_id": job.id, + "claim_result": claimed.status, + }, + ) + return claimed + claimed_job, claim_token = claimed + try: + result = _send_claimed_campaign_job( + session, + job=claimed_job, + claim_token=claim_token, + context=context, + use_rate_limit=use_rate_limit, + enqueue_imap_task=enqueue_imap_task, + ) + except Exception: + _finish_job_delivery_recovery(recovery_operation, job_id=job.id) + raise + _finish_job_delivery_recovery(recovery_operation, job_id=job.id) + return result + + def _preflight_send_campaign_job( session: Session, job: CampaignJob, @@ -3100,15 +3523,21 @@ def _send_job_delivery_context( def _claimed_campaign_job_for_delivery( session: Session, job: CampaignJob, + *, + claim_token: str | None = None, ) -> tuple[CampaignJob, str] | SendJobResult: - claim_token = _claim_job_for_sending(session, job) - if claim_token is None: + effective_claim = _claim_job_for_sending( + session, + job, + claim_token=claim_token, + ) + if effective_claim is None: return _not_claimed_send_job_result(session, job) job = session.get(CampaignJob, job.id) if job is None: raise SendJobError("Claimed campaign job disappeared before send.") - return job, claim_token + return job, effective_claim def _not_claimed_send_job_result(session: Session, job: CampaignJob) -> SendJobResult: @@ -3878,7 +4307,12 @@ def _imap_attempt_count(session: Session, job_id: str) -> int: ) -def _claim_job_for_imap_append(session: Session, job: CampaignJob) -> str | None: +def _claim_job_for_imap_append( + session: Session, + job: CampaignJob, + *, + claim_token: str | None = None, +) -> str | None: """Atomically grant one worker permission to invoke the IMAP provider.""" if not _mail_was_accepted( @@ -3891,7 +4325,7 @@ def _claim_job_for_imap_append(session: Session, job: CampaignJob) -> str | None ), ): return None - claim_token = str(uuid4()) + effective_claim_token = claim_token or str(uuid4()) changed = ( session.query(CampaignJob) .filter( @@ -3905,7 +4339,7 @@ def _claim_job_for_imap_append(session: Session, job: CampaignJob) -> str | None { CampaignJob.imap_status: JobImapStatus.APPENDING.value, CampaignJob.imap_claimed_at: _utcnow(), - CampaignJob.imap_claim_token: claim_token, + CampaignJob.imap_claim_token: effective_claim_token, CampaignJob.last_error: None, }, synchronize_session=False, @@ -3913,7 +4347,7 @@ def _claim_job_for_imap_append(session: Session, job: CampaignJob) -> str | None ) session.commit() session.expire_all() - return claim_token if changed == 1 else None + return effective_claim_token if changed == 1 else None def _record_imap_attempt_start( @@ -4269,10 +4703,17 @@ def _imap_append_dry_run( def _claim_imap_append( - session: Session, job: CampaignJob + session: Session, + job: CampaignJob, + *, + claim_token: str | None = None, ) -> _ClaimedImapAppend | AppendSentResult: - claim_token = _claim_job_for_imap_append(session, job) - if claim_token is None: + effective_claim = _claim_job_for_imap_append( + session, + job, + claim_token=claim_token, + ) + if effective_claim is None: current = session.get(CampaignJob, job.id) if current is None: raise SendJobError(f"Job disappeared while claiming IMAP append: {job.id}") @@ -4290,8 +4731,8 @@ def _claim_imap_append( raise SendJobError("Claimed campaign job disappeared before IMAP append") return _ClaimedImapAppend( job=claimed_job, - attempt=_record_imap_attempt_start(session, claimed_job, claim_token), - claim_token=claim_token, + attempt=_record_imap_attempt_start(session, claimed_job, effective_claim), + claim_token=effective_claim, ) @@ -4392,6 +4833,127 @@ def _perform_imap_append( ) +def _begin_imap_append_recovery( + *, + job: CampaignJob, + context: _ImapAppendContext, + claim_token: str, +): + claim_sha256 = hashlib.sha256(claim_token.encode("utf-8")).hexdigest() + return begin_durable_recovery_operation( + get_database().SessionLocal, + identity=process_runtime_identity(), + module_id="campaigns", + operation_type="imap-sent-append", + idempotency_key=f"campaign-imap:{job.id}:{claim_sha256[:32]}", + request={ + "tenant_id": job.tenant_id, + "campaign_id": job.campaign_id, + "version_id": job.campaign_version_id, + "job_id": job.id, + "message_sha256": job.eml_sha256, + "folder_sha256": hashlib.sha256( + context.folder.encode("utf-8") + ).hexdigest(), + "imap_transport_revision_sha256": hashlib.sha256( + context.snapshot.imap_transport_revision.encode("utf-8") + ).hexdigest(), + }, + recovery_plan=RecoveryPlan( + mode=RecoveryMode.FORWARD_RECOVERY, + preconditions=( + "SMTP acceptance is durable", + "the exact EML and IMAP transport revision passed preflight", + ), + forward_recovery_steps=( + "inspect the target mailbox and append attempt", + "reconcile as appended or not appended before retry", + ), + verification_steps=( + "reload the Campaign IMAP state through an independent session", + "compare the latest append attempt", + ), + ), + precondition_evidence={ + "job_id": job.id, + "send_status": job.send_status, + "imap_status": job.imap_status, + "message_sha256": job.eml_sha256, + "claim_sha256": claim_sha256, + }, + lease_resource_key=f"campaign:imap:{job.tenant_id}:{job.id}", + lease_ttl_seconds=15 * 60, + resource_type="campaign_job", + resource_id=job.id, + metadata={"resources": ["postgresql", "imap"]}, + ) + + +def _imap_recovery_evidence(job_id: str) -> tuple[str, dict[str, Any]]: + with get_database().SessionLocal() as evidence_session: + job = evidence_session.get(CampaignJob, job_id) + if job is None: + return "recovery_required", { + "verified": False, + "job_present": False, + } + latest = ( + evidence_session.query(ImapAppendAttempt) + .filter(ImapAppendAttempt.job_id == job.id) + .order_by(ImapAppendAttempt.attempt_number.desc()) + .first() + ) + evidence = { + "verified": True, + "checks": { + "job_state_reloaded": True, + "append_attempt_compared": True, + }, + "job_present": True, + "job_id": job.id, + "send_status": job.send_status, + "imap_status": job.imap_status, + "attempt_count": _imap_attempt_count(evidence_session, job.id), + "latest_attempt_status": latest.status if latest else None, + } + if job.imap_status == JobImapStatus.APPENDED.value: + return "succeeded", evidence + if job.imap_status == JobImapStatus.FAILED.value: + return "failed", evidence + if job.imap_status == JobImapStatus.OUTCOME_UNKNOWN.value: + return "outcome_unknown", evidence + return "recovery_required", evidence + + +def _finish_imap_append_recovery( + operation: DurableRecoveryOperation, + *, + job_id: str, +) -> None: + outcome, evidence = _imap_recovery_evidence(job_id) + if outcome == "succeeded": + operation.succeed(evidence=evidence) + elif outcome == "failed": + operation.reject( + summary="The IMAP provider definitively rejected the append", + evidence=evidence, + ) + elif outcome == "outcome_unknown": + operation.unresolved( + status=RecoveryStatus.OUTCOME_UNKNOWN, + summary="The IMAP append outcome requires mailbox reconciliation", + evidence=evidence, + failure_summary="The exact message may already exist in the Sent folder", + ) + else: + operation.unresolved( + status=RecoveryStatus.RECOVERY_REQUIRED, + summary="The IMAP append did not reach a verified terminal state", + evidence=evidence, + failure_summary="The IMAP append requires forward recovery", + ) + + def append_sent_for_job( session: Session, *, job_id: str, dry_run: bool = False ) -> AppendSentResult: @@ -4408,10 +4970,53 @@ def append_sent_for_job( return prepared if dry_run: return _imap_append_dry_run(session, job, prepared) - claimed = _claim_imap_append(session, job) + claim_token = str(uuid4()) + try: + recovery_start = _begin_imap_append_recovery( + job=job, + context=prepared, + claim_token=claim_token, + ) + except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc: + return AppendSentResult( + job_id=job.id, + status="recovery_blocked", + attempt_number=_imap_attempt_count(session, job.id), + message=str(exc), + ) + if recovery_start.replayed or recovery_start.operation is None: + return AppendSentResult( + job_id=job.id, + status="already_completed", + attempt_number=_imap_attempt_count(session, job.id), + ) + recovery_operation = recovery_start.operation + claimed = _claim_imap_append( + session, + job, + claim_token=claim_token, + ) if isinstance(claimed, AppendSentResult): + recovery_operation.reject( + summary="The IMAP append claim was rejected before any provider effect", + evidence={ + "verified": True, + "checks": { + "provider_effect_started": False, + "claim_rejected": True, + }, + "provider_effect_started": False, + "claim_result": claimed.status, + }, + ) return claimed - return _perform_imap_append(session, claimed, prepared) + try: + result = _perform_imap_append(session, claimed, prepared) + except Exception: + _finish_imap_append_recovery(recovery_operation, job_id=job.id) + raise + _finish_imap_append_recovery(recovery_operation, job_id=job.id) + return result def enqueue_pending_imap_appends( diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..74fd646 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +import pytest + +from govoplan_core.core import runtime_coordination +from govoplan_core.core.runtime_coordination import ( + RuntimeIdentity, + bind_process_runtime_identity, +) + + +@pytest.fixture(scope="session", autouse=True) +def _bind_test_runtime_identity(): + previous = runtime_coordination._process_runtime_identity + bind_process_runtime_identity( + RuntimeIdentity( + installation_id="campaign-tests", + node_id="campaign-test-process", + incarnation="campaign-test-incarnation", + role="test", + software_version="test", + composition_hash="a" * 64, + ) + ) + try: + yield + finally: + bind_process_runtime_identity(previous) diff --git a/tests/test_campaign_delivery_recovery.py b/tests/test_campaign_delivery_recovery.py new file mode 100644 index 0000000..5aeefa7 --- /dev/null +++ b/tests/test_campaign_delivery_recovery.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +from unittest.mock import Mock + +import pytest + +from govoplan_core.core.recovery import RecoveryStatus +from govoplan_campaign.backend.sending import jobs + + +@pytest.mark.parametrize( + ("outcome", "expected_method", "expected_status"), + [ + ("succeeded", "succeed", None), + ("failed", "reject", None), + ("outcome_unknown", "unresolved", RecoveryStatus.OUTCOME_UNKNOWN), + ("recovery_required", "unresolved", RecoveryStatus.RECOVERY_REQUIRED), + ], +) +def test_delivery_recovery_maps_verified_job_state( + monkeypatch, + outcome: str, + expected_method: str, + expected_status: RecoveryStatus | None, +) -> None: + evidence = { + "verified": outcome != "recovery_required", + "checks": {"job_state_reloaded": True}, + "send_status": outcome, + } + monkeypatch.setattr( + jobs, + "_job_delivery_recovery_evidence", + lambda _job_id: (outcome, evidence), + ) + operation = Mock() + + jobs._finish_job_delivery_recovery(operation, job_id="job-1") + + method = getattr(operation, expected_method) + method.assert_called_once() + if expected_status is not None: + assert method.call_args.kwargs["status"] == expected_status + for other in {"succeed", "reject", "unresolved"} - {expected_method}: + getattr(operation, other).assert_not_called() + + +@pytest.mark.parametrize( + ("outcome", "expected_method", "expected_status"), + [ + ("succeeded", "succeed", None), + ("failed", "reject", None), + ("outcome_unknown", "unresolved", RecoveryStatus.OUTCOME_UNKNOWN), + ("recovery_required", "unresolved", RecoveryStatus.RECOVERY_REQUIRED), + ], +) +def test_imap_recovery_maps_verified_append_state( + monkeypatch, + outcome: str, + expected_method: str, + expected_status: RecoveryStatus | None, +) -> None: + evidence = { + "verified": outcome != "recovery_required", + "checks": {"job_state_reloaded": True}, + "imap_status": outcome, + } + monkeypatch.setattr( + jobs, + "_imap_recovery_evidence", + lambda _job_id: (outcome, evidence), + ) + operation = Mock() + + jobs._finish_imap_append_recovery(operation, job_id="job-1") + + method = getattr(operation, expected_method) + method.assert_called_once() + if expected_status is not None: + assert method.call_args.kwargs["status"] == expected_status + for other in {"succeed", "reject", "unresolved"} - {expected_method}: + getattr(operation, other).assert_not_called() + + +@pytest.mark.parametrize( + ("outcome", "expected_method", "expected_status"), + [ + ("succeeded", "succeed", None), + ("failed", "reject", None), + ("outcome_unknown", "unresolved", RecoveryStatus.OUTCOME_UNKNOWN), + ("recovery_required", "unresolved", RecoveryStatus.RECOVERY_REQUIRED), + ], +) +def test_single_action_recovery_maps_verified_action_state( + monkeypatch, + outcome: str, + expected_method: str, + expected_status: RecoveryStatus | None, +) -> None: + evidence = { + "verified": outcome != "recovery_required", + "checks": {"action_state_reloaded": True}, + "action_status": outcome, + } + monkeypatch.setattr( + jobs, + "_single_action_recovery_evidence", + lambda _action_id: (outcome, evidence), + ) + operation = Mock() + + jobs._finish_single_action_delivery_recovery(operation, action_id="action-1") + + method = getattr(operation, expected_method) + method.assert_called_once() + if expected_status is not None: + assert method.call_args.kwargs["status"] == expected_status + for other in {"succeed", "reject", "unresolved"} - {expected_method}: + getattr(operation, other).assert_not_called() diff --git a/tests/test_imap_append_idempotency.py b/tests/test_imap_append_idempotency.py index 45e9687..c333369 100644 --- a/tests/test_imap_append_idempotency.py +++ b/tests/test_imap_append_idempotency.py @@ -159,6 +159,11 @@ def test_post_provider_persistence_failure_freezes_imap_retry() -> None: patch("govoplan_campaign.backend.sending.jobs._load_eml_bytes_for_job", return_value=b"message"), patch("govoplan_campaign.backend.sending.jobs._claim_job_for_imap_append", return_value="claim-1"), patch("govoplan_campaign.backend.sending.jobs._record_imap_attempt_start", return_value=attempt), + patch( + "govoplan_campaign.backend.sending.jobs._begin_imap_append_recovery", + return_value=SimpleNamespace(replayed=False, operation=MagicMock()), + ), + patch("govoplan_campaign.backend.sending.jobs._finish_imap_append_recovery"), patch("govoplan_campaign.backend.sending.jobs.mail_integration", return_value=Mail()), patch( "govoplan_campaign.backend.sending.jobs._record_imap_append_success", diff --git a/tests/test_single_message_actions.py b/tests/test_single_message_actions.py index 465ed01..ae98869 100644 --- a/tests/test_single_message_actions.py +++ b/tests/test_single_message_actions.py @@ -170,6 +170,13 @@ class CampaignSingleMessageActionTests(unittest.TestCase): "govoplan_campaign.backend.sending.jobs.mail_integration", return_value=mail, ), + patch( + "govoplan_campaign.backend.sending.jobs._begin_single_action_delivery_recovery", + return_value=SimpleNamespace(replayed=False, operation=Mock()), + ), + patch( + "govoplan_campaign.backend.sending.jobs._finish_single_action_delivery_recovery" + ), patch( "govoplan_campaign.backend.sending.jobs.audit_event" ), @@ -269,6 +276,13 @@ class CampaignSingleMessageActionTests(unittest.TestCase): "govoplan_campaign.backend.sending.jobs.mail_integration", return_value=mail, ), + patch( + "govoplan_campaign.backend.sending.jobs._begin_single_action_delivery_recovery", + return_value=SimpleNamespace(replayed=False, operation=Mock()), + ), + patch( + "govoplan_campaign.backend.sending.jobs._finish_single_action_delivery_recovery" + ), patch( "govoplan_campaign.backend.sending.jobs.files_integration" ) as files,