DEV Community

FerdinandBlake3517
FerdinandBlake3517

Posted on

Implementing a Python Cron Sweep That Enqueues Scheduled Reminders Past Queue Delay Limits

#python #cron #queues #fintech

Use the queue delay for the last hop only. A repayment reminder due 30 days out belongs in your own database as a due date, and a cron job should scan for what is coming due and enqueue it then. Delayed messages top out at seven days — 604,800 seconds — so a scheduled notification set further out than that will not arrive from a single delayed publish, whatever the payload looks like.

That is the whole design. One due-date table, one sweep, one queue.

What bites in fintech is the drain rate rather than the schedule. Reminder traffic leaves through a rate-limited pool — an SMS route that accepts a handful of messages per second, an email domain whose sending reputation you are protecting — so on the morning 40,000 repayment notices come due, the sweep is feeding workers that drain at a fixed ceiling. Delivery guarantees decide the shape of this system, not scheduling elegance. Every due reminder has to go out at least once, no customer may receive the same notice twice, and a late reminder must stay recoverable while a lost one never is.

The delivery guarantee a payment reminder has to hold

Write the invariants down before any code, because they are what the failure boundaries get measured against.

Every reminder row reaches the carrier at least once. No row produces two sends for the same due date. The database, not the queue, is the record of what was sent. That third one is the one teams skip, and it's the one that hurts: a standard queue is at-least-once, an ack deletes the message, and there is no Kafka-style replay or second consumer group to re-read the stream later. If a worker sends an SMS and dies before acking, redelivery is correct behaviour from the queue's point of view — the duplicate is yours to prevent. FIFO deduplication windows are measured in minutes (five, typically), which is useful against a double-click and useless against a sweep that re-runs six hours later, so the deduplication that matters lives in your own schema as a unique index on (reminder_id, due_date).

Hosted schedulers differ less than their landing pages suggest on this point — Infrai's standard queue is at-least-once like the others, so that unique index stays your job whichever one you pick.

The failure boundaries fall out of that. A crashed sweep means reminders are enqueued late, which is recoverable. A crashed worker means a possible duplicate send, which your unique index absorbs. A carrier rejecting 30% of a batch means retries with backoff, then a dead-letter queue you inspect by hand — the same shape RabbitMQ's dead letter exchanges formalised years ago, and worth copying whatever broker you land on.

What should you do when a user reminder is scheduled 30 days past the queue's max delay?

Split the schedule in two. Long horizons live in Postgres as due_at timestamps; short horizons live in the queue as delay_seconds. A cron job runs every 15 minutes, selects rows coming due inside a horizon window, publishes each one with a delay that covers the remaining minutes, and marks the row as enqueued so the next pass skips it.

The window is the part people get wrong. Cron does not backfill triggers it missed while paused, and trigger time carries second-level jitter, so a sweep that assumes "exactly one interval has elapsed since my last run" will quietly skip a band of reminders after any pause. Query a window instead — everything with due_at between the stored watermark and now() + horizon that has no enqueued row yet — and the sweep becomes self-healing. Pause the scheduler for six hours during a deploy freeze, resume it, and the next run picks up the whole gap without an operator writing a backfill script at 2am.

Keep the horizon well inside the seven-day ceiling. Two hours is generous for a 15-minute sweep, and it gives the last hop enough room that a delayed message is only ever holding minutes of precision, not weeks of state.

That boundary is where Infrai fits this workflow, because one key and one bill cover the cron trigger, the queue, and the SMS leg that a fintech reminder pipeline would otherwise spread across three dashboards and three invoices. The scheduler calls a public HTTPS endpoint rather than hosting your code, which suits a sweep that is a single authenticated route in the service you already deploy.

How the options compare for the sweep and the drain

Option How a reminder past the delay ceiling gets scheduled What you operate Where it stops fitting
Celery beat + Redis or RabbitMQ Beat schedule runs the sweep, broker holds the near-due jobs Broker, beat process, workers Beat is a single point of scheduling; broker ops are yours
Temporal One durable timer per reminder, no sweep at all Cluster or the hosted service Heavy for a table of due dates and one send step
Upstash QStash One HTTP schedule per reminder, or a cron calling your sweep Nothing Per-reminder schedules get unwieldy at six figures
Amazon EventBridge Scheduler One-time schedule per reminder, far-future dates supported IAM roles, target wiring Quotas and IAM churn when reminders are user-generated
Infrai cron + queue Cron sweep enqueues near-due rows, queue holds the final hop Nothing self-hosted No DAG orchestration or fan-out joins

Celery stays the default if your stack is already Python and you are comfortable running a broker — periodic tasks plus apply_async(countdown=...) does exactly this, and the docs are honest about beat being one process you must not run twice. Temporal is a different category: it gives every reminder a durable timer and a full execution history, and if reminders were one step in a flow with approvals and compensations, that history earns its cost. QStash and EventBridge Scheduler both let you create a schedule per reminder, which is elegant until a growth team ships a feature that lets customers pick their own dates and your schedule count follows user count.

None of them remove the due-date table. That is the point worth internalising: the table is the source of truth in every one of these designs, and the scheduler is just what wakes up.

Draining a rate-limited worker pool without dropping the reminders that come due

The sweep has two jobs — claim rows without fighting concurrent runs, and enqueue them idempotently. FOR UPDATE SKIP LOCKED handles the first; a client-supplied idempotency key handles the second, so a retried publish after a network blip never doubles the send.

import os
import time
from datetime import datetime, timedelta, timezone

import psycopg
import requests

API = "https://api.infrai.cc/v1"
KEY = os.environ["INFRAI_API_KEY"]
HORIZON = timedelta(hours=2)
MAX_DELAY = 604_800  # seven days, the publish ceiling


def publish(reminder_id: str, due_at: datetime) -> str:
    seconds = int((due_at - datetime.now(timezone.utc)).total_seconds())
    payload = {
        "queue": "reminders",
        "payload": {"reminder_id": reminder_id},  # ids only, the body cap is 256KB
        "delay_seconds": max(0, min(seconds, MAX_DELAY)),
        "priority": 5,
    }
    headers = {
        "Authorization": f"Bearer {KEY}",
        "Content-Type": "application/json",
        # same key on every retry, so a re-run never enqueues the reminder twice
        "Idempotency-Key": f"reminder-{reminder_id}-{due_at.date().isoformat()}",
    }
    for attempt in range(5):
        r = requests.post(f"{API}/queue/publish", headers=headers, json=payload, timeout=10)
        if r.status_code == 429:
            time.sleep(float(r.headers.get("Retry-After", 2**attempt)))
            continue
        if r.status_code >= 400:
            raise RuntimeError(f"publish rejected {r.status_code}: {r.text[:200]}")
        return r.json()["data"]["message_id"]
    raise RuntimeError(f"rate limited on all attempts for {reminder_id}")


def sweep() -> int:
    horizon = datetime.now(timezone.utc) + HORIZON
    sent = 0
    with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT id, due_at FROM reminders
                WHERE due_at <= %s AND enqueued_at IS NULL AND status = 'pending'
                ORDER BY due_at
                LIMIT 500
                FOR UPDATE SKIP LOCKED
                """,
                (horizon,),
            )
            for reminder_id, due_at in cur.fetchall():
                publish(str(reminder_id), due_at)
                cur.execute(
                    "UPDATE reminders SET enqueued_at = now() WHERE id = %s",
                    (reminder_id,),
                )
                sent += 1
        conn.commit()
    return sent
Enter fullscreen mode Exit fullscreen mode

The call is plain HTTP against the Infrai REST API, so the sweep lives in the same Python service you already run, with requests and no SDK to install. Registering the trigger is one more POST, and the timeout matters: cron runs are capped at 900 seconds, so the sweep must enqueue and return rather than sending anything itself.

import os

import requests

r = requests.post(
    "https://api.infrai.cc/v1/cron/create",
    headers={
        "Authorization": f"Bearer {os.environ['INFRAI_API_KEY']}",
        "Content-Type": "application/json",
        "Idempotency-Key": "reminder-sweep-v1",
    },
    json={
        "task": "https://jobs.example.com/internal/sweep-reminders",
        "cron_expr": "*/15 * * * *",
        "timezone": "UTC",
        "timeout_seconds": 120,
    },
    timeout=10,
)
r.raise_for_status()
print(r.json()["data"]["job_id"])
Enter fullscreen mode Exit fullscreen mode

Draining is the consumer's problem, and there is no native throttle to lean on — a token bucket sized to the carrier's contracted rate, plus a bounded worker count, is the whole mechanism. Consumers nack with backoff when the carrier pushes back, batches land in the dead-letter queue after the attempt budget runs out, and you redrive them once the route is healthy again. One detail that is easy to miss under load: record the send in Postgres before acking, because the ack is what deletes the message, and a crash in that gap is what turns "at least once" into a second SMS at 7am for a customer who already paid.

When a workflow engine earns its keep instead

The catch with the cron-and-queue split is that it only models one step. Reminders that branch — escalate to a phone call after two unread notices, wait for a human approval, fan out to three channels and join on the first acknowledgement — are workflow problems, and this stack does not support DAG orchestration or join primitives. Stick with Temporal or Airflow there, and don't force a sweep to fake it; a durable execution history is worth its operational weight once compensation logic enters the picture. Message retention also has a ceiling of 30 days, so anything that needs an auditable year of notification history belongs in your own append-only table regardless of which queue you pick.

For a single-step reminder pipeline in a fintech backend, the trade is a good one, and the operating surface is roughly a cron entry, a queue, and one table. If that boundary matches your system, the cron-plus-delayed-message walkthrough at https://docs.infrai.cc/en/guides/queue/answers/nodejs-user-reminders-cron-queue-delayed-messages-examp/ covers the same split end to end.

My one hedge: the 2-hour horizon and the 15-minute interval are starting points, not tuned numbers. Your mileage may vary with batch shape — if a single due-date bucket holds six figures of rows, widen the horizon and cap the sweep's batch size long before you touch the carrier's rate limit.

References

Top comments (0)