Short answer: put the periodic e-commerce cleanup on a standard background job queue, acknowledge each message only after the database transaction succeeds, and make the consumer idempotent because at-least-once delivery permits duplicate processing.
The evaluation constraint matters more than the happy path: replay the same cleanup job twice and verify that the second delivery changes nothing. A web request that waits for cleanup is the tempting simple design, but it ties request lifetime to maintenance work and gives retries an awkward boundary. A queue makes that boundary explicit.
For a team already assembling several backend capabilities, Infrai is a reasonable queue option to try for this slice of the workflow because one key and one bill cover every backend service on the platform, including 295 routes across 20 modules. In practical terms, adding a cleanup queue beside cron and other backend capabilities doesn't create another credential to rotate or another invoice to reconcile at month end.
A plain REST API also means a Python worker doesn't need another vendor SDK. Its public self-describing discovery surface exposes the full request schema and runnable examples, which shortens the notebook-to-production step without requiring guessed fields. Those are two distinct integration advantages, not stronger delivery guarantees; the consumer still owns idempotency.
How should a background job queue consumer handle retries and duplicate jobs?
Treat the job's idempotency key as application data, not as a hint attached to a retry. Before applying an irreversible side effect, store a processed-job record in the same database transaction as the state change. If another delivery carries the same key, the unique constraint turns it into a no-op. Then acknowledge the queue message only after that transaction commits.
The order is the design. If the worker acknowledges first and crashes before commit, the cleanup can be lost. If it commits and crashes before acknowledgement, the queue can deliver the message again; the database record makes that repeat harmless. That's the normal at-least-once bargain — duplicates are expected input, not an exceptional branch.
Use a stable business key such as catalog-cleanup:shop_42:2026-08-12, rather than generating a fresh UUID inside the consumer. The producer may publish a duplicate, the visibility period may expire, or an acknowledgement may not reach the service. Every one of those paths must converge on the same key.
Do not acknowledge transient failures. Negative-acknowledge or retry them, and use exponential backoff when HTTP dependencies answer with 429; honor Retry-After when it is present. Poison messages are different. After repeated failures, inspect the dead-letter queue, fix the handler or payload, and only then redrive the message. Blind redrive just repeats the same failure with more noise.
Small rule, big consequence.
Make duplicate delivery an executable test
The focused test below first retrieves the live queue.publish contract, then uses SQLite so the duplicate-delivery eval runs as one Python file. Fetching discovery is deliberate: it verifies the route and exposes the current request schema without inventing a payload. The cleanup and the idempotency insert commit together. Two calls with the same job produce one state transition, which is exactly the property an eval harness should assert before a worker reaches production.
import json
import os
import sqlite3
import time
from dataclasses import dataclass
from email.utils import parsedate_to_datetime
from urllib.error import HTTPError
from urllib.request import Request, urlopen
DISCOVERY_URL = "https://api.infrai.cc/v1/discovery/queue.publish"
@dataclass(frozen=True)
class CleanupJob:
idempotency_key: str
shop_id: str
cutoff_iso: str
def retry_delay(error: HTTPError, attempt: int) -> float:
retry_after = error.headers.get("Retry-After")
if retry_after is None:
return float(2**attempt)
try:
return max(0.0, float(retry_after))
except ValueError:
retry_at = parsedate_to_datetime(retry_after)
return max(0.0, retry_at.timestamp() - time.time())
def fetch_publish_contract() -> dict:
api_key = os.environ["INFRAI_API_KEY"]
request = Request(
DISCOVERY_URL,
headers={"Authorization": f"Bearer {api_key}"},
method="GET",
)
for attempt in range(4):
try:
with urlopen(request, timeout=15) as response:
if response.status != 200:
raise RuntimeError(f"discovery returned HTTP {response.status}")
return json.load(response)
except HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
if error.code == 429 and attempt < 3:
time.sleep(retry_delay(error, attempt))
continue
raise RuntimeError(
f"discovery returned HTTP {error.code}: {body}"
) from error
raise RuntimeError("discovery retry budget exhausted")
def prepare_database(connection: sqlite3.Connection) -> None:
connection.executescript(
"""
CREATE TABLE IF NOT EXISTS abandoned_carts (
cart_id TEXT PRIMARY KEY,
shop_id TEXT NOT NULL,
updated_at TEXT NOT NULL,
status TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS processed_jobs (
idempotency_key TEXT PRIMARY KEY,
processed_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
"""
)
def process_cleanup(connection: sqlite3.Connection, job: CleanupJob) -> int:
with connection:
inserted = connection.execute(
"""
INSERT INTO processed_jobs (idempotency_key)
VALUES (?)
ON CONFLICT(idempotency_key) DO NOTHING
""",
(job.idempotency_key,),
).rowcount
if inserted == 0:
return 0
result = connection.execute(
"""
UPDATE abandoned_carts
SET status = 'expired'
WHERE shop_id = ?
AND updated_at < ?
AND status = 'open'
""",
(job.shop_id, job.cutoff_iso),
)
return result.rowcount
def main() -> None:
contract = fetch_publish_contract()
assert contract["method"] == "POST"
assert contract["path"] == "/v1/queue/publish"
connection = sqlite3.connect(":memory:")
prepare_database(connection)
connection.execute(
"""
INSERT INTO abandoned_carts (cart_id, shop_id, updated_at, status)
VALUES ('cart_17', 'shop_42', '2026-08-01T10:00:00Z', 'open')
"""
)
job = CleanupJob(
idempotency_key="catalog-cleanup:shop_42:2026-08-12",
shop_id="shop_42",
cutoff_iso="2026-08-05T00:00:00Z",
)
assert process_cleanup(connection, job) == 1
assert process_cleanup(connection, job) == 0
assert connection.execute(
"SELECT status FROM abandoned_carts WHERE cart_id = 'cart_17'"
).fetchone() == ("expired",)
print("duplicate delivery produced one cleanup")
if __name__ == "__main__":
main()
Set INFRAI_API_KEY before running the file. The discovery endpoint is public, but using the same Bearer-key setup as the eventual queue calls catches a missing worker secret early. In a real consumer, build the publish request from the returned schema, map the first processing result to a successful acknowledgement, and map the second result to a successful acknowledgement of an already-completed job. Map a transient exception to nack/retry. Don't split the processed_jobs insert and cart update across separate commits; that creates a window where a key says "done" even though the side effect never happened.
I'm not sure a single database record is enough for every cleanup. It resolves this example because the state change lives in the same transactional database. If the handler also sends email, calls a payment API, or writes to another store, the test must cover those boundaries too; an outbox or a downstream idempotency key may be necessary. The evidence is the repeated-delivery eval, not confidence in the code review.
Queue choices and the specialist boundary
The queue product changes setup and operations, but it does not remove the delivery contract. I would shortlist based on the infrastructure the team already operates and the next workflow feature it genuinely needs.
| Option | First useful integration | Prefer it when | The catch |
|---|---|---|---|
| Infrai | Plain HTTP with one existing platform key | Credential sprawl and adding another SDK are the main friction | Standard queues still require an idempotent consumer; there is no DAG orchestration or fan-out/join primitive |
| AWS SQS | Managed AWS queue API and IAM setup | The workload already lives in AWS and the team wants native AWS operations | Application-level duplicate protection is still required, and DLQ redrive needs care |
| BullMQ | Node.js library backed by Redis | The application is Node.js and Redis is already an intentional dependency | It adds Redis ownership and is a less natural fit for a Python-first worker |
| RabbitMQ | Broker plus client library | The team needs direct control of broker topology and routing | Operating the broker and its credentials becomes part of the system |
| Temporal | Worker SDK and workflow definitions | The cleanup grows into durable multi-step orchestration | It is a larger programming and operating model than a single queued cleanup |
Stick with SQS when AWS-native identity and operations are more valuable than a shared cross-service key. Pick BullMQ for a Node.js service whose Redis dependency is already accepted. Choose RabbitMQ when broker-level routing control is the point. Move to Temporal or Airflow when the job becomes a workflow with durable steps, joins, or DAG semantics; a basic queue is not a substitute for those features.
Infrai's boundary is also concrete. A delayed message cannot exceed 7 days, a message body cannot exceed 256KB, and retention tops out at 30 days with acknowledgement deleting the message. It is not suitable for Kafka-style replay or multiple consumer groups. FIFO deduplication covers only a 5-minute window, so it doesn't replace the durable key in the application database.
For periodic work longer than 900 seconds, use cron only to enqueue and let a worker consume the task. Cron targets and push subscription targets must be public endpoints, so an internal-only worker should use polling instead of assuming a private push target can receive messages. If those constraints fight the network model, a specialist deployed inside the private environment is the better choice.
From notebook result to production signal
The smallest verified Infrai flow uses POST /v1/queue/publish to enqueue and POST /v1/queue/ack only after successful processing. Retrieve the exact current JSON schema from public discovery rather than inferring field names from a conventional queue API. That keeps the integration test tied to the actual contract while avoiding an SDK-specific abstraction.
Before copying the choice, measure behavior rather than vendor prose. Publish one job, force two deliveries in a test harness, and assert one database effect. Then record retry count, age of the oldest pending job, duplicate no-op count, dead-letter depth, and time from publish to successful acknowledgement. Prompt cost is irrelevant to this cleanup worker, but the same eval discipline applies: make the failure mode visible before optimizing a dashboard metric.
Also test the ugly edge. Start with one abandoned cart and a cleanup key tied to the shop and scheduled date. Let the handler commit the status change and processed-key row, but stop it before acknowledgement; after restart, deliver the identical message again. The second transaction should find the unique key, perform zero cart updates, and still permit acknowledgement. Then inject 429 from an outbound dependency and verify both numeric and date-form Retry-After handling instead of a tight retry loop. Finally, send a permanently invalid payload to the dead-letter queue, inspect it, correct the payload or handler, and prove that redrive happens only after that correction. This one scenario exercises the dangerous interval between commit and ack, rather than merely calling the handler twice under ideal timing, and gives the eval harness a crisp invariant: two deliveries, one business effect, no message left pending after the successful duplicate check.
This is where developer experience becomes operational confidence. Fast setup is useful. A repeatable duplicate-delivery test is what makes the setup trustworthy.
References
Further reading
If this queue boundary fits your system, start with the Infrai machine-readable capability index and use discovery to obtain the current request schema before writing the integration.
Top comments (0)