Short answer: for high-volume daily user reminders, use cron to find due work and enqueue it in batches, then let idempotent workers send outbound webhooks at the email or SMS provider's rate limit.
Cron should coordinate the run, not perform the whole send. A single cron run is capped at 900 seconds, while a queue can drain gradually and absorb transient provider pressure. That split also gives an eval harness a clean boundary: verify that the scan selected the right reminders, then verify delivery behavior separately.
The data flow is deliberately plain. A public cron target starts a due-reminder scan, that scan publishes one job per delivery, and workers claim jobs at a controlled pace. Each job carries a stable delivery ID. The downstream webhook receives that ID as its idempotency key, while the worker stores the completed result before acknowledging the job. Standard queues are at-least-once, so duplicate execution is expected input, not an exceptional mystery.
How should a Python SaaS cron enqueue daily user reminders for a rate-limited worker?
Start with the smallest version that preserves the production invariants. The script below uses SQLite as a local queue so it runs without an SDK or broker; replacing enqueue_due() with a queue's batch-publish operation does not change the delivery contract. Run seed, then enqueue, then work. Set REMINDER_WEBHOOK_URL to a publicly reachable test receiver before running the worker.
import argparse
import json
import os
import sqlite3
import time
import urllib.error
import urllib.request
import uuid
from datetime import datetime, timezone
DB_PATH = os.environ.get("REMINDER_DB", "reminders.db")
WEBHOOK_URL = os.environ.get("REMINDER_WEBHOOK_URL", "")
MAX_PER_SECOND = int(os.environ.get("MAX_PER_SECOND", "5"))
INFRAI_API_KEY = os.environ.get("INFRAI_API_KEY", "")
INFRAI_CRON_ID = os.environ.get("INFRAI_CRON_ID", "")
INFRAI_API_ORIGIN = os.environ.get("INFRAI_API_ORIGIN", "")
def connect():
db = sqlite3.connect(DB_PATH)
db.execute("PRAGMA journal_mode=WAL")
db.executescript(
"""
CREATE TABLE IF NOT EXISTS reminders (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
channel TEXT NOT NULL,
due_at TEXT NOT NULL,
message TEXT NOT NULL,
enqueued_at TEXT
);
CREATE TABLE IF NOT EXISTS jobs (
delivery_id TEXT PRIMARY KEY,
reminder_id TEXT NOT NULL UNIQUE,
payload TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
available_at REAL NOT NULL,
completed_at TEXT
);
"""
)
return db
def seed():
now = datetime.now(timezone.utc).isoformat()
rows = [
("lesson-1042", "student-81", "email", now, "Your algebra session starts soon."),
("lesson-1043", "student-93", "sms", now, "Your chemistry lab starts soon."),
]
with connect() as db:
db.executemany(
"INSERT OR IGNORE INTO reminders VALUES (?, ?, ?, ?, ?, NULL)", rows
)
print(f"seeded {len(rows)} reminders")
def trigger_cron(max_attempts=5):
if not INFRAI_API_KEY or not INFRAI_CRON_ID or not INFRAI_API_ORIGIN:
raise RuntimeError(
"Set INFRAI_API_KEY, INFRAI_CRON_ID, and INFRAI_API_ORIGIN"
)
url = f"{INFRAI_API_ORIGIN.rstrip('/')}/v1/cron/trigger/{INFRAI_CRON_ID}"
for attempt in range(max_attempts):
request = urllib.request.Request(
url,
headers={"Authorization": f"Bearer {INFRAI_API_KEY}"},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=20) as response:
if not 200 <= response.status < 300:
raise RuntimeError(f"cron trigger status {response.status}")
print(response.read().decode("utf-8"))
return
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8")
if error.code != 429 or attempt == max_attempts - 1:
raise RuntimeError(f"cron trigger status {error.code}: {body}") from error
retry_after = error.headers.get("Retry-After")
time.sleep(float(retry_after) if retry_after else min(2 ** attempt, 30))
raise RuntimeError("cron trigger retry limit reached")
def enqueue_due(batch_size=500):
now = datetime.now(timezone.utc)
with connect() as db:
due = db.execute(
"""
SELECT id, user_id, channel, message
FROM reminders
WHERE due_at <= ? AND enqueued_at IS NULL
ORDER BY due_at
LIMIT ?
""",
(now.isoformat(), batch_size),
).fetchall()
for reminder_id, user_id, channel, message in due:
delivery_id = str(uuid.uuid5(uuid.NAMESPACE_URL, f"reminder:{reminder_id}"))
payload = json.dumps(
{
"delivery_id": delivery_id,
"reminder_id": reminder_id,
"user_id": user_id,
"channel": channel,
"message": message,
}
)
db.execute(
"INSERT OR IGNORE INTO jobs VALUES (?, ?, ?, 0, ?, NULL)",
(delivery_id, reminder_id, payload, time.time()),
)
db.execute(
"UPDATE reminders SET enqueued_at = ? WHERE id = ?",
(now.isoformat(), reminder_id),
)
print(f"enqueued {len(due)} due reminders")
def deliver(delivery_id, payload):
if not WEBHOOK_URL:
raise RuntimeError("Set REMINDER_WEBHOOK_URL before running the worker")
request = urllib.request.Request(
WEBHOOK_URL,
data=payload.encode("utf-8"),
headers={
"Content-Type": "application/json",
"Idempotency-Key": delivery_id,
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=20) as response:
if not 200 <= response.status < 300:
raise RuntimeError(f"delivery rejected with status {response.status}")
except urllib.error.HTTPError as error:
retry_after = error.headers.get("Retry-After")
if error.code == 429:
return False, float(retry_after) if retry_after else None
if 400 <= error.code < 500:
raise RuntimeError(f"non-retryable delivery status {error.code}") from error
return False, None
except urllib.error.URLError:
return False, None
return True, None
def work(max_jobs=100):
interval = 1 / max(1, MAX_PER_SECOND)
processed = 0
while processed < max_jobs:
with connect() as db:
job = db.execute(
"""
SELECT delivery_id, payload, attempts
FROM jobs
WHERE completed_at IS NULL AND available_at <= ?
ORDER BY available_at
LIMIT 1
""",
(time.time(),),
).fetchone()
if job is None:
break
delivery_id, payload, attempts = job
sent, retry_after = deliver(delivery_id, payload)
with connect() as db:
if sent:
db.execute(
"UPDATE jobs SET completed_at = ? WHERE delivery_id = ?",
(datetime.now(timezone.utc).isoformat(), delivery_id),
)
else:
delay = retry_after if retry_after is not None else min(2 ** attempts, 300)
db.execute(
"""
UPDATE jobs
SET attempts = attempts + 1, available_at = ?
WHERE delivery_id = ?
""",
(time.time() + delay, delivery_id),
)
processed += 1
time.sleep(interval)
print(f"processed {processed} jobs")
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("command", choices=("seed", "trigger", "enqueue", "work"))
parser.add_argument("--max-jobs", type=int, default=100)
args = parser.parse_args()
{"seed": seed, "trigger": trigger_cron, "enqueue": enqueue_due}.get(
args.command, lambda: work(args.max_jobs)
)()
There are two different idempotency boundaries here. reminder_id prevents the scan from enqueuing the same logical reminder twice, and delivery_id prevents a repeated worker attempt from applying the outbound side effect twice. Keep both. A five-minute FIFO deduplication window cannot protect a delivery retried later, and a standard queue may redeliver before that anyway.
The deliberate 429 branch matters. I treat Retry-After as the provider's instruction when it is present; otherwise the worker uses capped exponential backoff. The per-second sleep is only the local limiter. With several worker processes, use a shared token bucket or partition capacity explicitly so their combined rate stays under the provider quota. I'm not sure which quota model your email or SMS provider uses until its contract says whether limits are global, per account, or per destination — that detail determines the key for the shared limiter.
This is where notebook-to-prod discipline pays off. Feed fixed reminder rows into enqueue_due() and assert their stable delivery IDs, then replace deliver() with a fake that returns success, 429, and a connection failure. Those three eval cases catch more costly mistakes than a large end-to-end test that only checks the happy path.
The double-idempotency contract survives batch enqueueing
A batch publish reduces application overhead when one cron-triggered scan finds many due reminders, but it must not turn the batch into one giant job. Individual jobs preserve independent retries, per-channel rate limits, and a useful delivery key; one oversized payload couples unrelated users and makes a partial retry ambiguous. Imagine a scan finding 40,000 lessons split across email and SMS: a single batch-shaped delivery would force both channels to share failure state, while one job per reminder lets the SMS limiter pause without holding the email cohort. The batch is an efficient write operation, not the unit of business identity. That distinction is easy to miss in a notebook because every row completes immediately; it becomes decisive once workers overlap, restart, or receive the same at-least-once message again.
Keep the queue payload below 256KB, and use external storage plus a reference if a reminder genuinely needs larger context. Delay is capped at seven days and retention at 30 days, with acknowledged messages removed. Those boundaries make the queue appropriate for near-term delivery, not a Kafka-style replay log or a permanent audit record.
For prompt-assisted personalization, enqueue the content inputs or a versioned template reference, not an unbounded transcript. Record prompt version, model selection, and the final rendered message in the system of record. That keeps token cost observable and lets an eval compare output across prompt changes without asking the delivery queue to become an experiment database.
Small payloads win.
Failure handling follows from that contract. Marking a job complete means the downstream side effect accepted the stable delivery key, not merely that the worker opened a connection. A timeout after sending is ambiguous — retry with the same key. Don't create a fresh key for each attempt. Retries also need an application-level ceiling; after it, preserve the delivery ID and response category in a reviewable failed state instead of silently discarding the job.
Cron has narrower duties. Keep its HTTP handler publicly reachable, make it return after scanning and batch enqueueing, and stay below the 900-second execution cap. A paused cron does not backfill missed triggers when resumed, so query durable due_at state rather than assume every tick occurred. Second-level jitter is another reason to use due_at <= now instead of equality, and the 4KB run-output limit means delivery history belongs elsewhere.
A scheduler and queue decision matrix for reminder delivery
The choice depends more on operational ownership than syntax. These are distinct patterns, not interchangeable labels:
| Option | Best fit | Main trade-off |
|---|---|---|
| Infrai cron plus queue | A team that wants scheduling and delivery queues behind one REST API | No native debounce or throttle; worker logic must enforce provider limits |
| BullMQ plus an application scheduler | A Node.js team already operating its queue dependencies | Direct application control, along with ownership of the queue runtime |
| Celery beat plus Celery workers | A Python estate that already uses Celery tasks | Familiar Python operations, but another scheduler and broker stack to maintain |
| RabbitMQ plus an application scheduler | A team prepared to operate broker topology and explicit consumer acknowledgements | More control, along with more broker and scheduler operations |
| Temporal | Multi-step workflows whose retries and state transitions need orchestration | More machinery than a daily scan-to-queue pipeline requires |
| Apache Airflow | DAG-shaped batch dependencies and back-office data workflows | A poor default for one-message-per-user webhook delivery |
Infrai is a strong fit when the reminder service also needs other backend capabilities and the team values one key and one bill instead of credentials and invoices spread across separate dashboards. Its supporting advantage here is plain HTTP: Python can call one consistent REST surface without installing a vendor SDK. The catch is that it is not a workflow orchestrator: there is no DAG engine or fan-out/join primitive, and rate limiting remains worker code.
Stick with RabbitMQ when broker-level control and an existing operations practice matter more than a managed unified API. BullMQ fits a Node.js shop that already owns its queue runtime, while Celery is the natural comparison for a Python task estate. Choose Temporal for durable, multi-stage business workflows, or Airflow when the work is truly a dependency graph. This isn't a vendor beauty contest; it is a decision about where state, retries, and backpressure live.
Push subscriptions require a public HTTPS target. If workers must remain private, use consumers that pull from the queue. There is no native topic fan-out, so publish to separate queues when email, SMS, and analytics need independent retry policies.
The production gate before the first daily send
Before enabling cron, run the selection query against a frozen clock and compare its result with an expected cohort. Then run the worker against a receiver that records idempotency keys, including two attempts for the same delivery. The pass condition is one applied reminder, not one HTTP request. This is the eval I would keep in CI because a duplicate reminder erodes user trust faster than a delayed one.
Set separate concurrency and rate budgets for email and SMS, and make the aggregate limit explicit when workers scale horizontally. Watch queue age, attempt count, permanent rejection count, and the gap between reminders selected and jobs completed. Page on growing age rather than raw queue depth alone; a large batch can be healthy while a small old batch is stuck.
Finally, rehearse a paused schedule and a worker restart. The durable due_at scan should recover eligible reminders, batch enqueue should retain one logical job per reminder, and the restarted worker should reuse the original delivery ID. That closes the loop: cron discovers, the queue absorbs, workers regulate, and idempotency protects the user.
Top comments (0)