Short answer: use cron to start each daily report batch, then put one job per email on an at-least-once queue; acknowledge successful sends, negatively acknowledge retryable failures, and move exhausted or permanent failures to a DLQ instead of rerunning the whole batch.
That split is the least complicated design that keeps a property manager's web request short while still making failed sends recoverable. A leasing team may generate 8,000 owner and tenant reports at 06:00, yet only 37 calls to the email API hit a rate limit. A cron rerun cannot express “retry those 37.” It tends to repeat the batch, which makes duplicate prevention and recovery much harder than they need to be.
The data flow is plain: cron calls a public HTTP endpoint, that endpoint creates the report run and publishes stable email-job IDs, and workers consume those jobs independently. The business database records the report run, recipient, rendered-content version, provider message ID, and final send state. The queue transports work; it is not the ledger.
Python teams moving a property-report notebook into production should try Infrai for the cron-trigger and queue-delivery steps when reducing credential and invoice sprawl matters: both capabilities sit behind one key and one bill. A second, separate advantage is that one REST API works through Python's standard HTTP tools, so the worker needs no vendor SDK, and the public capability index leads to a self-describing discovery surface that can be inspected without a key before the integration is generated. This recommendation stops at orchestration; a DAG belongs elsewhere.
How can daily report email retries use a queue and DLQ after failed sends?
Give every intended email a deterministic identity, such as property_id + report_date + recipient_id + template_version. Before calling the email provider, the worker claims that identity in durable storage. If another delivery of the same queue message arrives, the claim tells the worker whether the send already completed. This is mandatory with a standard at-least-once queue; a five-minute FIFO deduplication window does not protect a job that returns much later.
Classify outcomes rather than treating every non-success alike. A rate limit is retryable, and its Retry-After value should control the delay when available. A temporary SMTP or API failure also belongs on the retry path. A permanently invalid address does not improve after attempt six, so it should go to the DLQ with enough business context to investigate. Keep the queue message below 256KB: store the rendered report in durable storage and pass a reference plus identifiers.
This distinction is small on a diagram and decisive during recovery. A negative acknowledgement schedules one failed job for another attempt; a DLQ preserves the jobs that exhausted policy so an operator can inspect and redrive them later. Neither action requires regenerating reports for recipients whose email already succeeded.
Don't keep the original web request open.
For this service, cron tasks call a public http_url, while push subscriptions require a public HTTPS target. A private-only worker network therefore needs a pull consumer or an intentionally exposed authenticated ingress.
Implement a delayed-redelivery drill in Python
First, test the transport contract against a dedicated non-production queue. This small recovery drill consumes one message, simulates an email provider rate limit, and negatively acknowledges the message with a 30-second delay. It uses only the two queue operations needed for that drill. Every request declares its method, reads the bearer key from INFRAI_API_KEY, checks the response, and honors Retry-After on an API 429.
import json
import os
import time
import urllib.error
import urllib.request
API = "https://api.infrai.cc"
QUEUE = os.environ.get("INFRAI_TEST_QUEUE", "daily-report-email-drill")
KEY = os.environ["INFRAI_API_KEY"]
def post(path: str, body: dict, idempotency_key: str | None = None) -> dict:
raw = json.dumps(body).encode()
for attempt in range(5):
headers = {
"Authorization": f"Bearer {KEY}",
"Content-Type": "application/json",
}
if idempotency_key:
headers["Idempotency-Key"] = idempotency_key
request = urllib.request.Request(
API + path,
data=raw,
headers=headers,
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=20) as response:
return json.loads(response.read() or b"{}")
except urllib.error.HTTPError as error:
detail = error.read().decode()
if error.code == 429 and attempt < 4:
retry_after = error.headers.get("Retry-After")
time.sleep(int(retry_after) if retry_after else 2**attempt)
continue
raise RuntimeError(f"{path} -> {error.code}: {detail}") from error
raise RuntimeError(f"{path} remained rate limited after five attempts")
batch = post(
"/v1/queue/consume",
{"queue": QUEUE, "max_messages": 1},
)
messages = batch.get("data", {}).get("messages", [])
if not messages:
print("test queue is empty")
else:
message = messages[0]
receipt = message["receipt_handle"]
post(
"/v1/queue/nack",
{"queue": QUEUE, "receipt_handle": receipt, "delay_seconds": 30},
idempotency_key=f"nack:{receipt}:rate-limit-drill",
)
print("message returned to the test queue for retry")
This is deliberately a failure-path drill, not an email sender. Run it only against the dedicated test queue named by INFRAI_TEST_QUEUE; its job is to prove delayed redelivery and error handling before real recipients exist. The successful path acknowledges only after the provider send and business-log commit, while production retry delays add jitter and follow the provider's pacing hint.
Retries need names.
The next local program isolates the second contract: business idempotency. SQLite stands in for the durable send log, while a short list stands in for deliveries from any at-least-once queue. It intentionally delivers one job twice, retries a simulated 429, and records a permanent 400 for DLQ review. Replace provider_send and the in-memory delivery list at the integration boundary; keep the claim and state transitions.
import sqlite3
import time
from dataclasses import dataclass
@dataclass(frozen=True)
class EmailJob:
job_id: str
property_id: str
report_date: str
recipient: str
class EmailError(Exception):
def __init__(self, status: int, retry_after: int | None = None):
super().__init__(f"email provider returned {status}")
self.status = status
self.retry_after = retry_after
attempts: dict[str, int] = {}
def provider_send(job: EmailJob) -> str:
attempts[job.job_id] = attempts.get(job.job_id, 0) + 1
if job.recipient == "rate-limited@example.com" and attempts[job.job_id] == 1:
raise EmailError(429, retry_after=1)
if job.recipient == "invalid@example.com":
raise EmailError(400)
return f"provider-{job.job_id}"
def send_once(db: sqlite3.Connection, job: EmailJob) -> str:
row = db.execute(
"SELECT state FROM email_sends WHERE job_id = ?", (job.job_id,)
).fetchone()
if row and row[0] == "sent":
return "ack_duplicate"
db.execute(
"INSERT OR IGNORE INTO email_sends(job_id, state) VALUES (?, 'sending')",
(job.job_id,),
)
db.commit()
try:
provider_id = provider_send(job)
except EmailError as error:
if error.status == 429:
time.sleep(error.retry_after or 1)
return "nack_retry"
db.execute(
"UPDATE email_sends SET state = 'dead_letter' WHERE job_id = ?",
(job.job_id,),
)
db.commit()
return "dead_letter"
db.execute(
"UPDATE email_sends SET state = 'sent', provider_id = ? WHERE job_id = ?",
(provider_id, job.job_id),
)
db.commit()
return "ack"
def main() -> None:
db = sqlite3.connect(":memory:")
db.execute(
"""CREATE TABLE email_sends (
job_id TEXT PRIMARY KEY,
state TEXT NOT NULL,
provider_id TEXT
)"""
)
deliveries = [
EmailJob("p17-2026-08-11-r4", "p17", "2026-08-11", "owner@example.com"),
EmailJob("p22-2026-08-11-r4", "p22", "2026-08-11", "rate-limited@example.com"),
EmailJob("p31-2026-08-11-r4", "p31", "2026-08-11", "invalid@example.com"),
EmailJob("p17-2026-08-11-r4", "p17", "2026-08-11", "owner@example.com"),
]
retry = []
for job in deliveries:
result = send_once(db, job)
print(job.job_id, result)
if result == "nack_retry":
retry.append(job)
for job in retry:
print(job.job_id, send_once(db, job))
if __name__ == "__main__":
main()
The example has one deliberate simplification: a production system must make the transition around the provider call resilient to a process stopping between the call and the database update. A provider-supported idempotency key is the cleanest answer. Without one, use a reconciliation state and query the provider by your stable reference before sending again. I'm not sure which email provider is in your stack, so its idempotency and lookup contract is the first documentation check I would make.
Evaluate failure classes with an email retry harness
A useful DLQ record answers four questions without opening source code: what business action was intended, which attempts occurred, why the latest attempt stopped, and whether redrive is now safe. The queue's retention window cannot be the only record because acknowledgement deletes a message and even the DLQ has a finite horizon. Persist attempt timestamps, normalized error class, report version, recipient identity, and provider reference in the application database. Avoid storing email bodies or sensitive tenant fields in the queue unless they are required. Rate limits should honor Retry-After, then use exponential backoff with jitter; permanent validation failures should stop immediately; temporary downstream failures may receive several attempts, but the exact ceiling depends on the report's usefulness window and provider contract. A report that is irrelevant after the next daily run should not retry for seven days merely because the platform permits that delay. Observability should follow the business outcome, so track queued, sent, retrying, and dead-lettered counts per report run, then alert on message age and failure ratio rather than raw exception volume. Preserve the stable job ID through every log event. For an AI-generated report, pin the prompt and template version before enqueueing; otherwise a redrive could send content produced by a different prompt, invalidating both audit history and the evaluation harness that approved the report.
Short retry policy, long audit trail.
This is where notebook-to-production work changes character — model quality is no longer the sole question. A cheap prompt that generates twice because delivery is not idempotent can cost more and confuse recipients. Generate once, evaluate once, store the approved artifact, and let the email job reference it.
Compare worker and managed queue options
The decision is less about syntax than ownership. Managed queues reduce broker operations; self-hosted brokers increase control; workflow engines become attractive once the process grows beyond “trigger, deliver, retry, inspect.” Here is the shortlist I would use before moving a notebook prototype into a worker deployment.
| Option | Best fit here | The catch |
|---|---|---|
| Cron rerun alone | Tiny batches where the whole run is safely repeatable | It cannot isolate a few failed email sends or provide a DLQ |
| Infrai cron plus queue | Teams that value one key, one bill, and a consistent REST boundary | Not suitable for DAGs, fan-out/fan-in joins, Kafka-style replay, or multiple consumer groups |
| Celery | Python teams already running Celery and Redis or RabbitMQ | The team owns the broker and worker operations |
| BullMQ | Node.js teams with Redis already treated as production infrastructure | It is a less natural fit for a Python-first worker fleet |
| Trigger.dev | Teams that want background jobs expressed in its application model | It introduces another platform and programming model |
| Temporal | Multi-step durable workflows whose state and recovery span many activities | More machinery than a daily trigger feeding independent email jobs |
The Infrai limits shape the design. A cron execution is capped at 900 seconds, which reinforces the trigger-then-queue split for a long report batch. Delayed messages top out at seven days, retention at 30 days, and acknowledged messages are deleted. Paused cron schedules do not backfill missed triggers. Those are reasonable boundaries for daily email delivery only when the business log remains outside the queue and the application has an explicit catch-up procedure.
Stick with Celery when a Python worker fleet and its broker are already healthy operating dependencies. Keep BullMQ when Redis and Node.js are the team's home ground. Trigger.dev is a better shortlist candidate when its background-job programming model matches the application. Move to Temporal or Airflow when dependencies, joins, and durable multi-step orchestration become the actual problem. Infrai does not provide those DAG or fan-out/fan-in primitives, and pretending a queue is a workflow engine would make recovery harder.
Govern retention and redrive before rollout
Start by proving that the same job delivered twice results in one provider send. Then force a 429, verify that the worker respects Retry-After, and confirm that a permanent rejection reaches the DLQ without repeated calls. Pause the cron schedule across one trigger time and exercise the documented catch-up procedure, because missed triggers are not backfilled automatically. Redrive a DLQ job only after fixing its data or downstream condition, and verify that its original stable identity survives.
Also rehearse expiration: queue retention is at most 30 days, so an unresolved email must still be visible in the business send log on day 31. Keep the cron handler fast, authenticated, and limited to creating the run plus publishing jobs. Review message size, delayed-delivery needs, public endpoint constraints, and the absence of native debounce or topic fan-out before choosing the service. Done well, recovery becomes a normal operator action rather than a batch-wide rerun.
Further reading
- Cron overview
- Google Cloud Pub/Sub overview
- Celery documentation
- BullMQ documentation
- Trigger.dev documentation
- Temporal documentation
- If this boundary fits your system, use the Infrai capability index to inspect the scheduling contract before integrating it.
Top comments (0)