Loom Background Jobs

Background Jobs

Overview

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

1. At-least-once is the default. Design every handler to be idempotent. Redis-backed queues (Sidekiq, BullMQ, Celery+Redis), SQS standard, and a queue retrying after a crash can deliver work more than once. FIFO producer deduplication does not make a consumer's external side effect exactly once. The achievable goal is exactly-once effect = durable idempotency at the sink plus retry-safe handling.
2. 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 make the durable sink reject duplicates. A separate “done” flag cannot atomically cover an external side effect and a crash.

def handle(job):
    key = job["idempotency_key"]          # e.g. f"receipt:{order_id}"
    # One transaction: UNIQUE(key) makes duplicate deliveries a successful no-op.
    with db.transaction():
        inserted = db.execute(
            "INSERT INTO receipts (idempotency_key, order_id) VALUES (?, ?) "
            "ON CONFLICT (idempotency_key) DO NOTHING",
            [key, job["order_id"]],
        ).rowcount == 1
        if inserted:
            db.execute(
                "INSERT INTO outbox (idempotency_key, kind, aggregate_id) "
                "VALUES (?, 'send-receipt', ?)",
                [key, job["order_id"]],
            )
    # The outbox makes the intent durable; the email sender must deduplicate by key too.

• ⚠ Do not set a “done” marker before the effect: a crash after the claim loses work. Do not mark it only after a non-idempotent external effect either: a crash can repeat that effect. Prefer an idempotency facility at the external provider; otherwise persist an intent/outbox transactionally and make the downstream consumer idempotent.
• If a provider or store uses a processed-marker TTL, keep it longer than the max retry window plus queue retention, or a late redelivery can re-run the effect.
• 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.

Delivery guarantees & broker selection

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:

1. Keep handlers well under T. Size the timeout to p99 duration × safety factor.
2. 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).
3. 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 jitter rand(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 acks
class Base(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)

// BullMQ — exponential backoff, bounded attempts, retention
new Queue("email", { defaultJobOptions: {
  attempts: 5,
  backoff: { type: "exponential", delay: 2000 }, // BullMQ jitters internally
  removeOnComplete: 1000,     // cap completed set or Redis grows unbounded
  removeOnFail: 5000,
}});

# Sidekiq — default 25 retries (~21 days) w/ built-in jittered backoff
sidekiq_options queue: :default, retry: 10, dead: true
sidekiq_retries_exhausted { |msg, ex| DeadJobNotifier.notify(msg, ex) }

Dead letter queues & poison pills

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.

// Minimal DLQ hand-off (BullMQ): route exhausted jobs, keep forensics
worker.on("failed", async (job, err) => {
  if (job.attemptsMade >= job.opts.attempts!) {
    await dlq.add("dead", {
      original: job.data, error: err.message,
      attempts: job.attemptsMade, failedAt: Date.now(),
    });
    metrics.increment("dlq.parked", { queue: job.queueName });
  }
});

Idempotent enqueue (producer-side dedup)

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 supports messages up to 1 MiB; verify the configured queue/quota, and Redis consumes memory) are for coordination, not bulk data.

{ "job": "transcode", "s3_key": "uploads/abc.mov", "sha256": "…", "size": 734003200 }

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
}

await queue.add("cleanup", {}, { repeat: { pattern: "0 * * * *" }, jobId: "cleanup-hourly" });

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 () => {
  await Promise.all(workers.map(w => w.pause()));      // stop pulling new jobs
  await Promise.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)
def train(self, run_id, resume_from=None):
    state = load_ckpt(resume_from) if resume_from else fresh_state()
    for epoch in range(state.epoch, TOTAL):
        step(state)
        if epoch % CKPT_EVERY == 0:
            save_ckpt(run_id, epoch, state)     # durable → survives crash/redeliver
            self.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 + producer dedup. Tune visibility timeout to job length; messages are up to 1 MiB (verify queue/quota) → use claim-check for larger payloads.

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.
• Fixed / un-jittered backoff → synchronized retry storms.
• Retrying non-transient errors → wasted capacity, delayed DLQ, masked bugs.
• 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 or SQS size-limit rejections (up to 1 MiB; verify queue/quota); use claim-check.
• Shared queue for fast + slow work → head-of-line blocking; separate by workload class.
• git add-ing the whole .loom/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
• Max-attempts → DLQ with payload + error + attempt count; DLQ depth alerts
• Job runtime < visibility timeout / lock duration, OR heartbeat/extend, OR checkpoint & resume
• Large payloads passed by reference (claim-check), not inline
• Producer-side dedup where retries can double-enqueue (jobId / FIFO dedup id)

Before shipping a scheduled job:

• Overlap prevention: leader lock / SET NX PX so only one runs across replicas
• Schedule stored/evaluated in UTC; DST verified
• Missed-run policy chosen (catch-up vs skip) and implemented — no naive replay of every missed tick
• "Skip if previous still running" guard for jobs that can exceed their interval

Before shipping worker infra:

• Concurrency sized to workload (CPU≈cores/processes; IO≫cores/async), against downstream limits
• SIGTERM drains in-flight jobs; orchestrator grace period ≥ longest job
• Separate queues/workers per workload class; no fast+slow mixing
• Priority scheme can't starve low-priority (weighting/aging); per-tenant fairness if multi-tenant
• Monitoring: waiting depth, oldest-message age, in-flight, throughput, error rate, DLQ depth, p99 latency
• Backpressure/autoscale keyed on depth or age; retained job sets bounded
• Workers recycled to bound memory leaks (max_tasks_per_child / max_memory_per_child)