How to send email in Python—Django, Flask, FastAPI, and reliable workers

Python can send an email in a few lines with smtplib, an HTTP request, or a provider SDK. That first successful message proves the credentials and payload are plausible. It does not prove the application can survive a duplicate form submission, a timeout after remote acceptance, a worker crash, a deployment during a send, or two delivery events arriving out of order.

This guide builds the production path around those failure modes. The examples are intentionally framework-neutral, then show where Django, Flask, FastAPI, Celery-style workers, and plain scripts fit. They use modern Python, HTTPX, the standard email package, PostgreSQL, and Email Bump’s transactional endpoint, but the boundaries apply to other providers and queue systems.

Choose HTTP, a provider SDK, or SMTP deliberately

Python has mature options for every transport. An HTTPS API usually provides structured errors, provider message IDs, metadata, and delivery events. A supported provider SDK can reduce request boilerplate. SMTP is a standards-based fit for an existing relay, a portable infrastructure contract, or a Django deployment already configured around email backends.

OPTION              GOOD FIT                               YOU STILL OWN
HTTP with HTTPX     stable provider API; small dependency       typing, timeout policy, status mapping
Provider SDK         provider-specific capabilities              upgrades, compatibility, error mapping
Django backend       existing Django email configuration         backend settings, MIME, connection policy
smtplib              managed SMTP relay or infrastructure rule   MIME, TLS, reply classes, pooling, timeouts
Local sendmail       controlled host-level mail queue             binary, permissions, portability, monitoring

Transport never replaces domain authentication, suppression handling,
idempotency, template safety, observability, or delivery-event processing.

SMTP is not inherently a development-only option. A managed SMTP provider can be operationally sound. The important distinction is where responsibility lives: your Python client only submits to the relay; the relay still owns receiver negotiation, reputation, and event reporting. A personal Gmail account and app password is not a substitute for a production transactional provider or managed organizational relay.

For local development, prefer a provider sandbox, a capture inbox, or Django’s console, file, or in-memory backends. Keep development identities unable to contact real customer lists. A test that passes by sending through an engineer’s personal mailbox creates a credential, privacy, and reproducibility problem.

Parse configuration once, before the process is ready

os.environ behaves like a mutable mapping of strings. Parse it at startup into a small immutable object and reject missing keys, malformed sender addresses, invalid URLs, and nonsensical limits before a web process or worker declares itself healthy.

from dataclasses import dataclass
from os import environ
from urllib.parse import urlparse


def required(name: str) -> str:
    value = environ.get(name, "").strip()
    if not value:
        raise RuntimeError(f"Missing required setting: {name}")
    return value


def positive_int(name: str, default: int) -> int:
    try:
        value = int(environ.get(name, str(default)))
    except ValueError as exc:
        raise RuntimeError(f"{name} must be an integer") from exc
    if value <= 0:
        raise RuntimeError(f"{name} must be positive")
    return value


@dataclass(frozen=True, slots=True)
class Settings:
    email_api_key: str
    email_from: str
    public_app_url: str
    worker_concurrency: int


def load_settings() -> Settings:
    settings = Settings(
        email_api_key=required("EMAIL_BUMP_API_KEY"),
        email_from=required("EMAIL_FROM"),
        public_app_url=required("PUBLIC_APP_URL").rstrip("/"),
        worker_concurrency=positive_int("EMAIL_WORKER_CONCURRENCY", 8),
    )
    if urlparse(settings.public_app_url).scheme != "https":
        raise RuntimeError("PUBLIC_APP_URL must use HTTPS")
    return settings

• Use the deployment platform’s secret store in production; never commit a populated .env file.
• Use separate API keys and sending identities for local, test, preview, and production environments.
• Validate configuration in the application factory, ASGI lifespan, Django startup, or worker boot path.
• Do not silently substitute a production recipient or sender when a variable is missing.
• Rotate credentials and make ownership clear by service and environment.
• Pass only required settings into child processes; do not treat the process environment as a typed global API.

Match the client to the execution model

Django under WSGI, Flask, a traditional Celery task, and most command-line scripts are naturally synchronous. Use a long-lived httpx.Client there. FastAPI and Django ASGI views can use httpx.AsyncClient when the entire call path is async. Reusing a client matters because it retains connection pooling; constructing one inside every hot-loop send forfeits much of that value.

CONTEXT                    USE                              AVOID
Flask / WSGI Django        httpx.Client                     asyncio.run() per request
FastAPI / ASGI             shared httpx.AsyncClient         blocking smtplib on event loop
Celery prefork task        sync client per worker process   client created before unsafe fork
Pure async worker          shared AsyncClient + semaphore   unbounded create_task loops
One-off management command context-managed sync client      a process-global never-closed client

Async improves concurrency while waiting for I/O.
It does not persist work, guarantee retries, or survive process termination.

If legacy blocking SMTP or an SDK must run inside an async service, moving one bounded call to a thread can protect the event loop, but it does not make cancellation or shutdown magically safe. Prefer an async-native HTTP transport at that boundary, or place blocking email work in a dedicated synchronous worker.

Send a first message through the Email Bump API

Install HTTPX with pip install httpx, set EMAIL_BUMP_API_KEY and EMAIL_FROM, and run this script with a restricted development key. Replace the recipient with an inbox you control. This confirms credentials, sender verification, payload shape, and the provider response before framework integration.

import os
import httpx

payload = {
    "from": os.environ["EMAIL_FROM"],
    "to": "you@example.com",
    "subject": "Hello from Python",
    "html": "<h1>Your Python integration works</h1><p>This came from Email Bump.</p>",
    "text": "Your Python integration works\n\nThis came from Email Bump.",
}

with httpx.Client(timeout=10.0) as client:
    response = client.post(
        "https://emailbump.com/api/v1/emails",
        headers={"Authorization": f"Bearer {os.environ['EMAIL_BUMP_API_KEY']}"},
        json=payload,
    )
    response.raise_for_status()
    result = response.json()

print(f"Email accepted: {result['id']}")

Provider acceptance is only the first milestone. The adapter below adds typed input, reusable connection pooling, timeouts, and explicit permanent, temporary, and ambiguous failure states.

Create one typed HTTP provider adapter

Keep credentials, endpoint shape, timeouts, response parsing, and error classification in one adapter. Give business services a small protocol so tests can replace transport without patching half the application.

from dataclasses import dataclass
from typing import Literal
import httpx


@dataclass(frozen=True, slots=True)
class OutboundEmail:
    message_id: str
    sender: str
    recipient: str
    subject: str
    html: str
    text: str
    reply_to: str | None = None


class EmailSubmitError(Exception):
    def __init__(
        self,
        message: str,
        *,
        kind: Literal["transient", "permanent", "ambiguous"],
        status: int | None = None,
        retry_after: float | None = None,
    ) -> None:
        super().__init__(message)
        self.kind = kind
        self.status = status
        self.retry_after = retry_after


class EmailBumpClient:
    def __init__(self, api_key: str, client: httpx.Client) -> None:
        self._api_key = api_key
        self._client = client

    def send(self, message: OutboundEmail) -> str:
        try:
            response = self._client.post(
                "https://emailbump.com/api/v1/emails",
                headers={"Authorization": f"Bearer {self._api_key}"},
                json={
                    "from": message.sender,
                    "to": message.recipient,
                    "reply_to": message.reply_to,
                    "subject": message.subject,
                    "html": message.html,
                    "text": message.text,
                },
            )
        except httpx.TransportError as exc:
            raise EmailSubmitError(
                "Email request ended without a response",
                kind="ambiguous",
            ) from exc

        if response.is_success:
            return str(response.json()["id"])
        if response.status_code in {408, 429} or response.status_code >= 500:
            raise EmailSubmitError(
                "Temporary provider response",
                kind="transient",
                status=response.status_code,
                retry_after=parse_retry_after(response.headers.get("retry-after")),
            )
        raise EmailSubmitError(
            "Provider rejected the request",
            kind="permanent",
            status=response.status_code,
        )

Construct the injected HTTPX client with an explicit timeout, for example separate connect, read, write, and pool limits. A single vague number hides where time was spent. Set connection-pool limits to match worker concurrency, and record latency without logging Authorization headers, full HTML, action tokens, or unnecessary recipient data.

timeout = httpx.Timeout(
    connect=3.0,
    read=8.0,
    write=8.0,
    pool=2.0,
)
limits = httpx.Limits(
    max_connections=20,
    max_keepalive_connections=10,
)

with httpx.Client(timeout=timeout, limits=limits) as http:
    provider = EmailBumpClient(settings.email_api_key, http)
    run_worker(provider)

# In ASGI, create one AsyncClient during lifespan and close it on shutdown.
# Keep an async adapter separate or share only pure response-classification code.

A response-less failure is ambiguous, not automatically safe to retry. The provider may have accepted the message just before the connection failed or the read timeout expired. Use a provider idempotency key when available. Otherwise store the ambiguous outcome and reconcile provider activity before resending a high-impact message.

Keep Django, Flask, and FastAPI endpoints thin

A public POST /send-email endpoint that accepts arbitrary to, from, subject, and html fields is an open-relay design. Define narrow commands such as invite_workspace_member, resend_verification, submit_contact_form, or send_invoice_copy. Authenticate and authorize the actor, load authoritative account data, and let the service select the sender and template.

@dataclass(frozen=True, slots=True)
class InviteMember:
    actor_id: UUID
    workspace_id: UUID
    invited_email: str


def invite_member(command: InviteMember, uow: UnitOfWork) -> UUID:
    with uow:
        membership = uow.memberships.get(command.workspace_id, command.actor_id)
        if membership is None or membership.role != "admin":
            raise PermissionError("Admin membership required")

        invite = uow.invites.create_or_reuse(
            workspace_id=command.workspace_id,
            email=normalize_email(command.invited_email),
        )
        job = uow.email_outbox.add_unique(
            dedupe_key=f"workspace-invite:{invite.id}:v2",
            purpose="workspace-invite",
            recipient=invite.email,
            template_key="workspace-invite",
            template_version="2",
            variables={"invite_id": str(invite.id)},
        )
        uow.commit()
        return job.public_id

A Django view can call this service inside its normal authentication and transaction policy. A Flask route can translate validated JSON into the command. A FastAPI route can depend on an authenticated principal and call an async equivalent. All three should return 202 Accepted once important work is durably owned—not claim that the email has been delivered.

• Limit request-body size and require the expected content type.
• Validate runtime input; Python type annotations do not validate data by themselves.
• Rate-limit public and recovery flows by account, purpose, network, and normalized recipient where appropriate.
• Fix contact-form From and To identities on the server; use a validated visitor address only as Reply-To.
• Return neutral recovery responses so callers cannot enumerate registered accounts.
• Resolve the recipient from trusted application state for receipts, alerts, and account messages.

Render safe HTML and a real plain-text alternative

Do not interpolate untrusted names, comments, or URLs into HTML with f-strings. Use Django templates with autoescaping or a Jinja environment configured for HTML autoescape. Define a variable contract for each template, normalize values once, construct links from a trusted HTTPS origin, and render both parts from the same semantic content.

from dataclasses import dataclass
from html import escape


@dataclass(frozen=True, slots=True)
class VerificationVariables:
    display_name: str
    verification_url: str
    expires_minutes: int


def render_verification(v: VerificationVariables) -> RenderedEmail:
    # In a real app, render a versioned Django/Jinja template with autoescape.
    safe_name = escape(v.display_name)
    safe_url = escape(v.verification_url, quote=True)
    return RenderedEmail(
        subject="Verify your email address",
        html=(
            f"<p>Hello {safe_name},</p>"
            f"<p><a href=\"{safe_url}\">Verify your email</a></p>"
            f"<p>This link expires in {v.expires_minutes} minutes.</p>"
        ),
        text=(
            f"Hello {v.display_name},\n\n"
            f"Verify your email: {v.verification_url}\n\n"
            f"This link expires in {v.expires_minutes} minutes."
        ),
    )

The small escape example demonstrates the boundary, not a recommendation to maintain large templates in Python strings. Store reviewable, versioned template files. Snapshot immutable facts needed by future retries, such as a receipt number and paid amount. Recheck mutable eligibility, such as whether an overdue invoice has since been paid, immediately before sending.

Build standards-correct MIME when you use SMTP

The standard-library email package should construct the message; smtplib should submit it. EmailMessage manages headers, Unicode, transfer encodings, and multipart structure. set_content creates the plain-text body, and add_alternative adds HTML as a multipart alternative.

from email.message import EmailMessage
import smtplib
import ssl


def build_message(rendered: RenderedEmail, sender: str, recipient: str) -> EmailMessage:
    message = EmailMessage()
    message["From"] = sender
    message["To"] = recipient
    message["Subject"] = rendered.subject
    message.set_content(rendered.text)
    message.add_alternative(rendered.html, subtype="html")
    return message


def submit_smtp(message: EmailMessage, settings: SmtpSettings) -> None:
    context = ssl.create_default_context()
    with smtplib.SMTP(settings.host, settings.port, timeout=10) as smtp:
        smtp.ehlo()
        smtp.starttls(context=context)
        smtp.ehlo()
        smtp.login(settings.username, settings.password)
        refused = smtp.send_message(message)
        if refused:
            raise RecipientRefused(refused)

# For implicit TLS, use SMTP_SSL(..., context=context).
# Never disable certificate verification to make a deployment pass.

Classify SMTP replies carefully and in the context of the relay contract. A 4xx reply is generally temporary and a 5xx reply generally permanent, but authentication, sender policy, and per-recipient failures deserve explicit handling. smtplib exceptions expose useful reply information; do not flatten every failure into one catch-all retry.

Connection reuse reduces repeated TLS and authentication cost, but a global SMTP connection can go stale and may not be safe across threads or forked workers. Open it within one worker process, serialize access unless the client contract says otherwise, reconnect after transport failure, cap messages or age per connection, and close cleanly. Match socket timeouts to the worker lease.

Commit important email with a transactional outbox

Your database transaction cannot atomically include a remote email API or SMTP relay. If the application commits an order and then crashes before sending, the receipt is lost. If it sends first and the order transaction rolls back, the customer receives a receipt for state the application does not own. The transactional outbox commits business state and email intent together, then lets a separate worker perform the network call.

CREATE TABLE email_outbox (
  id                  uuid PRIMARY KEY,
  public_id           uuid NOT NULL UNIQUE,
  dedupe_key          text NOT NULL UNIQUE,
  purpose             text NOT NULL,
  recipient           text NOT NULL,
  template_key        text NOT NULL,
  template_version    text NOT NULL,
  variables           jsonb NOT NULL,
  state               text NOT NULL DEFAULT 'pending',
  attempts            integer NOT NULL DEFAULT 0,
  next_attempt_at     timestamptz NOT NULL DEFAULT now(),
  lease_owner         text,
  lease_expires_at    timestamptz,
  provider_message_id text,
  last_error_class    text,
  created_at          timestamptz NOT NULL DEFAULT now(),
  accepted_at         timestamptz,
  delivered_at        timestamptz
);

CREATE INDEX email_outbox_claimable
  ON email_outbox (next_attempt_at, created_at)
  WHERE state IN ('pending', 'retry_wait');

Celery, RQ, Dramatiq, or a cloud queue can wake the worker, but publishing a task is not automatically atomic with a Django ORM or SQLAlchemy transaction. A crash can occur after database commit and before publish. Use an outbox publisher, a database-backed queue, or another documented transactional bridge. Pass a durable job ID to the queue rather than a huge rendered message or a secret-bearing model object.

Claim jobs with short transactions and expiring leases

Multiple worker processes should compete without submitting the same row concurrently. PostgreSQL FOR UPDATE SKIP LOCKED is one practical primitive. Claim and commit quickly, then render and call the remote provider outside the transaction. Holding a row lock across network I/O turns an email outage into a database contention incident.

WITH candidate AS (
  SELECT id
  FROM email_outbox
  WHERE (
    state IN ('pending', 'retry_wait') AND next_attempt_at <= now()
  ) OR (
    state = 'leased' AND lease_expires_at < now()
  )
  ORDER BY next_attempt_at, created_at
  FOR UPDATE SKIP LOCKED
  LIMIT 1
)
UPDATE email_outbox AS job
SET state = 'leased',
    lease_owner = %(worker_id)s,
    lease_expires_at = now() + interval '60 seconds',
    attempts = attempts + 1
FROM candidate
WHERE job.id = candidate.id
RETURNING job.*;

Choose a lease longer than normal rendering and submission, renew it for legitimately long work, and require the same lease owner in the final UPDATE. If a process freezes past expiry, another worker can recover the row. Leases reduce simultaneous work; they cannot create exactly-once behavior across a remote network, so semantic and provider idempotency still matter.

def process_job(job: EmailJob, worker_id: str, provider: EmailProvider) -> None:
    eligibility = recheck_eligibility(job)
    if not eligibility.allowed:
        outbox.suppress(job.id, worker_id, eligibility.reason)
        return

    rendered = render_versioned_template(job)
    try:
        provider_id = provider.send(to_outbound(job, rendered))
    except EmailSubmitError as exc:
        if exc.kind == "ambiguous":
            outbox.mark_ambiguous(job.id, worker_id)
        elif exc.kind == "permanent":
            outbox.mark_permanent_failure(job.id, worker_id, exc.status)
        else:
            outbox.schedule_retry(job.id, worker_id, next_attempt(job, exc))
    else:
        outbox.mark_accepted(job.id, worker_id, provider_id)

# Every mutation is conditional on lease_owner = worker_id.
# Losing the lease means this worker must not finalize the row.

Bound concurrency and preserve cancellation

Email submission is I/O-bound, but unbounded concurrency can exhaust provider limits, database connections, sockets, memory, or shutdown time. For synchronous prefork workers, set process concurrency from measurements. For a pure async worker, combine a bounded claim batch with a semaphore or TaskGroup, and keep HTTP connection limits aligned.

import asyncio


async def handle(job: EmailJob, semaphore: asyncio.Semaphore) -> None:
    async with semaphore:
        try:
            await process_job(job)
        except asyncio.CancelledError:
            await asyncio.shield(release_or_shorten_lease(job.id))
            raise  # cancellation must continue to propagate


async def process_batch(jobs: list[EmailJob], limit: int) -> None:
    semaphore = asyncio.Semaphore(limit)
    async with asyncio.TaskGroup() as group:
        for job in jobs:
            group.create_task(handle(job, semaphore))

CancelledError is a control-flow signal. Catch it only when cleanup is necessary, then re-raise it. Swallowing cancellation can make graceful shutdown hang or make a TaskGroup believe a job completed. TaskGroup also cancels remaining tasks when one child fails with a non-cancellation exception, so decide whether batch-wide fail-fast semantics are appropriate; independent jobs may be better isolated and reported individually.

Context variables can attach a trace ID and email job ID to logs across async tasks without passing them through every call. They are process memory, not durable state. Persist identifiers on the outbox or queue message so a retry in another process retains the same correlation.

Retry temporary failures without creating a retry storm

Authentication failures, malformed payloads, unverified senders, invalid recipients, and policy rejections usually require repair. Rate limits, many 5xx responses, and some connection failures may recover. Retry only the classified temporary set, honor a valid Retry-After, add jitter, cap attempts and total job age, and move exhausted jobs to a visible dead-letter state.

from datetime import datetime, timedelta, timezone
import random


def retry_at(attempt: int, retry_after: float | None = None) -> datetime:
    if retry_after is None:
        base_seconds = 2.0
        cap_seconds = 30 * 60.0
        maximum = min(cap_seconds, base_seconds * (2 ** min(attempt - 1, 20)))
        delay = random.uniform(maximum * 0.5, maximum)
    else:
        delay = max(0.0, retry_after)
    return datetime.now(timezone.utc) + timedelta(seconds=delay)

POLICY = """
Recheck current product state before every delayed send.
Use different age limits for resets, receipts, and lifecycle reminders.
Never blindly retry an ambiguous high-impact submission.
Alert on dead letters and oldest ready-job age—not only queue depth.
"""

Jitter spreads worker retries after an outage. A circuit breaker can pause new submissions during a clear provider-wide failure, but queued intent must remain visible. Separate web readiness from email subsystem health so a provider incident does not disappear behind green application probes.

Design idempotency at three layers

SOURCE EVENT
stripe_event_id, webhook_event_id, or command id
Prevents the same input from being interpreted twice.

MESSAGE INTENT
payment-receipt:{payment_id}:v4
Prevents two inputs from creating the same customer communication.

PROVIDER REQUEST
email_outbox.id where the provider supports idempotency
Prevents duplicate remote acceptance after response loss.

ATTEMPT EVIDENCE
email_outbox.id + attempt number
Correlates logs and timings without changing message identity.

A random UUID distinguishes one attempt but cannot recognize repeated business intent. A dedupe key based only on recipient or subject blocks legitimate future messages. Put a unique constraint on the semantic intent, create it in the same transaction as the business event, and retain a stable provider idempotency identity across retries.

Shut down workers as a state transition

Python installs signal handlers in the main thread, and Python-level handlers run in that thread. Keep the handler small: set a stop flag or event. The normal worker loop should stop claiming, wait for in-flight work up to the platform deadline, release or shorten unfinished leases, close HTTP, SMTP, broker, and database resources, then return control to the process supervisor.

from threading import Event
import signal

stop_requested = Event()


def request_stop(signum: int, frame: object) -> None:
    stop_requested.set()


signal.signal(signal.SIGTERM, request_stop)
signal.signal(signal.SIGINT, request_stop)

try:
    while not stop_requested.is_set():
        job = outbox.claim(worker_id, wait_seconds=1)
        if job is not None:
            process_job(job, worker_id, provider)
finally:
    # Stop claiming first; drain bounded in-flight work before this point.
    outbox.release_owned_leases(worker_id)
    provider.close()
    database.close()

Do not perform complicated database or network work directly in the signal handler. Celery and other supervisors have their own shutdown lifecycle; integrate with it instead of installing competing handlers blindly. Even graceful code cannot rely on cleanup always running—a hard kill or machine loss must still recover through expiring leases and idempotent attempts.

Treat provider webhooks as delivery evidence

The synchronous send response generally proves provider acceptance, not delivery to the receiving server, placement in an inbox, or human reading. Verify delivery webhook signatures over the exact bytes required by the provider, enforce a body limit and timestamp tolerance, insert the provider event ID uniquely, and return 2xx only after durable storage.

POST /webhooks/email
  → read raw body with a strict size limit
  → verify timestamp and signature before JSON trust
  → insert provider_event_id ON CONFLICT DO NOTHING
  → commit, then return 2xx

EVENT WORKER
  → map provider_message_id to email_outbox
  → apply provider-specific, order-safe state rules
  → record delivered, deferred, bounced, complained, or suppressed
  → update suppression and alert policy
  → retain raw payload only as privacy and audit rules require

Events may be duplicated or reordered. Do not let an older deferred event erase terminal delivery evidence without a provider-specific reason. A bounced receipt also does not undo a successful payment; communication state and product state remain separate ledgers.

Test the failure model, not just the happy response

Inject the provider, clock, random source, and repositories where practical. unittest.mock can patch a dependency where the code under test looks it up, and AsyncMock can model awaited collaborators. Prefer an HTTP mock transport, local SMTP sink, or restricted provider sandbox for integration tests over sending to real people.

CONFIGURATION
missing secret · invalid URL · bad integer · environment isolation

COMMAND
unauthorized actor · invalid recipient · duplicate and concurrent command

TEMPLATE
HTML escaping · plain text · Unicode · long URL · missing variable

OUTBOX
business rollback · unique intent · expired lease reclaim · stale suppression

PROVIDER
202 accepted · 400 permanent · 429 Retry-After · 500 transient
connect failure · read timeout after possible acceptance · malformed response

ASYNC / PROCESS
CancelledError during submit · TaskGroup failure · SIGTERM idle and in-flight

EVENTS
invalid signature · duplicate event · reordered state · unknown message ID

Assert database transitions and observable side effects, not only the return value. A timeout test should prove the job becomes ambiguous. A cancellation test should prove cleanup runs and cancellation propagates. A SIGTERM test should prove no new claim is made. A duplicate webhook should result in one durable event.

Production checklist

• Choose HTTPS, a provider SDK, Django backend, or SMTP from operational requirements.
• Parse immutable configuration before reporting a web process or worker ready.
• Match sync or async clients to the runtime and reuse connection pools safely.
• Centralize transport, timeout, response parsing, and error classification.
• Expose authorized business commands instead of an arbitrary send endpoint.
• Render versioned, escaped HTML and a reviewed plain-text alternative.
• Commit important business state and email intent in one transaction.
• Bridge the database and task broker with an outbox or equivalent guarantee.
• Claim jobs in short transactions with leases and conditional final updates.
• Bound process or task concurrency from provider, pool, and shutdown limits.
• Preserve asyncio cancellation and treat hard process loss as inevitable.
• Retry only temporary failures with jitter, caps, and current-state checks.
• Preserve ambiguous submissions and reconcile before risky resends.
• Deduplicate source events, message intent, and provider submissions separately.
• Stop claiming, drain work, and close resources during graceful shutdown.
• Verify, deduplicate, and order delivery events before updating evidence.
• Persist correlation IDs and keep secrets and message content out of logs.
• Test transport, database, duplicate, cancellation, and termination failures.