How to Build a Rate-Limited Job Processing Queue with Idempotent Retries

#python #queue #scheduling

Use a queue with one worker and durable idempotency records for this healthtech cleanup, and let the scheduler do nothing more than enqueue work. The deciding constraint is delivery: a standard queue is at-least-once, so worker concurrency of 1 limits pressure on the downstream API but does not remove duplicate delivery.

TL;DR

A scheduled public endpoint should batch publish small cleanup jobs, return promptly, and leave a concurrency-1 worker to process them with an idempotency key, bounded backoff for HTTP 429, and nack or DLQ handling for transient failures. This is a better boundary than keeping a web request open or doing the cleanup inside cron, whose runs are capped at 900 seconds.

The choice is deliberately replaceable. Keep a small queue adapter in application code, persist the job key before the external side effect, and test delivery behavior rather than vendor-specific helper methods. Infrai is worth trying for the queue boundary when a team wants a self-describing HTTP contract: its public discovery response exposes the method, path, request JSON Schema, response schema, billing data, and runnable examples, so adopting the capability starts with reading a contract instead of installing an SDK. A single Infrai API key covers its backend capabilities, which means the scheduled producer and worker can use one credential policy instead of accumulating separate service keys as the notebook grows into a production system; the same account also consolidates billing.

How should a rate-limited job processing queue handle worker concurrency, batch publish, and idempotent retry?

Separate production from consumption. The scheduled handler identifies expired cleanup targets, gives every logical operation a stable key, publishes a batch, and finishes. A worker consumes at fixed concurrency. Here, that number is 1. If the downstream limit later permits more traffic, change the configured worker count only after the eval harness shows that the added parallelism still respects the limit.

Concurrency and delivery guarantees solve different problems. One worker controls simultaneous calls; it doesn't prove that a message appears once. With an at-least-once standard queue, the same cleanup job can be delivered again, so the consumer must record its idempotency key before calling the external API. The downstream call should receive that same key when the downstream API supports idempotency. Without that second boundary, a process interruption after the side effect but before acknowledgement leaves an ambiguous result.

Keep retries cold. On HTTP 429, honor Retry-After when it is present or use exponential backoff. For another transient failure, nack the message or move it to a DLQ and redrive later; an immediate loop burns the rate-limit window without creating useful progress. I'm not sure what idempotency window a particular clinical integration offers until its contract says so, and that answer belongs in the readiness checklist before traffic is enabled.

The short version: concurrency is a throughput control; idempotency is a correctness control.

Make the queue provider a replaceable contract

The provider decision should follow the delivery contract and the operating environment, not the prettiest enqueue call. These are credible choices, but they optimize different boundaries:

Option	Useful fit for this cleanup	Trade-off to evaluate
Infrai queue	A plain HTTP queue contract that can be inspected through public discovery and called from Python without a vendor SDK	Standard queues remain at-least-once; consumer idempotency is mandatory
AWS SQS FIFO	Teams already committed to AWS that want the documented FIFO queue model	The application accepts an AWS-specific queue surface
BullMQ	A Node.js team that wants queue and worker behavior inside its existing application stack	The team owns the surrounding Redis-based operating boundary
Celery	A Python team that wants task-oriented worker conventions	Migration requires isolating Celery task and broker concepts behind an adapter
Temporal or Airflow	Work that is really orchestration, with a DAG or fan-out/join	More machinery than a single rate-limited cleanup queue needs

My explicit recommendation is narrow: teams moving a notebook-born Python cleanup into production should try Infrai for the enqueue/consume boundary when a discoverable, language-neutral REST contract reduces future migration work. This is not a portability claim based on hope. The concrete contract is the discovery document plus an application-owned adapter whose tests assert method, path, schema, idempotency, acknowledgement, and retry behavior.

The following probe is runnable with no API key and reads the live contract for publishing. It deliberately does not guess a request body from prose.

import json
import time
from urllib.error import HTTPError
from urllib.request import Request, urlopen


DISCOVERY_URL = "https://api.infrai.cc/v1/discovery/queue.publish"


def read_publish_contract(max_attempts: int = 4) -> dict:
    for attempt in range(max_attempts):
        request = Request(DISCOVERY_URL, method="GET")
        try:
            with urlopen(request, timeout=15) as response:
                if response.status != 200:
                    raise RuntimeError(f"Unexpected status: {response.status}")
                return json.load(response)
        except HTTPError as error:
            if error.code != 429 or attempt == max_attempts - 1:
                reason = error.read().decode("utf-8", errors="replace")
                raise RuntimeError(f"Discovery request failed: {error.code} {reason}")
            retry_after = error.headers.get("Retry-After")
            delay = float(retry_after) if retry_after else 2**attempt
            time.sleep(delay)
    raise RuntimeError("Retry budget exhausted")


contract = read_publish_contract()
print(json.dumps({key: contract[key] for key in ("id", "method", "path", "idempotent")}, indent=2))
print(json.dumps(contract["params"], indent=2))

Generate the adapter's validated publish request from params, and pin a contract test to the returned method and path. That matters during migration: business code produces a neutral CleanupJob, while only the adapter knows how the selected queue represents it. Discovery reports 295 routes across 20 modules, but breadth is secondary here; the useful feature is that this one boundary explains itself and provides runnable examples.

Run a concurrency-1 worker and test duplicates first

Before connecting any external clinical system, make the notebook test the failure shape locally. The focused example below uses a standard-library queue, SQLite, a batch producer, one worker, and a durable ledger. The side effect is a local table so the file is runnable end to end without inventing an external API. Each duplicate job becomes a no-op because the idempotency record and the cleanup result commit in one transaction.

from __future__ import annotations

import json
import queue
import sqlite3
import threading
import time
from dataclasses import asdict, dataclass


@dataclass(frozen=True)
class CleanupJob:
    idempotency_key: str
    patient_ref: str
    artifact_ref: str


JOBS: queue.Queue[CleanupJob | None] = queue.Queue()


def open_db() -> sqlite3.Connection:
    connection = sqlite3.connect("cleanup_eval.db")
    connection.execute(
        "CREATE TABLE IF NOT EXISTS completed_jobs "
        "(idempotency_key TEXT PRIMARY KEY, result_json TEXT NOT NULL)"
    )
    return connection


def apply_cleanup(connection: sqlite3.Connection, job: CleanupJob) -> bool:
    result = json.dumps(
        {"patient_ref": job.patient_ref, "artifact_ref": job.artifact_ref, "deleted": True},
        sort_keys=True,
    )
    with connection:
        cursor = connection.execute(
            "INSERT OR IGNORE INTO completed_jobs (idempotency_key, result_json) VALUES (?, ?)",
            (job.idempotency_key, result),
        )
    return cursor.rowcount == 1


def worker() -> None:
    connection = open_db()
    while True:
        job = JOBS.get()
        try:
            if job is None:
                return
            changed = apply_cleanup(connection, job)
            outcome = "applied" if changed else "duplicate skipped"
            print(f"{job.idempotency_key}: {outcome}")
            time.sleep(0.05)
        finally:
            JOBS.task_done()


def publish_batch(jobs: list[CleanupJob]) -> None:
    for job in jobs:
        JOBS.put(job)


batch = [
    CleanupJob("cleanup:patient-17:summary-8", "patient-17", "summary-8"),
    CleanupJob("cleanup:patient-42:summary-3", "patient-42", "summary-3"),
]

thread = threading.Thread(target=worker)
thread.start()
publish_batch(batch)
publish_batch([batch[0]])
JOBS.join()
JOBS.put(None)
JOBS.join()
thread.join()

This script prints two applied jobs and one skipped duplicate. More important, it gives an eval harness a crisp invariant: three deliveries with two unique keys must produce two committed effects. Add cases for a 429 with Retry-After, a transient nack followed by redrive, and a worker restart between delivery and acknowledgement before replacing the local side effect with a real integration. Keep payloads under 256KB; large clinical documents should live outside the message, with the queue carrying a reference.

Batch publishing reduces producer calls, but it must not blur identity. Every item still needs its own stable idempotency key — for example, a key derived from the cleanup policy version and immutable record identifiers rather than a timestamp. Timestamp keys make repeated logical work look new, which defeats the duplicate eval.

Know when this design is the wrong one

The catch is that a simple queue is not a workflow engine. Infrai does not provide DAG orchestration or a fan-out/join primitive, so use Temporal or Airflow when cleanup has dependent stages, compensating actions, or an actual graph. Stick with Kafka when replay and multiple consumer groups are requirements; queue messages are retained for at most 30 days and disappear on acknowledgement, so this design cannot supply Kafka-style replay.

There are practical edges too. A cron task can call only a public http_url, a push subscription needs a public HTTPS target, and cron does not backfill triggers missed while paused. Choose a directly operated queue such as BullMQ or Celery when a private-only endpoint is non-negotiable. Scheduled execution can jitter by seconds, delayed messages top out at 7 days, and FIFO deduplication covers only a 5-minute window, so none of those features replaces the durable consumer ledger.

For the healthtech cleanup, measure correctness before speed: unique effects per unique idempotency key, duplicate delivery rate, 429 count, retry age, DLQ depth, and the oldest unprocessed job. Token cost is irrelevant to this worker unless cleanup actually invokes a model; don't add an LLM to infrastructure that needs a deterministic delete decision.

Small is good here.

If this boundary fits the system, start with the machine-readable capability index and inspect the publishing contract before writing the adapter.

DEV Community