GovOPlaN/govoplan-campaign
1
0
Fork 0
Code Issues 46 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(campaign): bound synchronous delivery

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 08:21:42 +02:00
parent a8c0750dd7
commit 7e1660344d
10 changed files with 689 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

291
src/govoplan_campaign/backend/sending/jobs.py
View File

@@ -28,7 +28,17 @@ from govoplan_campaign.backend.db.models import (
ImapAppendAttempt,
SendAttempt,
)
from govoplan_campaign.backend.sending.execution import ExecutionSnapshot, ExecutionSnapshotError, ensure_execution_snapshot
from govoplan_campaign.backend.delivery_policy import (
CampaignDeliveryPolicyError,
SynchronousSendPolicy,
effective_synchronous_send_policy,
)
from govoplan_campaign.backend.sending.execution import (
ExecutionSnapshot,
ExecutionSnapshotError,
ensure_execution_snapshot,
profile_delivery_summary,
)
from govoplan_campaign.backend.runtime import get_registry
from govoplan_campaign.backend.integrations import (
ImapAppendError,
@@ -49,6 +59,29 @@ class SendJobError(RuntimeError):
pass
class SynchronousSendRejected(QueueingError):
def __init__(
self,
message: str,
*,
reason: str,
eligible_count: int | None = None,
policy: SynchronousSendPolicy | None = None,
) -> None:
super().__init__(message)
self.reason = reason
self.eligible_count = eligible_count
self.policy = policy
def audit_details(self) -> dict[str, Any]:
return {
"delivery_mode": "synchronous",
"rejection_reason": self.reason,
"eligible_recipient_job_count": self.eligible_count,
"synchronous_send_policy": self.policy.as_dict() if self.policy is not None else None,
}
@dataclass(frozen=True, slots=True)
class QueueCampaignResult:
campaign_id: str
@@ -80,6 +113,8 @@ class SendCampaignNowResult:
failed_count: int
outcome_unknown_count: int
skipped_count: int
preflight_count: int = 0
synchronous_send_policy: dict[str, Any] | None = None
dry_run: bool = False
results: list[dict[str, Any]] | None = None
@@ -92,6 +127,9 @@ class SendCampaignNowResult:
"failed_count": self.failed_count,
"outcome_unknown_count": self.outcome_unknown_count,
"skipped_count": self.skipped_count,
"preflight_count": self.preflight_count,
"delivery_mode": "synchronous",
"synchronous_send_policy": self.synchronous_send_policy or {},
"dry_run": self.dry_run,
"results": self.results or [],
}
@@ -444,6 +482,115 @@ def _select_campaign_jobs_for_queue(
return queued, skipped_count, blocked_count
def synchronous_send_candidate_jobs(
version: CampaignVersion,
jobs: list[CampaignJob],
*,
include_warnings: bool = True,
) -> list[CampaignJob]:
"""Return the exact persisted job set an immediate send may attempt.
This projection is deliberately side-effect free so the delivery-options
API and adaptive documentation can explain the current built execution
without changing queue state. The final command repeats the projection
after queueing and before any SMTP effect.
"""
allowed_validation = _queue_validation_statuses(include_warnings=include_warnings)
reviewed_needs_review_keys = _reviewed_needs_review_keys(version)
candidates: list[CampaignJob] = []
for job in jobs:
if (
job.queue_status == JobQueueStatus.QUEUED.value
and job.send_status == JobSendStatus.QUEUED.value
):
candidates.append(job)
continue
if job.send_status in INITIAL_QUEUE_SKIPPED_SEND_STATUSES:
continue
if job.queue_status in INITIAL_QUEUE_SKIPPED_QUEUE_STATUSES:
continue
validation_allowed = job.validation_status in allowed_validation or (
job.validation_status == JobValidationStatus.NEEDS_REVIEW.value
and _job_review_key(job) in reviewed_needs_review_keys
)
if job.build_status != JobBuildStatus.BUILT.value or not validation_allowed:
continue
if not job.eml_local_path and not job.eml_storage_key:
continue
candidates.append(job)
return candidates
def synchronous_send_options(
session: Session,
*,
tenant_id: str,
campaign_id: str,
version_id: str | None = None,
include_warnings: bool = True,
) -> dict[str, Any]:
"""Describe the actor-independent effective delivery-mode constraints."""
campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id)
version = _get_version_for_campaign(session, campaign, version_id=version_id)
jobs = _campaign_jobs_for_version(session, tenant_id=tenant_id, version_id=version.id)
candidates = synchronous_send_candidate_jobs(version, jobs, include_warnings=include_warnings)
try:
policy = effective_synchronous_send_policy(session, tenant_id=tenant_id)
except CampaignDeliveryPolicyError as exc:
return {
"campaign_id": campaign.id,
"version_id": version.id,
"worker_queue_available": bool(_celery_enabled()),
"synchronous_send": {
"allowed": False,
"reason": "policy_configuration_invalid",
"message": str(exc),
"eligible_recipient_job_count": len(candidates),
"policy": {},
},
}
ready = _version_is_validated_and_locked(version) and bool(version.build_summary)
eligible_count = len(candidates)
if not ready:
reason = "version_not_ready"
elif eligible_count == 0:
reason = "no_eligible_recipient_jobs"
elif eligible_count > policy.max_recipient_jobs:
reason = "recipient_limit_exceeded"
else:
reason = None
return {
"campaign_id": campaign.id,
"version_id": version.id,
"worker_queue_available": bool(_celery_enabled()),
"synchronous_send": {
"allowed": reason is None,
"reason": reason,
"eligible_recipient_job_count": eligible_count,
"policy": policy.as_dict(),
},
}
def _campaign_jobs_for_version(
session: Session,
*,
tenant_id: str,
version_id: str,
) -> list[CampaignJob]:
return (
session.query(CampaignJob)
.filter(
CampaignJob.tenant_id == tenant_id,
CampaignJob.campaign_version_id == version_id,
)
.order_by(CampaignJob.entry_index.asc())
.all()
)
def _persist_campaign_queue(
session: Session,
*,
@@ -546,13 +693,36 @@ def send_campaign_now(
) -> SendCampaignNowResult:
"""Queue and send all eligible jobs for one campaign synchronously.
The service currently has no server-side job-count bound. Callers must use
the worker queue for ordinary batches and reserve this path for deliberately
small interactive runs.
The effective deployment/tenant limit is checked against the immutable,
persisted job set before queue state changes. After queueing, every message
and transport revision is preflighted before the first SMTP effect starts.
"""
campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id)
version = _get_current_version(session, campaign, version_id=version_id)
_ensure_version_validated_and_locked(version)
_ensure_campaign_execution_snapshot(session, version)
try:
synchronous_policy = effective_synchronous_send_policy(session, tenant_id=tenant_id)
except CampaignDeliveryPolicyError as exc:
raise SynchronousSendRejected(
f"Synchronous Campaign delivery is disabled by an invalid policy configuration: {exc}",
reason="policy_configuration_invalid",
) from exc
initial_jobs = _campaign_jobs_for_queue(
session,
tenant_id=tenant_id,
version_id=version.id,
)
initial_candidates = synchronous_send_candidate_jobs(
version,
initial_jobs,
include_warnings=include_warnings,
)
_ensure_synchronous_send_count_allowed(
len(initial_candidates),
policy=synchronous_policy,
)
queue_result = queue_campaign_jobs(
session,
@@ -572,20 +742,31 @@ def send_campaign_now(
failed_count=0,
outcome_unknown_count=0,
skipped_count=queue_result.skipped_count + queue_result.blocked_count,
preflight_count=len(initial_candidates),
synchronous_send_policy=synchronous_policy.as_dict(),
dry_run=True,
results=[queue_result.as_dict()],
)
jobs = (
session.query(CampaignJob)
.filter(
CampaignJob.tenant_id == tenant_id,
CampaignJob.campaign_version_id == version.id,
CampaignJob.queue_status == JobQueueStatus.QUEUED.value,
CampaignJob.send_status.in_([JobSendStatus.QUEUED.value, JobSendStatus.FAILED_TEMPORARY.value]),
jobs = [
job
for job in _campaign_jobs_for_version(
session,
tenant_id=tenant_id,
version_id=version.id,
)
.order_by(CampaignJob.entry_index.asc())
.all()
if job.queue_status == JobQueueStatus.QUEUED.value
and job.send_status == JobSendStatus.QUEUED.value
]
# Repeat the hard bound against the post-queue set. This closes the window
# where a concurrent queue operation could otherwise enlarge an immediate
# run between the initial decision and the first provider effect.
_ensure_synchronous_send_count_allowed(len(jobs), policy=synchronous_policy)
delivery_contexts = _preflight_synchronous_send_batch(
session,
version=version,
jobs=jobs,
policy=synchronous_policy,
)
results: list[dict[str, Any]] = []
@@ -595,13 +776,19 @@ def send_campaign_now(
skipped_after_queue = 0
for job in jobs:
try:
result = send_campaign_job(
session,
job_id=job.id,
dry_run=False,
use_rate_limit=use_rate_limit,
enqueue_imap_task=enqueue_imap_task,
)
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_dict = result.as_dict()
results.append(result_dict)
if result.status in {JobSendStatus.SMTP_ACCEPTED.value, "already_accepted"}:
@@ -622,11 +809,75 @@ def send_campaign_now(
failed_count=failed_count,
outcome_unknown_count=outcome_unknown_count,
skipped_count=queue_result.skipped_count + queue_result.blocked_count + skipped_after_queue,
preflight_count=len(delivery_contexts),
synchronous_send_policy=synchronous_policy.as_dict(),
dry_run=False,
results=results,
)
def _ensure_synchronous_send_count_allowed(
eligible_count: int,
*,
policy: SynchronousSendPolicy,
) -> None:
if eligible_count <= 0:
raise SynchronousSendRejected(
"No eligible built recipient messages are available for synchronous delivery.",
reason="no_eligible_recipient_jobs",
eligible_count=eligible_count,
policy=policy,
)
if eligible_count > policy.max_recipient_jobs:
raise SynchronousSendRejected(
(
f"This built run contains {eligible_count} eligible recipient jobs, above the effective "
f"synchronous limit of {policy.max_recipient_jobs}. Queue it for background workers instead."
),
reason="recipient_limit_exceeded",
eligible_count=eligible_count,
policy=policy,
)
def _preflight_synchronous_send_batch(
session: Session,
*,
version: CampaignVersion,
jobs: list[CampaignJob],
policy: SynchronousSendPolicy,
) -> dict[str, _SendJobDeliveryContext]:
"""Freeze and validate every local input before the first SMTP effect."""
contexts: dict[str, _SendJobDeliveryContext] = {}
try:
for job in jobs:
state_result = _preflight_send_campaign_job(session, job, dry_run=True)
if state_result is not None:
raise SendJobError(
f"Job {job.id} changed to {state_result.status} during synchronous preflight"
)
contexts[job.id] = _send_job_delivery_context(session, job)
profile_summary = profile_delivery_summary(session, version)
expected_revision = next(iter(contexts.values())).snapshot.smtp_transport_revision
if profile_summary.get("smtp_transport_revision") != expected_revision:
raise SendJobError(
"The selected Mail profile changed after this execution was built. Revalidate and rebuild before delivery."
)
except (ExecutionSnapshotError, MailProfileError, SmtpConfigurationError, SendJobError, OSError) as exc:
raise SynchronousSendRejected(
(
"Synchronous preflight stopped before contacting SMTP because one or more frozen message "
"inputs or the selected Mail profile no longer match the reviewed build. Rebuild and review "
"the version before trying again."
),
reason="batch_preflight_failed",
eligible_count=len(jobs),
policy=policy,
) from exc
return contexts
def enqueue_existing_queued_jobs(session: Session, *, tenant_id: str, campaign_id: str) -> int:
if not _celery_enabled():
return 0