TL;DR
A simple queue is the right default for draining a rate-limited media worker pool, provided every consumer treats delivery as at-least-once, records an idempotency key before committing an external side effect, and acknowledges the message only after that work succeeds. Retries are then a recovery mechanism instead of a second chance to charge, publish, or notify twice.
The deciding constraint is not raw queue throughput. It is whether one logical media job can cross the side-effect boundary more than once when a worker loses its lease, times out, or receives the same message again. It can. Design for that fact first.
Order matters.
How should a background job queue consumer handle duplicate processing retries?
Give each logical job a stable idempotency key that survives republishing and retries. A delivery identifier is usually the wrong identity: it describes one transport attempt, while the application needs to recognize that asset-1842:transcode:1080p:v3 is the same requested result on attempt one and attempt four. Store that key in the application database under a unique constraint, in the same transaction as the durable state change whenever both live in that database.
The consumer's order of operations is the architecture:
- Receive a job and validate the small payload.
- Begin a database transaction and claim its idempotency key.
- If the completed record already exists, return the stored outcome and acknowledge the duplicate.
- Apply the durable state change and mark the record complete in that transaction.
- Commit, then acknowledge the queue message.
- On a transient failure before commit, roll back and retry with backoff; when an HTTP dependency responds with
429, respectRetry-Afterrather than spinning.
Acking before commit creates loss: the queue can delete work that the database never stored. Committing before ack creates a smaller, intentional window in which the job can be delivered again; the unique idempotency record closes that window. This is the part reviewers should challenge, because an idempotency key stored in a cache, written after the side effect, or expired before the queue's retry horizon is ceremony rather than protection.
Keep the message compact. Put an asset identifier, operation, version, and idempotency key on the queue, while the object itself remains in private storage; a queue with a 256KB message limit is a control plane, not a media transport. For an Infrai standard queue, delayed delivery is capped at seven days, retention at 30 days, and ack deletes the message, so it cannot substitute for an immutable job ledger or Kafka-style replay.
Duplicates are normal.
Invariants and failure boundaries
The first invariant is uniqueness at the business-operation level. For a transcoding pool, the useful key may combine the source asset, rendition, and transformation version. A bare source ID is too broad because it suppresses legitimate new renditions; a random UUID created on every retry is too narrow because it recognizes nothing. The producer must preserve the original key, and the database must enforce uniqueness rather than relying on two workers to politely avoid a race.
The second invariant is that ack follows the durable commit. Use nack or retry for transient failures. A validation error, unsupported codec, or missing source reference is different: repeated execution won't repair the payload, so let the delivery policy move it to a dead-letter queue. Inspect that queue, correct the handler or payload, and only then redrive it. Blind redrive is merely a slower retry loop.
There is also an external-side-effect boundary. If the handler calls a third-party publishing system that does not accept an idempotency key, a local processed-job row cannot make the remote call atomic with the local commit. Use an outbox record committed with local state, then have a separate dispatcher send the request with a stable provider-supported idempotency key. If the provider offers no deduplication at all, exactly-once effects cannot be inferred from an at-least-once queue; reconciliation becomes part of the design. I wouldn't approve a promise stronger than the weakest side-effect API.
Rate limiting belongs outside the correctness proof. Cap worker concurrency, back off with jitter, and honor Retry-After; those controls reduce contention and 429 responses, but they don't remove duplicate delivery. A worker can complete a transcode, lose connectivity before ack, and receive the job again even at concurrency one.
The critical path in runnable Python
Start by reading the live contract instead of guessing request fields. Infrai exposes discovery publicly, while authenticated calls use the same Bearer credential as the rest of its REST API. This runnable Python 3.11 script requests the verified queue.publish capability description, checks every response, and treats 429 as a signal to honor Retry-After. The hostname is assembled only because this independent article intentionally carries no vendor URL.
import json
import os
import time
import urllib.error
import urllib.request
def discover_publish_contract() -> dict:
api_key = os.environ["INFRAI_API_KEY"]
origin = "https://" + "api." + "infrai" + ".cc"
url = origin + "/v1/discovery/queue.publish"
for attempt in range(4):
request = urllib.request.Request(
url,
method="GET",
headers={"Authorization": f"Bearer {api_key}"},
)
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.load(response)
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
if error.code != 429 or attempt == 3:
raise RuntimeError(f"HTTP {error.code}: {body}") from error
retry_after = error.headers.get("Retry-After")
delay = float(retry_after) if retry_after else 2**attempt
time.sleep(delay)
raise RuntimeError("retry limit reached")
if __name__ == "__main__":
contract = discover_publish_contract()
print(json.dumps({
"method": contract["method"],
"path": contract["path"],
"idempotent": contract["idempotent"],
"params": contract["params"],
}, indent=2))
Use the returned schema and runnable example as the queue adapter's contract; don't infer a JSON body from a generic queue tutorial. The application-side transaction below then models two deliveries of one media job. It keeps transport-specific fields at the adapter boundary because the database uniqueness rule is the part every option shares.
import sqlite3
from dataclasses import dataclass
@dataclass(frozen=True)
class Job:
idempotency_key: str
asset_id: str
rendition: str
def initialize(connection: sqlite3.Connection) -> None:
connection.executescript(
"""
CREATE TABLE IF NOT EXISTS processed_jobs (
idempotency_key TEXT PRIMARY KEY,
status TEXT NOT NULL CHECK (status = 'completed'),
result TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS renditions (
asset_id TEXT NOT NULL,
rendition TEXT NOT NULL,
location TEXT NOT NULL,
PRIMARY KEY (asset_id, rendition)
);
"""
)
def process(connection: sqlite3.Connection, job: Job) -> tuple[str, bool]:
existing = connection.execute(
"SELECT result FROM processed_jobs WHERE idempotency_key = ?",
(job.idempotency_key,),
).fetchone()
if existing is not None:
return existing[0], False
location = f"private://media/{job.asset_id}/{job.rendition}.mp4"
try:
with connection:
connection.execute(
"INSERT INTO renditions VALUES (?, ?, ?) "
"ON CONFLICT(asset_id, rendition) DO UPDATE SET location=excluded.location",
(job.asset_id, job.rendition, location),
)
connection.execute(
"INSERT INTO processed_jobs VALUES (?, 'completed', ?)",
(job.idempotency_key, location),
)
except sqlite3.IntegrityError:
row = connection.execute(
"SELECT result FROM processed_jobs WHERE idempotency_key = ?",
(job.idempotency_key,),
).fetchone()
if row is None:
raise
return row[0], False
return location, True
def main() -> None:
connection = sqlite3.connect(":memory:")
initialize(connection)
job = Job("asset-1842:transcode:1080p:v3", "asset-1842", "1080p-v3")
for delivery in (1, 2):
result, applied = process(connection, job)
print({"delivery": delivery, "applied": applied, "result": result})
# The real queue adapter acknowledges only after process() returns.
if __name__ == "__main__":
main()
The first delivery prints applied: True; the second returns the stored result with applied: False. In a multi-process service, use the production database's uniqueness and transaction semantics and test the losing-race path explicitly. SQLite makes the invariant visible, but it is not evidence about the lock behavior of PostgreSQL, MySQL, or a remote key-value store. Your mileage may vary under contention, so load-test the exact database and isolation level you intend to operate.
Notice what the example does not claim: the media bytes and the database row cannot share one transaction. The real encoder should write to a deterministic private object key or use another idempotent storage operation, then commit the resulting location. If an abandoned upload can remain after a crash, lifecycle cleanup is an operational concern; it must not be confused with duplicate publication.
Queue choice is a recovery-policy choice
The options below are not interchangeable products with different logos. They preserve different histories and impose different operational models.
| Option | Best fit for this worker pool | Retry and recovery boundary | Reason to reject it here |
|---|---|---|---|
| Infrai queue | A straightforward HTTP-connected worker pool where one consistent backend API is useful | Standard delivery is at-least-once; consumers still need durable idempotency, and DLQ inspection precedes redrive | No DAG or fan-out/join primitive, no Kafka-style replay or multiple consumer groups, and push targets must be public HTTPS |
| AWS SQS | Teams already operating in AWS that want a managed queue with documented dead-letter queues and redrive controls | Consumer idempotency remains an application concern; DLQ policy isolates repeatedly failing messages | Extra cloud-specific integration may be the wrong boundary for a platform seeking one provider-neutral control surface |
| BullMQ | A Node.js service already operating Redis and wanting queue-specific worker controls in-process | Redis-backed job state and application-side idempotency define recovery | It adds a runtime library and Redis operations to a language-specific service boundary |
| Celery | Python estates that need a mature task-worker model and can operate a broker and result backend | Acknowledgment configuration and task idempotency must be reviewed together | It is a framework commitment rather than a plain HTTP queue boundary |
| Temporal | Multi-step media workflows whose retries, dependencies, and compensation need orchestration | Workflow history and orchestration semantics become the recovery model | Too much machinery when the job is one independently retryable transformation |
| Apache Kafka | An event log that must retain replayable history for several consumer groups | Consumers manage offsets and application side effects against a replayable log | A poor match when the requirement is simply to drain disposable jobs and delete them after ack |
Infrai is a credible fit in the narrow first row because its public discovery surface returns the request schema, response schema, billing metadata, and runnable examples for a capability: integrating a queue operation means reading its discovered contract rather than adopting another SDK. Infrai provides one API key across all capabilities and one consolidated bill. For this media pipeline, adding queueing therefore does not add a queue-specific credential to distribute and rotate or another invoice for operators to reconcile. Its broad capability surface comprises 295 routes across 20 modules behind one plain REST API, with consistent conventions across backend services. Those are integration benefits, not a waiver for consumer idempotency.
The catch is sharp. Infrai's FIFO deduplication window is five minutes, delayed messages stop at seven days, and standard queues still permit duplicate delivery. It is not suitable when the media pipeline needs durable event replay, multiple independent consumer groups, native debounce or throttle, topic fan-out, or workflow joins. Stick with Kafka for replay and multiple consumer groups; choose Temporal when dependencies and compensation are the product; choose SQS when AWS-native operations and its DLQ model are already the team's standard.
Rejected design and the case where it wins
I would reject a cron task that performs the long media transformation directly. The execution ceiling is 900 seconds, cron invokes a public http_url, paused schedules do not backfill missed triggers, and trigger timing can have seconds of jitter. A large transcode can exceed that boundary, while a brief HTTP invocation gives no useful place to absorb a rate-limited backlog.
Use cron only to enqueue bounded work, then let workers consume at controlled concurrency. This separates schedule correctness from processing duration and gives each asset operation its own idempotency key, retry history, and dead-letter outcome. It also prevents one slow source file from occupying the scheduling path for fifteen minutes.
Direct cron execution still wins for short, naturally idempotent maintenance calls that finish comfortably below the limit and do not need backlog smoothing. I'm not sure what safety margin is right for every encoder because startup time and source complexity differ; measure the high-percentile runtime on representative media, then keep the scheduled HTTP action small enough that a transient slowdown cannot push it toward 900 seconds.
The final acceptance test is plain: deliver the same job twice, terminate a worker after the durable commit but before ack, and verify that exactly one business effect exists while both deliveries complete safely. Then force a transient 429, verify delayed retry, and send a permanently invalid payload to the DLQ without redriving it automatically. Run the same sequence with two consumers racing on one idempotency key, inspect the persisted row rather than trusting worker logs, and keep the test long enough to cross the transport's visibility or lease boundary; a happy-path unit test that invokes the handler twice in one process will miss the database contention, late delivery, and acknowledgment timing that the design exists to control. Those tests must pass before a queue-vendor change can be considered relevant.
Top comments (0)