diff --git a/README.md b/README.md index de8d255..6ed2fd5 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,11 @@ Platform RBAC and governance rules are documented in `govoplan-core/docs/`. - [Campaign handbook](docs/CAMPAIGN_HANDBOOK.md) provides the adaptive user, process, governance, technical, and operations perspectives. - [Campaign delivery runbook](docs/CAMPAIGN_DELIVERY_RUNBOOK.md) covers queueing, local vs Celery operation, retries, reconciliation, reports, and the live SMTP/IMAP test checklist. - Immediate delivery is bounded to 25 exact eligible recipient jobs by default. Deployments may set `GOVOPLAN_CAMPAIGN_SYNCHRONOUS_SEND_MAX_RECIPIENTS` (0–500), and tenants may narrow that ceiling through `campaign_delivery_policy.synchronous_send_max_recipients` in tenant settings. +- Immediate Mail delivery preflights the selected SMTP transport before the + first effect and reuses a healthy bounded connection through Mail. Review and + send reports the batch state, connection/reconnect counts, and paused count. + A systemic authentication, sender, or connectivity failure pauses remaining + jobs; correct and test the Mail profile before explicitly resuming them. - Report-email preview uses the selected version's stored v5 Mail-profile evidence. Live report email fails closed until [govoplan-mail#17](https://git.add-ideas.de/GovOPlaN/govoplan-mail/issues/17) provides a durable, idempotent Mail-owned outbox and transport-attempt ledger; per-job CSV is off by default and requires `campaigns:recipient:export` when requested. - [Campaign/Mail profile boundary](docs/MAIL_PROFILE_BOUNDARY.md) defines profile-only delivery, runtime resolution, execution evidence, and the fail-closed legacy migration path. - [Recipient import guide](docs/RECIPIENT_IMPORT_GUIDE.md) covers user/admin workflows, mapping profiles, validation, and import evidence. diff --git a/docs/CAMPAIGN_DELIVERY_RUNBOOK.md b/docs/CAMPAIGN_DELIVERY_RUNBOOK.md index dce71a0..d730f0e 100644 --- a/docs/CAMPAIGN_DELIVERY_RUNBOOK.md +++ b/docs/CAMPAIGN_DELIVERY_RUNBOOK.md @@ -62,6 +62,10 @@ Before the first live send for a sender domain or mail-server profile: 6. If a synchronous request is used, keep Review and send open: it polls the durable counters while the request runs. A rejection occurs before SMTP and directs oversized runs to workers. +7. Review the SMTP batch line. `ready` means DNS/connectivity/TLS/auth preflight + succeeded. Connection and reconnect counts explain reuse. `paused` means a + systemic transport failure stopped the remaining jobs before their SMTP + effect; test/correct the Mail profile and explicitly resume the queue. ## Outcome Handling @@ -88,6 +92,9 @@ unknown provider attempt merely to repair the other layer's state. - `failed_temporary`: Retry explicitly after checking the error and retry count. - `failed_permanent`: Retry only if the operator has corrected the root cause and intentionally includes permanent failures. +- `paused` after a systemic SMTP failure: do not resume until the shared Mail + profile passes its connection test. Authentication, sender rejection, and + unavailable connectivity affect the batch rather than one recipient. - `outcome_unknown`: Do not retry directly. Check SMTP logs, mailbox evidence, or provider control panels, then reconcile as accepted or not sent. - `claimed` or `sending` that does not progress: treat as a worker interruption. diff --git a/docs/CAMPAIGN_HANDBOOK.md b/docs/CAMPAIGN_HANDBOOK.md index 71ead8d..eefa5c9 100644 --- a/docs/CAMPAIGN_HANDBOOK.md +++ b/docs/CAMPAIGN_HANDBOOK.md @@ -611,7 +611,6 @@ The following are part of the selected reference journey but are not implied by the current baseline: - the final audited **test / single send / single resend** semantics; -- reusable SMTP batch sessions and their measured throughput benefit; - durable, idempotent Campaign report delivery through a Mail-owned outbox ([`govoplan-mail#17`](https://git.add-ideas.de/GovOPlaN/govoplan-mail/issues/17)); - a fully packaged one-command Campaign reference composition with production diff --git a/src/govoplan_campaign/backend/integrations.py b/src/govoplan_campaign/backend/integrations.py index f41e2eb..33f6742 100644 --- a/src/govoplan_campaign/backend/integrations.py +++ b/src/govoplan_campaign/backend/integrations.py @@ -74,11 +74,21 @@ class SmtpConfigurationError(RuntimeError): class SmtpSendError(RuntimeError): def __init__( - self, message: str, *, temporary: bool = False, outcome_unknown: bool = False + self, + message: str, + *, + temporary: bool = False, + outcome_unknown: bool = False, + systemic: bool = False, + reason_code: str | None = None, + phase: str = "send", ) -> None: super().__init__(message) self.temporary = temporary self.outcome_unknown = outcome_unknown + self.systemic = systemic + self.reason_code = reason_code + self.phase = phase class ImapConfigurationError(RuntimeError): @@ -298,6 +308,9 @@ class MailCampaignIntegration: str(exc), temporary=bool(getattr(exc, "temporary", False)), outcome_unknown=bool(getattr(exc, "outcome_unknown", False)), + systemic=bool(getattr(exc, "systemic", False)), + reason_code=str(getattr(exc, "reason_code", "") or "") or None, + phase=str(getattr(exc, "phase", "send") or "send"), ) from exc except getattr( delegate, "SmtpConfigurationError", SmtpConfigurationError @@ -306,6 +319,30 @@ class MailCampaignIntegration: except getattr(delegate, "MailProfileError", MailProfileError) as exc: raise MailProfileError(str(exc)) from exc + @contextmanager + def campaign_smtp_batch(self, *args: Any, **kwargs: Any) -> Iterator[Any]: + delegate = self._require() + method = getattr(delegate, "campaign_smtp_batch", None) + if not callable(method): + yield None + return + try: + with method(*args, **kwargs) as state: + yield state + except getattr(delegate, "SmtpSendError", SmtpSendError) as exc: + raise SmtpSendError( + str(exc), + temporary=bool(getattr(exc, "temporary", False)), + outcome_unknown=bool(getattr(exc, "outcome_unknown", False)), + systemic=bool(getattr(exc, "systemic", False)), + reason_code=str(getattr(exc, "reason_code", "") or "") or None, + phase=str(getattr(exc, "phase", "preflight") or "preflight"), + ) from exc + except getattr(delegate, "SmtpConfigurationError", SmtpConfigurationError) as exc: + raise SmtpConfigurationError(str(exc)) from exc + except getattr(delegate, "MailProfileError", MailProfileError) as exc: + raise MailProfileError(str(exc)) from exc + def append_campaign_message_to_sent(self, *args: Any, **kwargs: Any) -> Any: delegate = self._require() try: diff --git a/src/govoplan_campaign/backend/manifest.py b/src/govoplan_campaign/backend/manifest.py index d7c6000..feaf07e 100644 --- a/src/govoplan_campaign/backend/manifest.py +++ b/src/govoplan_campaign/backend/manifest.py @@ -913,7 +913,7 @@ manifest = ModuleManifest( id="campaigns.mail-profile-operations", title="Operate profile-backed campaign delivery", summary="Workers re-authorize and resolve Mail profiles at execution time while Campaign retains only opaque Mail-owned revisions and outcomes.", - body="A legacy snapshot, unauthorized or inactive profile, profile-reference mismatch, or changed SMTP/IMAP transport revision stops delivery. Preserve the record, migrate or correct the profile selection, revalidate, rebuild, and only then queue again. Password-only rotation remains possible without copying secrets into Campaign. Uncertain SMTP and IMAP effects remain blocked until an evidence-backed operator reconciliation. If Campaign becomes unavailable to the tenant after a job was accepted, the worker leaves the job untouched and reports an operator action instead of sending or dropping it.", + body="A legacy snapshot, unauthorized or inactive profile, profile-reference mismatch, or changed SMTP/IMAP transport revision stops delivery. Synchronous Mail batches preflight DNS, connectivity, TLS, and authentication before their first effect, reuse a bounded healthy SMTP connection, and reconnect before a later message when the old connection is stale. Review and send shows batch, connection, reconnect, failure, and pause counts. A systemic authentication, sender, or connectivity failure pauses remaining queued jobs with a stable reason code; correct and test the Mail profile before explicitly resuming. A connection loss after DATA begins stays outcome-unknown and is never replayed automatically. Preserve a stopped record, migrate or correct the profile selection, revalidate, rebuild, and only then queue again. Password-only rotation remains possible without copying secrets into Campaign. Uncertain SMTP and IMAP effects remain blocked until an evidence-backed operator reconciliation. If Campaign becomes unavailable to the tenant after a job was accepted, the worker leaves the job untouched and reports an operator action instead of sending or dropping it.", layer="configured", documentation_types=("admin",), audience=("campaign_sender", "campaign_operator", "mail_admin"), diff --git a/src/govoplan_campaign/backend/response_security.py b/src/govoplan_campaign/backend/response_security.py index a4f2268..44fb0cf 100644 --- a/src/govoplan_campaign/backend/response_security.py +++ b/src/govoplan_campaign/backend/response_security.py @@ -48,7 +48,12 @@ _SEND_NOW_RESULT_KEYS = ( "failed_count", "outcome_unknown_count", "skipped_count", + "paused_count", "preflight_count", + "batch_state", + "batch_pause_reason_code", + "smtp_connection_count", + "smtp_reconnect_count", "delivery_mode", "dry_run", ) diff --git a/src/govoplan_campaign/backend/sending/jobs.py b/src/govoplan_campaign/backend/sending/jobs.py index f45ad03..d4746f6 100644 --- a/src/govoplan_campaign/backend/sending/jobs.py +++ b/src/govoplan_campaign/backend/sending/jobs.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib import json from collections import Counter +from contextlib import nullcontext from dataclasses import asdict, dataclass from datetime import datetime, timezone from email import policy @@ -225,6 +226,11 @@ class SendCampaignNowResult: failed_count: int outcome_unknown_count: int skipped_count: int + paused_count: int = 0 + batch_state: str = "not_started" + batch_pause_reason_code: str | None = None + smtp_connection_count: int = 0 + smtp_reconnect_count: int = 0 preflight_count: int = 0 synchronous_send_policy: dict[str, Any] | None = None dry_run: bool = False @@ -239,7 +245,12 @@ class SendCampaignNowResult: "failed_count": self.failed_count, "outcome_unknown_count": self.outcome_unknown_count, "skipped_count": self.skipped_count, + "paused_count": self.paused_count, "preflight_count": self.preflight_count, + "batch_state": self.batch_state, + "batch_pause_reason_code": self.batch_pause_reason_code, + "smtp_connection_count": self.smtp_connection_count, + "smtp_reconnect_count": self.smtp_reconnect_count, "delivery_mode": "synchronous", "synchronous_send_policy": self.synchronous_send_policy or {}, "dry_run": self.dry_run, @@ -1070,48 +1081,91 @@ def send_campaign_now( jobs=jobs, policy=synchronous_policy, ) - # Queue state and its inbox notification become durable only after every - # message and the selected transport revision have passed preflight. This - # preserves late-ack recovery without leaving rejected work eligible for a - # background worker. - session.commit() - results: list[dict[str, Any]] = [] sent_count = 0 failed_count = 0 outcome_unknown_count = 0 skipped_after_queue = 0 - for job in jobs: - try: - 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"}: - sent_count += 1 - elif result.status == JobSendStatus.OUTCOME_UNKNOWN.value: - outcome_unknown_count += 1 - else: - skipped_after_queue += 1 - except Exception as exc: # keep sending other jobs and return per-job details - failed_count += 1 - results.append({"job_id": job.id, "status": "failed", "message": str(exc)}) + attempted_count = 0 + paused_count = 0 + pause_reason_code: str | None = None + batch_state = "ready" + batch_manager = _synchronous_smtp_batch_manager( + session, + jobs=jobs, + contexts=delivery_contexts, + ) + try: + with batch_manager as smtp_batch: + # Queue state becomes durable only after local and SMTP + # DNS/connectivity/TLS/auth preflight succeeds. + session.commit() + for index, job in enumerate(jobs): + attempted_count += 1 + try: + 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"}: + sent_count += 1 + elif result.status == JobSendStatus.OUTCOME_UNKNOWN.value: + outcome_unknown_count += 1 + else: + skipped_after_queue += 1 + except Exception as exc: + failed_count += 1 + results.append({"job_id": job.id, "status": "failed", "message": str(exc)}) + if isinstance(exc, SmtpSendError) and exc.systemic: + pause_reason_code = exc.reason_code or "smtp_systemic_failure" + paused_count = _pause_jobs_after_systemic_smtp_failure( + session, + campaign_id=job.campaign_id, + exclude_job_id=job.id, + reason_code=pause_reason_code, + ) + batch_state = "paused" + for remaining in jobs[index + 1 :]: + results.append( + { + "job_id": remaining.id, + "status": "paused", + "message": "Batch paused after a systemic SMTP failure.", + } + ) + break + smtp_connection_count = int(getattr(smtp_batch, "connection_count", 0) or 0) + smtp_reconnect_count = int(getattr(smtp_batch, "reconnect_count", 0) or 0) + except (MailProfileError, SmtpConfigurationError, SmtpSendError, OSError) as exc: + session.rollback() + reason_code = str(getattr(exc, "reason_code", "") or "smtp_batch_preflight_failed") + raise SynchronousSendRejected( + "SMTP batch preflight could not validate DNS, connectivity, TLS, and authentication; no message was sent.", + reason=reason_code, + eligible_count=len(jobs), + policy=synchronous_policy, + ) from exc return SendCampaignNowResult( campaign_id=campaign.id, version_id=version.id, - attempted_count=len(jobs), + attempted_count=attempted_count, sent_count=sent_count, failed_count=failed_count, outcome_unknown_count=outcome_unknown_count, skipped_count=queue_result.skipped_count + queue_result.blocked_count + skipped_after_queue, + paused_count=paused_count, + batch_state=batch_state, + batch_pause_reason_code=pause_reason_code, + smtp_connection_count=smtp_connection_count, + smtp_reconnect_count=smtp_reconnect_count, preflight_count=len(delivery_contexts), synchronous_send_policy=synchronous_policy.as_dict(), dry_run=False, @@ -1193,6 +1247,100 @@ def _preflight_synchronous_send_batch( return contexts +def _synchronous_smtp_batch_manager( + session: Session, + *, + jobs: list[CampaignJob], + contexts: dict[str, _SendJobDeliveryContext], +): + mail_items = [ + (job, contexts[job.id]) + for job in jobs + if DeliveryChannelPolicy(job.delivery_channel_policy).uses_mail + ] + if not mail_items: + return nullcontext(None) + first_job, first_context = mail_items[0] + envelope_froms = {str(context.envelope_from or "") for _job, context in mail_items} + transport_keys = { + ( + context.snapshot.mail_profile_id, + context.snapshot.smtp_transport_revision, + context.snapshot.smtp_server_id, + context.snapshot.smtp_credential_id, + ) + for _job, context in mail_items + } + if len(envelope_froms) != 1 or "" in envelope_froms or len(transport_keys) != 1: + raise SynchronousSendRejected( + "A synchronous SMTP batch requires one frozen sender and transport selection.", + reason="smtp_batch_transport_mismatch", + eligible_count=len(jobs), + ) + recipients = sorted( + { + recipient + for _job, context in mail_items + for recipient in context.envelope_recipients + } + ) + return mail_integration().campaign_smtp_batch( + session, + tenant_id=first_job.tenant_id, + campaign_id=first_job.campaign_id, + profile_id=first_context.snapshot.mail_profile_id, + envelope_from=str(first_context.envelope_from), + envelope_recipients=recipients, + from_header=_from_header_from_job(first_job), + expected_smtp_transport_revision=first_context.snapshot.smtp_transport_revision or "", + smtp_server_id=first_context.snapshot.smtp_server_id, + smtp_credential_id=first_context.snapshot.smtp_credential_id, + ) + + +def _pause_jobs_after_systemic_smtp_failure( + session: Session, + *, + campaign_id: str, + exclude_job_id: str, + reason_code: str, +) -> int: + reason = f"SMTP batch paused after systemic failure ({reason_code[:80]})." + changed = ( + session.query(CampaignJob) + .filter( + CampaignJob.campaign_id == campaign_id, + CampaignJob.id != exclude_job_id, + CampaignJob.queue_status == JobQueueStatus.QUEUED.value, + CampaignJob.send_status.in_( + [JobSendStatus.QUEUED.value, JobSendStatus.FAILED_TEMPORARY.value] + ), + ) + .update( + { + CampaignJob.queue_status: JobQueueStatus.PAUSED.value, + CampaignJob.last_error: reason, + }, + synchronize_session=False, + ) + ) + campaign = session.get(Campaign, campaign_id) + if changed and campaign is not None: + campaign.status = CampaignStatus.READY_TO_QUEUE.value + session.add(campaign) + audit_event( + session, + tenant_id=campaign.tenant_id if campaign is not None else None, + user_id=None, + action="campaign.smtp_batch_paused", + object_type="campaign", + object_id=campaign_id, + details={"reason_code": reason_code[:80], "paused_count": int(changed)}, + ) + session.commit() + return int(changed) + + def enqueue_existing_queued_jobs( session: Session, *, tenant_id: str, campaign_id: str ) -> int: @@ -3652,6 +3800,13 @@ def _send_claimed_mail_only_job( outcome_unknown = _record_smtp_send_error( session, job=job, attempt=attempt, exc=exc ) + if exc.systemic: + _pause_jobs_after_systemic_smtp_failure( + session, + campaign_id=job.campaign_id, + exclude_job_id=job.id, + reason_code=exc.reason_code or "smtp_systemic_failure", + ) if outcome_unknown is not None: return outcome_unknown raise diff --git a/tests/test_mail_profile_boundary.py b/tests/test_mail_profile_boundary.py index 11e953e..be52ef9 100644 --- a/tests/test_mail_profile_boundary.py +++ b/tests/test_mail_profile_boundary.py @@ -176,7 +176,7 @@ def test_new_execution_snapshot_stores_reference_and_evidence_not_transport_mate delivery=DeliveryConfig(), ) - assert payload["snapshot_version"] == "8" + assert payload["snapshot_version"] == "9" assert payload["mail_profile_id"] == "profile-1" assert "smtp" not in payload assert "imap" not in payload diff --git a/tests/test_send_now_response_security.py b/tests/test_send_now_response_security.py index 745612e..ef3d3a9 100644 --- a/tests/test_send_now_response_security.py +++ b/tests/test_send_now_response_security.py @@ -38,6 +38,11 @@ def test_send_now_omits_provider_and_recipient_text_from_response_and_audit() -> failed_count=1, outcome_unknown_count=0, skipped_count=0, + paused_count=1, + batch_state="paused", + batch_pause_reason_code="smtp_authentication_failed", + smtp_connection_count=1, + smtp_reconnect_count=0, preflight_count=2, synchronous_send_policy={ "max_recipient_jobs": 25, @@ -113,7 +118,12 @@ def test_send_now_omits_provider_and_recipient_text_from_response_and_audit() -> "failed_count", "outcome_unknown_count", "skipped_count", + "paused_count", "preflight_count", + "batch_state", + "batch_pause_reason_code", + "smtp_connection_count", + "smtp_reconnect_count", "delivery_mode", "dry_run", "synchronous_send_policy", diff --git a/tests/test_sending_jobs.py b/tests/test_sending_jobs.py index 058002b..68f45e6 100644 --- a/tests/test_sending_jobs.py +++ b/tests/test_sending_jobs.py @@ -2,7 +2,7 @@ from __future__ import annotations import unittest from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import ANY, patch from govoplan_campaign.backend.db.models import ( JobBuildStatus, @@ -17,6 +17,7 @@ from govoplan_campaign.backend.sending.jobs import ( _select_campaign_jobs_for_queue, _send_claimed_campaign_job, ) +from govoplan_campaign.backend.integrations import SmtpSendError class FakeSession: @@ -220,6 +221,62 @@ class CampaignQueueSelectionTests(unittest.TestCase): self.assertTrue(session.rolled_back) self.assertIn("Automatic retry is stopped", mark_unknown.call_args.kwargs["reason"]) + def test_systemic_smtp_failure_pauses_remaining_campaign_jobs(self): + job = SimpleNamespace( + id="job-1", + tenant_id="tenant-1", + campaign_id="campaign-1", + campaign_version_id="version-1", + delivery_channel_policy="mail", + resolved_recipients={"from": {"email": "sender@example.test"}}, + ) + context = SimpleNamespace( + snapshot=SimpleNamespace( + mail_profile_id="profile-1", + smtp_server_id=None, + smtp_credential_id=None, + smtp_transport_revision="revision-1", + delivery=SimpleNamespace(rate_limit=SimpleNamespace(messages_per_minute=60)), + ), + message_bytes=b"message", + envelope_from="sender@example.test", + envelope_recipients=["recipient@example.test"], + ) + + class Mail: + def wait_for_rate_limit(self, **_kwargs): + return None + + def send_campaign_email_bytes(self, *_args, **_kwargs): + raise SmtpSendError( + "SMTP authentication failed.", + systemic=True, + reason_code="smtp_authentication_failed", + ) + + with ( + patch("govoplan_campaign.backend.sending.jobs.mail_integration", return_value=Mail()), + patch("govoplan_campaign.backend.sending.jobs._record_attempt_start", return_value=SimpleNamespace(attempt_number=1)), + patch("govoplan_campaign.backend.sending.jobs._record_smtp_send_error", return_value=None), + patch("govoplan_campaign.backend.sending.jobs._pause_jobs_after_systemic_smtp_failure", return_value=4) as pause, + self.assertRaises(SmtpSendError), + ): + _send_claimed_campaign_job( + object(), # type: ignore[arg-type] + job=job, # type: ignore[arg-type] + claim_token="claim-1", + context=context, # type: ignore[arg-type] + use_rate_limit=False, + enqueue_imap_task=False, + ) + + pause.assert_called_once_with( + ANY, + campaign_id="campaign-1", + exclude_job_id="job-1", + reason_code="smtp_authentication_failed", + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_synchronous_delivery_policy.py b/tests/test_synchronous_delivery_policy.py index 7fa8d36..5f42d13 100644 --- a/tests/test_synchronous_delivery_policy.py +++ b/tests/test_synchronous_delivery_policy.py @@ -1,5 +1,6 @@ from __future__ import annotations +from contextlib import contextmanager from types import SimpleNamespace from unittest.mock import Mock, patch @@ -23,7 +24,9 @@ from govoplan_campaign.backend.sending.jobs import ( QueueCampaignResult, SynchronousSendRejected, _ensure_synchronous_send_count_allowed, + _pause_jobs_after_systemic_smtp_failure, _preflight_synchronous_send_batch, + _synchronous_smtp_batch_manager, queue_campaign_jobs, send_campaign_now, synchronous_send_candidate_jobs, @@ -364,6 +367,72 @@ def test_batch_preflight_checks_every_message_before_provider_effects() -> None: provider.send_campaign_email_bytes.assert_not_called() +def test_smtp_batch_manager_preflights_combined_recipients_and_frozen_transport() -> None: + snapshot = SimpleNamespace( + uses_mail=True, + mail_profile_id="profile-1", + smtp_transport_revision="revision-1", + smtp_server_id="server-1", + smtp_credential_id="credential-1", + ) + jobs = [ + SimpleNamespace( + id="one", + tenant_id="tenant-1", + campaign_id="campaign-1", + delivery_channel_policy="mail", + resolved_recipients={"from": {"email": "sender@example.test"}}, + ), + SimpleNamespace( + id="two", + tenant_id="tenant-1", + campaign_id="campaign-1", + delivery_channel_policy="mail", + resolved_recipients={"from": {"email": "sender@example.test"}}, + ), + ] + contexts = { + "one": SimpleNamespace(snapshot=snapshot, envelope_from="sender@example.test", envelope_recipients=["one@example.test"]), + "two": SimpleNamespace(snapshot=snapshot, envelope_from="sender@example.test", envelope_recipients=["two@example.test", "one@example.test"]), + } + state = SimpleNamespace(connection_count=1, reconnect_count=0) + + @contextmanager + def batch(_session, **kwargs): + batch.kwargs = kwargs + yield state + + provider = SimpleNamespace(campaign_smtp_batch=batch) + with patch("govoplan_campaign.backend.sending.jobs.mail_integration", return_value=provider): + with _synchronous_smtp_batch_manager(object(), jobs=jobs, contexts=contexts) as opened: # type: ignore[arg-type] + assert opened is state + + assert batch.kwargs["envelope_recipients"] == ["one@example.test", "two@example.test"] + assert batch.kwargs["expected_smtp_transport_revision"] == "revision-1" + + +def test_systemic_failure_pauses_only_remaining_queued_jobs() -> None: + session = Mock() + session.query.return_value.filter.return_value.update.return_value = 3 + campaign = SimpleNamespace(id="campaign-1", tenant_id="tenant-1", status="sending") + session.get.return_value = campaign + with patch("govoplan_campaign.backend.sending.jobs.audit_event") as audit: + paused = _pause_jobs_after_systemic_smtp_failure( + session, + campaign_id="campaign-1", + exclude_job_id="failed-job", + reason_code="smtp_authentication_failed", + ) + + assert paused == 3 + assert campaign.status == "ready_to_queue" + session.commit.assert_called_once_with() + assert audit.call_args.kwargs["details"] == { + "reason_code": "smtp_authentication_failed", + "paused_count": 3, + } + + def test_rejected_synchronous_preflight_rolls_back_staged_queue_before_audit() -> None: session = Mock() campaign = SimpleNamespace(id="campaign-1", current_version_id="version-1") diff --git a/webui/src/features/campaigns/ReviewSendPage.tsx b/webui/src/features/campaigns/ReviewSendPage.tsx index 39c2a4e..af5c7a6 100644 --- a/webui/src/features/campaigns/ReviewSendPage.tsx +++ b/webui/src/features/campaigns/ReviewSendPage.tsx @@ -932,10 +932,11 @@ export default function ReviewSendPage({ const sent = result.sent_count ?? 0; const failed = result.failed_count ?? 0; const unknown = result.outcome_unknown_count ?? 0; + const paused = result.paused_count ?? 0; setMessage( effectiveDryRun ? "i18n:govoplan-campaign.dry_run_finished_no_message_was_sent.c026c6cd" : - `Send finished. SMTP accepted ${String(sent)} message(s), failed ${String(failed)}, outcome unknown ${String(unknown)}.` + `Send finished. SMTP accepted ${String(sent)} message(s), failed ${String(failed)}, outcome unknown ${String(unknown)}, paused ${String(paused)}.` ); setSendConfirmOpen(false); await reload(); @@ -1765,6 +1766,13 @@ export default function ReviewSendPage({ {sendResult &&

i18n:govoplan-campaign.attempted.a9eb9c90 {String(sendResult.attempted_count ?? "—")}i18n:govoplan-campaign.smtp_accepted.a5d0dccc {String(sendResult.sent_count ?? "—")}i18n:govoplan-campaign.failed.fac9f871 {String(sendResult.failed_count ?? "—")}i18n:govoplan-campaign.outcome_unknown.4383023a {String(sendResult.outcome_unknown_count ?? 0)}i18n:govoplan-campaign.skipped.6b98496c {String(sendResult.skipped_count ?? "—")}.

+

+ SMTP batch: {humanize(String(sendResult.batch_state ?? "not_started"))} · connections {String(sendResult.smtp_connection_count ?? 0)} · reconnects {String(sendResult.smtp_reconnect_count ?? 0)} · paused {String(sendResult.paused_count ?? 0)}. +

+ {sendResult.batch_state === "paused" && + + Remaining messages were paused before SMTP after a systemic transport failure ({String(sendResult.batch_pause_reason_code ?? "smtp_systemic_failure")}). Review the Mail profile, then resume the campaign queue. + } {sendResultRows.length > 0 &&