Background job processing patterns including job queues, scheduled jobs, worker pools, retry strategies, and delivery guarantees. Use when implementing async processing, ETL pipelines, ML training jobs, cron schedules, or dead letter queues.
Instrucciones de origen · Vista previa de solo lectura
name
loom-background-jobs
description
Background job processing patterns including job queues, scheduled jobs, worker pools, retry strategies, and delivery guarantees. Use when implementing async processing, ETL pipelines, ML training jobs, cron schedules, or dead letter queues.
Reliable async task execution: enqueue work, process it in workers decoupled from the request cycle, and survive crashes/retries without corrupting state. This file covers job queues, retries/backoff, DLQs, scheduling, worker pools, and delivery guarantees.
For pub/sub, event sourcing, CQRS, sagas, and streaming brokers (Kafka/Pulsar), see loom-event-driven — don't reimplement those here.
The two invariants everything hangs off
At-least-once is the default. Design every handler to be idempotent. Redis-backed queues (Sidekiq, BullMQ, Celery+Redis), SQS standard, and any queue that retries on crash will deliver a job more than once. "Exactly-once delivery" does not exist over an unreliable network; the achievable goal is exactly-once effect = at-least-once delivery + idempotent handler.
Ack after success, never before. The job must stay owned by the worker until the side effect is durably committed. Ack-then-process = at-most-once = silent data loss on crash. Process-then-ack = at-least-once = duplicates you dedup away. Always choose the latter.
Idempotency: the non-negotiable pattern
Derive a stable key from the job's business identity (not a random UUID per enqueue), and gate the side effect on it.
defhandle(job):
key = job["idempotency_key"] # e.g. f"charge:{order_id}"if store.setnx(f"done:{key}", "1", ex=7*86400): # first time → claim
do_side_effect(job) # charge card, send email, insert row# else: already done → no-op, safe to return success
⚠ Claim before the effect, but only mark "done" after it commits — if you set the flag first and then crash, the retry sees "done" and skips a side effect that never happened. Best: idempotent write at the sink (unique constraint / upsert / conditional put) rather than a separate flag. The DB unique index is the most reliable dedup store.
The processed-marker TTL must exceed max retry window + queue retention, or a late redelivery re-runs the job.
Idempotency covers duplicate delivery; it does not cover concurrent duplicate execution (two workers at once). For that you need a lock or an atomic conditional write — see visibility-timeout races below.
Rule of thumb: for durability without running infra, prefer a managed broker (SQS) or a real AMQP/log broker. Redis-list queues are fast and simple but the queue lives in one Redis instance — plan for AOF/replication and accept the crash-window caveats above.
Visibility timeout / lock races (the #1 duplicate-execution bug)
A worker leases a job for T (SQS visibility timeout / Celery visibility_timeout / BullMQ lockDuration). If processing exceeds T, the queue assumes the worker died and hands the same job to a second worker — now it runs twice concurrently, and idempotency-by-marker won't save a non-atomic side effect.
Mitigations, in order of preference:
Keep handlers well under T. Size the timeout to p99 duration × safety factor.
Heartbeat / extend the lease for legitimately long jobs: SQS ChangeMessageVisibility, BullMQ auto-renews while the processor runs (older Bull needs care), Celery — split the work or raise visibility_timeout (but that also delays recovery of genuinely-dead jobs).
Make the sink atomic (conditional put / UPDATE ... WHERE status='pending') so a concurrent second run is a no-op.
❌ visibility_timeout = 60s, job takes 90s → guaranteed double-processing
✅ heartbeat every 30s, or checkpoint & keep each unit < timeout
Retries, backoff, and retry storms
Exponential backoff with FULL JITTER. Fixed or un-jittered backoff synchronizes all failed jobs to retry at the same instant → a thundering herd that re-topples a recovering dependency.
delay = random_between(0, min(cap, base * 2 ** attempt)) # full jitter (AWS)
Full jitterrand(0, backoff) beats "equal jitter" and beats no jitter for spreading load — use it by default.
Cap the delay (e.g. 10 min) and cap attempts. Unbounded retries = zombie jobs clogging the queue forever.
Classify errors before retrying. Retry only transient faults (timeouts, 429, 503, connection reset). Do not retry deterministic failures (validation error, 400, NotFound) — they'll fail identically N times, waste capacity, and delay the DLQ. Fail fast to DLQ instead.
Circuit breaker in front of a flapping dependency: after M consecutive failures, open the breaker and fast-fail (or pause the queue) for a cooldown so you stop hammering it and stop burning retry budget. See loom-error-handling for breaker internals.
Retry budget: total attempts × concurrency must not exceed the downstream's capacity, or retries themselves become the outage.
Framework retry config (canonical)
# Celery — autoretry with jittered backoff + late acksclassBase(Task):
autoretry_for = (ConnectionError, TimeoutError) # transient ONLY
retry_backoff = True# exponential
retry_backoff_max = 600# cap 10 min
retry_jitter = True# ON (default) — keep it
retry_kwargs = {"max_retries": 5}
app.conf.update(task_acks_late=True, task_reject_on_worker_lost=True)
A poison pill is a job that fails every attempt (bad data, unhandled shape). Without a ceiling it retries forever and can head-of-line-block the queue.
Policy: max attempts → move to DLQ → alert → keep the payload + error + attempt count → provide redrive tooling to replay after a fix.
SQS: set RedrivePolicy with maxReceiveCount; DLQ is a normal queue you can redrive from (console "start DLQ redrive" or re-enqueue).
Give the DLQ its own retention and monitoring — a silently-filling DLQ is an outage you haven't noticed. Alert on DLQ depth > 0.
Redrive after the fix, not blindly — replaying poison into the same broken handler just refills the DLQ. Fix root cause, then redrive in controlled batches.
Store enough context to reproduce: original payload, failing worker version, stack, first/last-failed timestamps.
Duplicates start at the producer too: a retried HTTP request or an at-least-once upstream enqueues the same job twice. Dedup on enqueue with a deterministic job id.
BullMQ / Bull: pass jobId — a job with an existing id is ignored while present.
SQS FIFO: MessageDeduplicationId (5-minute dedup window) or content-based dedup.
Celery: no native enqueue dedup — use a SETNX guard or a unique DB row keyed on the business id.
⚠ jobId dedup only holds while the job is still in the queue/known set. Once completed and evicted (see removeOnComplete), the same id can be enqueued again — so producer dedup complements, never replaces, handler idempotency.
Claim-check: don't put big payloads on the queue
Store the large blob (file, image, dataset row batch) in object storage / DB and enqueue only a reference + integrity hash. Queues (SQS 256KB limit, Redis memory) are for coordination, not bulk data.
Benefits: small fast messages, no broker bloat, payload survives independently of retries, and re-delivery re-reads the source of truth rather than a stale copy.
Scheduled / cron jobs
Distinct from queues: a scheduler decides when, then enqueues a normal job. Celery Beat, BullMQ repeatable jobs, Sidekiq-cron, or a DB-backed scheduler.
app.conf.beat_schedule = {
"nightly-report": {"task": "reports.daily", "schedule": crontab(hour=9, minute=0)},
"health": {"task": "ops.health", "schedule": 300.0}, # every 5 min
}
Scheduling gotchas (each has bitten someone in prod)
Overlap / double-fire. Beat, repeatable jobs, or a cron on N replicas will each fire the tick → the same run enqueued N times, or the previous run still executing when the next starts. Prevent with a leader lock: SET lock:job NX PX <ttl> (SETNX) around the run; only the holder proceeds. Run exactly one Beat process, or use a distributed lock for the scheduler itself.
if redis.set(f"cron:{name}", worker_id, nx=True, px=ttl_ms):
try: run()
finally: redis.delete(f"cron:{name}") # or let TTL expire; TTL > max runtime
Always schedule in UTC. Local-time cron double-runs (fall-back) or skips (spring-forward) an hour on DST transitions; a job at 02:30 local may run twice or never. Store/evaluate schedules in UTC and convert for display only.
Missed runs after downtime. If the scheduler was down at fire time, the tick is gone — most schedulers do not back-fill. Decide per job: catch-up (run once for the missed window — good for reports/aggregations that must not skip a period) vs skip (only run the latest — good for "sync current state" jobs where stale runs are useless). Don't naively replay every missed tick, or a 6-hour outage triggers 72 five-minute jobs at once.
Cron drift & long ticks. If the job takes longer than the interval, ticks pile up. Guard with overlap prevention (above) or a "skip if previous still running" flag.
Timezone of crontab(hour=…) follows CELERY_TIMEZONE/app tz — verify it; a silently-local Beat is a classic 1am/2am incident.
Worker pools, concurrency & sizing
CPU-bound (image/video, ML): concurrency ≈ number of cores; more than that just thrashes context switches. Use processes (Celery --concurrency, separate workers), not threads, to sidestep the GIL in Python.
IO-bound (HTTP calls, DB): concurrency can far exceed cores; use async/greenlets (Celery -P gevent/eventlet, Node's single-loop concurrency) and size against the downstream's connection/rate limit, not CPU.
Recycle workers to bound leaks: Celery worker_max_tasks_per_child, worker_max_memory_per_child.
prefetch_multiplier: high prefetch starves other workers of queued jobs and holds them past a crash. For long/uneven tasks set worker_prefetch_multiplier=1 (fair dispatch); for many short tasks a higher prefetch cuts round-trips.
Separate queues per workload class. One slow bulk queue on the same worker as latency-sensitive jobs = head-of-line blocking. Isolate email, images, compute onto dedicated queues + workers so a backlog in one never starves another. Route with task_routes (Celery) / distinct Queues (BullMQ).
Priority & fairness (starvation)
Priority queues let critical jobs jump ahead — but strict priority starves low-priority work if high-priority never drains. Mitigate with weighted/round-robin consumption or aging (bump a job's priority the longer it waits). A tenant flooding a shared queue also starves others → per-tenant queues or a fair scheduler that caps any one tenant's share of worker capacity.
Graceful shutdown (SIGTERM draining)
A deploy/scale-down sends SIGTERM. If you exit immediately, in-flight jobs are killed mid-run → at-least-once saves you only if the job wasn't acked yet. Drain instead:
process.on("SIGTERM", async () => {
awaitPromise.all(workers.map(w => w.pause())); // stop pulling new jobsawaitPromise.all(workers.map(w => w.close())); // wait for in-flight to finish
process.exit(0);
});
Give the orchestrator a grace period ≥ longest job (K8s terminationGracePeriodSeconds, else SIGKILL truncates the drain). For jobs longer than any sane grace period, rely on checkpointing (below) so a kill just resumes.
Celery warm shutdown = one TERM (finish current, stop taking new); a second TERM/QUIT = cold (interrupt). Configure the deployment to send one and wait.
With late acks + drain, a killed job is simply redelivered — which again requires an idempotent handler. Everything routes back to invariant #1.
Long-running jobs: checkpoint & resume
Any job that can exceed the visibility timeout or a deploy grace period must be resumable, not restart-from-zero. Checkpoint progress durably; on retry, resume from the last checkpoint.
@app.task(bind=True, acks_late=True)deftrain(self, run_id, resume_from=None):
state = load_ckpt(resume_from) if resume_from else fresh_state()
for epoch inrange(state.epoch, TOTAL):
step(state)
if epoch % CKPT_EVERY == 0:
save_ckpt(run_id, epoch, state) # durable → survives crash/redeliverself.update_state(state="PROGRESS", meta={"epoch": epoch})
return finalize(state)
Checkpoint to durable storage (S3/DB), not local disk a rescheduled worker can't see.
Make each checkpoint write idempotent/versioned so a duplicated final step doesn't double-commit results.
Break a huge job into many small enqueued units (per-batch jobs) when possible — smaller units mean cheaper retries, better parallelism, and no visibility-timeout fights. Fan-out then aggregate (see loom-event-driven for saga/chord orchestration).
Backpressure & queue-depth monitoring
The queue depth is your leading indicator. Monitor it and act on it — a queue growing faster than it drains is an unbounded-latency outage in slow motion.
Alert on: queue depth (waiting), oldest-message age (the truest "am I keeping up?" signal), processing latency p95/p99, error rate, DLQ depth > 0, active worker count.
Backpressure at the producer: when depth exceeds a threshold, shed load, reject/429 new work, or slow enqueue — don't let an unbounded queue defer the failure into a memory/retention blowup.
Autoscale workers on depth or age, not CPU alone (a queue can back up while workers idle on IO).
Cap retained completed/failed sets (removeOnComplete/removeOnFail, SQS retention) so bookkeeping doesn't exhaust the broker.
Golden signals for a queue:
waiting_depth · oldest_age · in_flight · completed/min · failed/min · dlq_depth · p99_latency
Framework quick reference
Celery (Python) — crash-safe defaults:
app.conf.update(
task_acks_late=True, # ack after success (crash-safe)
task_reject_on_worker_lost=True, # requeue if worker dies
worker_prefetch_multiplier=1, # fair dispatch for long tasks
task_time_limit=300, task_soft_time_limit=240, # hard + catchable soft limit
broker_transport_options={"visibility_timeout": 3600}, # ⚠ tune to > p99 runtime
)
# Composition: chain (sequential), group (parallel), chord (parallel + callback)
BullMQ (Node) — successor to bull; prefer it for new work. Worker/Queue/QueueEvents split; FlowProducer for parent/child DAGs; built-in rate limiter. Watch lockDuration vs job length (stalled-job re-run).
Sidekiq (Ruby) — at-least-once, idempotent handlers mandatory. OSS basic fetch loses in-flight jobs on hard kill; Pro super_fetch recovers them. sidekiq_options retry:, dead:; death_handlers for DLQ hooks.
SQS — managed, durable, native DLQ via RedrivePolicy. Standard = at-least-once/unordered; FIFO = ordered + dedup. Tune visibility timeout to job length; 256KB payload cap → claim-check.
Anti-patterns
Non-idempotent handler on an at-least-once queue → duplicate charges/emails/rows. The default and most common prod incident.
Acking before the work is done → silent job loss on crash.
Unbounded retries / no DLQ → poison pills clog the queue forever.
No overlap lock on cron → same scheduled run fires on every replica.
Local-time schedules → DST double/skip.
Big payloads in the queue → broker bloat, 256KB SQS rejections; use claim-check.
Shared queue for fast + slow work → head-of-line blocking; separate by workload class.
git add-ing the whole .work — n/a here, but analogously: don't retain unbounded completed/failed sets; they exhaust Redis.
Storing job state only in worker memory → lost on restart; persist checkpoints.
Checklists
Before shipping a job handler:
Handler is idempotent — dedup key derived from business identity, side effect gated by a unique constraint or SETNX, marker TTL > retry window + retention
Ack/commit happens after the side effect (late ack), not before
Errors classified: transient → retry, deterministic → fail fast to DLQ
Backoff is exponential with full jitter, capped delay, bounded attempts
