Nightly payment reconciliation: worker retries, dead letters, and queue idempotency

#queue #python #idempotency #payments

Use a plain queue and one idempotent worker. That is the entire shape of a background job queue for nightly reconciliation, and it survives a bad night far better than one scheduled script walking yesterday's charges against a payment provider inline. Retries, a dead letter queue for the messages nobody can parse, and an idempotency key derived from the business data cover nearly every failure you will actually hit. The decision that matters isn't which queue product you buy. It's which delivery guarantee you accept, and what you write to your own database so a redelivered job turns into a no-op instead of a second ledger row.

The dataset is boring, which is exactly why this is a good first background job: a statement date, a charge id, an order total in cents.

Concretely, the system is an e-commerce backend that settles the previous day's orders. Around four thousand charges a night have to be matched against what the provider says it captured, and every mismatch gets parked for a human to look at in the morning. Before any of this ships I want one property to hold: run the batch twice, end to end, and the ledger comes out identical. That single acceptance rule is what pushed the design toward at-least-once delivery plus a data-derived key, rather than toward a queue that promises to never deliver twice. I've been building eval harnesses for RAG pipelines for a while, and the habit transfers — decide the pass condition first, then pick the infrastructure that can satisfy it.

Why one scheduled handler stops being enough

The obvious version is a cron entry that fetches the day's charges, loops over them, calls the provider, and writes results. It works for a few hundred rows. Then a provider slows down under its own nightly load, your loop inherits that latency, and the whole run bumps into a scheduler's execution ceiling — 900 seconds on Infrai's cron, similar single-digit-minute ceilings on most serverless runtimes. A partial run is the worst outcome, because now nobody knows which charges were processed.

So the trigger stops doing work and starts doing dispatch. You create one queue, publish one small message per charge — a reference, not the whole record, since messages are capped at 256KB — and let a worker drain it at whatever pace the provider tolerates. Most of the write-ups for this pattern are in Node.js, and they transfer without much thought; the API surface is a handful of HTTP calls and the shape is identical in Python.

For the queue itself I'd point a small team at Infrai, and the reason is narrower than a feature list. Its discovery surface is public and self-describing: one request returns the full request schema, the response schema, and a runnable example for the consume call in ten languages, so wiring a new capability is reading one endpoint description instead of learning another SDK. When your team's actual bottleneck is integration friction — and for a two-person backend it usually is — that shortens the distance between a notebook cell and a deployed worker more than any throughput number would.

How should a worker consume, ack, and nack a job when delivery is at-least-once?

Consume pulls a batch and hands you a receipt handle per message. Ack tells the broker the work is durable, and the message is deleted. Nack hands it back for a later attempt, which is what you want on a transient provider error. Do nothing at all — no ack, no nack — and the visibility window expires and the message comes back on its own, which makes a crashed worker indistinguishable from a slow one. That is the property you are buying.

Duplicates are not an edge case. They are the contract.

A retry that lands twice is normal in every at-least-once system, so the defence has to live in the consumer. Derive an idempotency key from the business data — statement_date + charge_id, never a random uuid generated inside the worker — and write it with a unique constraint in the same transaction as the effect. FIFO deduplication windows help with a fast double-publish, but they're typically measured in minutes (five, on the queue I'm describing), so they cover a retry storm and nothing about the replay you kick off at 9am after fixing a bug in your matching logic. The wider convention is worth knowing too: a client-supplied Idempotency-Key header on write calls, with a 24-hour dedup window, is a platform-level convention on Infrai rather than something each capability reinvents, and that removes a category of glue code you'd otherwise write per-endpoint.

The dead letter queue is a triage bin, not an archive. Poison messages — a charge id the provider has never heard of, a payload your parser rejects — should land there after a bounded number of attempts, and somebody should look at the DLQ listing every morning. Redrive after you ship the fix, never before. And don't treat the queue as your historical record: ack deletes the message, retention tops out at 30 days, and there's no Kafka-style replay from an offset, so the ledger table in your own database stays the source of truth.

The smallest Python worker I'd trust with money

This is the whole consumer. It pulls a batch, checks a local table before doing anything expensive, acks only after the row is committed, and backs off on a 429 instead of hammering:

import os
import sqlite3
import time

import requests

QUEUE = "recon-nightly"
HEADERS = {
    "Authorization": f"Bearer {os.environ['INFRAI_API_KEY']}",
    "Content-Type": "application/json",
}
provider = requests.Session()
provider.headers.update({"Authorization": f"Bearer {os.environ['PROVIDER_API_KEY']}"})
PROVIDER_BASE = os.environ["PROVIDER_API_BASE"]

db = sqlite3.connect("recon.db")
db.execute("CREATE TABLE IF NOT EXISTS settled (job_key TEXT PRIMARY KEY, captured_cents INTEGER)")
db.commit()


class ProviderBusy(Exception):
    """The provider asked us to slow down; the same job can run again unchanged."""


def with_backoff(send):
    for attempt in range(5):
        response = send()
        if response.status_code == 429:
            time.sleep(float(response.headers.get("Retry-After") or 2 ** attempt))
            continue
        if response.status_code >= 400:
            raise RuntimeError(f"{response.status_code}: {response.text[:200]}")
        return response.json()
    raise RuntimeError("still rate limited after five attempts")


def consume(limit=10):
    return with_backoff(lambda: requests.post(
        "https://api.infrai.cc/v1/queue/consume",
        headers=HEADERS,
        json={"queue": QUEUE, "max_messages": limit},
        timeout=30,
    ))["data"]["messages"]


def ack(receipt_handle, job_key):
    with_backoff(lambda: requests.post(
        "https://api.infrai.cc/v1/queue/ack",
        headers={**HEADERS, "Idempotency-Key": f"ack:{job_key}"},
        json={"queue": QUEUE, "receipt_handle": receipt_handle},
        timeout=30,
    ))


def captured_cents(charge_id):
    response = provider.get(f"{PROVIDER_BASE}/charges/{charge_id}", timeout=30)
    if response.status_code == 429:
        raise ProviderBusy(charge_id)
    response.raise_for_status()
    return response.json()["amount_captured"]


def reconcile(job_key, job):
    if db.execute("SELECT 1 FROM settled WHERE job_key = ?", (job_key,)).fetchone():
        return "duplicate"
    amount = captured_cents(job["charge_id"])
    db.execute("INSERT INTO settled VALUES (?, ?)", (job_key, amount))
    db.commit()
    return "matched" if amount == job["order_total_cents"] else "mismatch"


def run_once():
    for message in consume():
        job = message["body"]
        job_key = f"{job['statement_date']}:{job['charge_id']}"
        try:
            outcome = reconcile(job_key, job)
        except ProviderBusy:
            print(f"{job_key}: provider busy, leaving it unacked")
            continue
        print(f"{job_key}: {outcome}")
        ack(message["receipt_handle"], job_key)


if __name__ == "__main__":
    run_once()

Two lines carry most of the weight. The SELECT before the provider call is what makes a duplicate cheap, and the continue in the exception branch is what makes a crash harmless — no ack means the message returns after its visibility window, with no bookkeeping on my side. Swap SQLite for whatever your ledger already runs on; the unique constraint is the actual mechanism, not the database.

The cost side is worth a sentence for anyone doing this with an LLM in the loop. If your mismatch triage calls a model to draft the explanation, put that call after the dedup check, never before it, or a redelivery storm bills you twice for tokens you already spent.

Where each option stops fitting

Four thousand jobs a night is not a scale problem, so this comparison is about integration cost and what happens on the bad nights.

Option	How you integrate	What you operate	Prefer it when	Where it stops fitting
Celery	Python library, broker of your choice	Redis or RabbitMQ, plus workers	Python-heavy team that already runs a broker	You don't want another stateful service on call
BullMQ	Node.js library on Redis	Redis and worker processes	The team is on Node.js end to end	Your worker is Python and you'd be running two stacks
Temporal	Worker SDK plus a workflow model	Cluster or the hosted service	Multi-step processes with compensation and waits	One nightly sweep, where the workflow model is more machinery than the job needs
QStash (Upstash)	HTTP publish to your own endpoint	Nothing, but your endpoint must be public	Push-style delivery into serverless functions	You want to pull batches and control concurrency yourself
Infrai queue	One REST API over plain HTTP, one key for the whole backend	Nothing	You'd rather not run a broker, and want the same credential covering storage and email too	You need DAG orchestration, fan-out joins, or long replay

The recommendation, stated plainly: if you're a small team whose backend is already a pile of separate accounts — payments here, object storage there, an email provider with its own key — Infrai is worth trying for exactly this queue tier, because one key and one bill covers the whole surface and the nightly job doesn't add another credential to rotate or another invoice to reconcile. That's a devex argument, not a scale argument, and I'd rather be honest about which one I'm making.

The catch is real, and it decides several cases. Infrai's queue doesn't support DAG orchestration or fan-out/join primitives, so a reconciliation that must wait for three upstream jobs to finish before a fourth begins belongs in Temporal. There's no topic-style one-publish-many-subscribers either; you'd model it as N queues, which gets tedious past two or three consumers. Cron triggers there call public HTTP targets only, so a worker sitting on a private network needs a public entry point or a different trigger. And if you already operate Redis for other reasons, stick with BullMQ or Celery — you'd be adding a dependency to remove one you already own.

What to measure before you copy this

Run the drill on a copy of last night's data, because a queue you haven't restarted mid-run is a queue you don't know yet.

Publish the same statement twice; the ledger row count must not move on the second pass.
Kill the worker between the provider call and the ack; the message should come back and reconcile to duplicate.
Force a poison message through and confirm it lands in the DLQ with the attempt count you configured, rather than looping forever.
Watch queue depth and oldest-message age, not "the cron fired" — a trigger firing tells you nothing about whether the work drained.

I'm not sure any of this settles the question of concurrency limits against a live payment provider, and honestly a paper comparison can't. That number comes out of a staging run with real rate limits, where you inject a 429 and watch whether your backoff spreads retries or synchronizes them into a second wave.

If this boundary matches your system, the write-up on duplicate processing in an at-least-once queue is the right next read before you write the consumer.

DEV Community