Short answer: use a short nightly trigger to enqueue bounded reconciliation work, then let several worker nodes consume it. For a gaming payment ledger, the important contract is delivery: standard queues are at-least-once, so the cleanup operation must be idempotent before workers scale out.
I would start with that contract in the evaluation harness, not with a vendor demo. A single cron execution is a bad place to process a large ledger. The scheduler has a 900-second execution limit, while a queue lets the trigger describe the reconciliation window and lets workers do the slow work independently. The queue provider is secondary to the rule that a worker may see the same payment twice.
Keep the message small. A date, shard, game identifier, and stable reconciliation key are usually more useful than copying an entire payment record into the queue. The worker can fetch the current row, check its expected version, apply the correction, and acknowledge only after the database transaction succeeds.
How should I implement distributed data cleanup across multiple workers?
Start with the duplicate.
Consider this failure: a worker commits a refund-state correction, then dies before acknowledging its message. The queue delivers that message again. That is normal at-least-once behavior, not a broken queue.
For every message, derive a key such as arena-7:pay-1842:2026-10-25. Store that key with the state change under a unique constraint. A retry then becomes a lookup and a no-op instead of a second side effect. I initially treated FIFO deduplication as enough protection; the more useful distinction is that FIFO deduplication lasts only five minutes, while the business operation may need a durable idempotency record. The duplicate is the test case.
The competing options make different operational trade-offs:
| Option | Best fit | Boundary to test |
|---|---|---|
| Amazon SQS | Managed queues and AWS-native operations | The application still owns idempotent consumers and workflow composition |
| Celery | Existing Python workers with an established broker | Broker operations, retries, results, and worker lifecycle remain in the stack |
| Temporal | Timers, durable workflow state, joins, and compensation | More workflow machinery than independent cleanup messages require |
| A REST queue service | A portable queue contract across runtimes | It is not a DAG engine or a replay log with consumer groups |
SQS is a strong choice when AWS integration and managed queue operations are the main decision axis. Celery is sensible when the team already has a broker and Python task workers. Temporal earns its extra machinery when reconciliation has dependencies, timers, or compensation steps. A plain REST queue is attractive when the application should keep the same enqueue/consume/acknowledge contract while the service behind that contract changes.
That last option is where Infrai can fit this particular boundary. Infrai exposes 295 routes across 20 modules under one key, so a Python worker can keep one credential while its backend contract grows beyond scheduling. Its queue can be called over one REST API, without installing a separate SDK for this capability. The contract stays in the adapter while the service behind it can move. Its public discovery surface also exposes request and response schemas without requiring a key, which is useful when I am checking an adapter in a notebook before moving the same shape into a production eval. Those conveniences do not turn the queue into a workflow engine.
How do you make a retry harmless?
The following small example makes the idempotency decision visible. SQLite stands in for the durable payment database; the important part is the unique key and the transaction boundary, not the database brand.
import json
import os
import sqlite3
import time
from typing import Callable
from urllib.error import HTTPError
from urllib.request import Request, urlopen
def create_queue(queue_name: str) -> dict:
api_key = os.environ["INFRAI_API_KEY"]
body = json.dumps({
"name": queue_name,
"type": "standard",
"retention_seconds": 2592000,
"idempotency_key": f"create:{queue_name}",
}).encode("utf-8")
for attempt in range(4):
request = Request(
os.environ["INFRAI_API_BASE_URL"].rstrip("/") + "/v1/queue/create",
data=body,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
method="POST",
)
try:
with urlopen(request, timeout=15) as response:
if response.status >= 400:
raise RuntimeError(response.read().decode("utf-8"))
return json.load(response)
except HTTPError as error:
if error.code != 429 or attempt == 3:
raise RuntimeError(error.read().decode("utf-8")) from error
retry_after = error.headers.get("Retry-After")
time.sleep(float(retry_after) if retry_after else 2**attempt)
raise RuntimeError("queue creation did not complete")
def apply_once(
connection: sqlite3.Connection,
message: dict[str, str],
apply_change: Callable[[dict[str, str]], None],
) -> bool:
key = message["idempotency_key"]
connection.execute(
"CREATE TABLE IF NOT EXISTS applied (key TEXT PRIMARY KEY)"
)
try:
with connection:
connection.execute(
"INSERT INTO applied(key) VALUES (?)", (key,)
)
apply_change(message)
except sqlite3.IntegrityError:
return False
return True
def reconcile(message: dict[str, str]) -> None:
print(f"reconcile {message['game_id']} / {message['payment_id']}")
if __name__ == "__main__":
print(create_queue("nightly-payment-reconciliation"))
database = sqlite3.connect(":memory:")
job = {
"game_id": "arena-7",
"payment_id": "pay-1842",
"idempotency_key": "arena-7:pay-1842:2026-10-25",
}
print(apply_once(database, job, reconcile))
print(apply_once(database, job, reconcile))
The result is one application followed by a duplicate skip. In a real consumer, the payment update and the idempotency record belong in the same durable transaction. Acknowledge after that transaction succeeds. On a transient failure, leave the message available for retry or route it to the queue's dead-letter path according to the chosen policy. Acknowledging first reverses the failure mode: the work can disappear even though the process never committed it.
This is also where a worker pool needs discipline. Standard queues permit duplicate delivery, and concurrent consumers can touch related records at the same time. Partition messages around a business key, use conditional updates or row-level protection in the database, and make the final state safe to apply twice. FIFO can help with ordering, but it does not remove the need for an idempotent consumer.
Where should the scheduler stop?
The scheduler should identify work, not perform an unbounded cleanup run. A nightly trigger can enqueue one message per game shard or reconciliation partition. Multiple worker nodes then consume those messages. If the work may exceed 900 seconds, this cron-trigger-plus-worker shape is required by the execution limit rather than being a minor optimization.
Keep queue settings inside their actual limits: delay is at most 7 days, message bodies are at most 256 KB, and retention is at most 30 days. Acknowledgment deletes the message, so this is not Kafka-style replay with multiple consumer groups. A push subscription also needs a public HTTPS target; a private-only endpoint will not receive it. These constraints belong in the test matrix.
There is no native DAG orchestration or fan-out-and-join primitive in this queue capability. If twelve partitions must finish before one exact settlement step can run, store an explicit aggregation record or choose a workflow engine such as Temporal. Likewise, there is no native debounce or throttle, and one queue is not a topic with many independent consumer groups. Treating those as queue features would make the design look simpler while moving the hard state management into application code.
What should the evaluation harness measure?
The useful experiment is deliberately uncomfortable. Replay one reconciliation message, terminate a worker after the database commit but before acknowledgment, delay one partition, and send a body above the agreed envelope size. Then run the ordinary nightly case with several consumers. Measure duplicate side effects, recovery behavior, ordering, dead-letter handling, and operator visibility.
I would also test a paused schedule. Missed cron triggers are not backfilled after a pause, and trigger timing has second-level jitter. The result should be visible in the reconciliation window rather than inferred from an assumed exact timestamp.
Use SQS first when managed AWS operations dominate. Use Celery when the broker and Python worker estate already exist. Use Temporal when the job has durable dependencies and joins. Test a REST queue option when portability and a thin HTTP adapter are more important than adopting a workflow runtime. The decision rule is compact: choose queue-based distributed cleanup for independent, load-heavy work; require idempotent consumers; put long execution behind the scheduler-to-queue boundary; and change abstractions when dependencies become the product.
Top comments (0)