# Patterns — Background Jobs

# Contents

Idempotency record
Backoff with full jitter
Transactional enqueue (enqueue after commit)
Chunked long-running job
Worker coordination with SKIP LOCKED
Distributed lock for scheduled jobs
Observability queries
Gotchas

# Idempotency record

python

# One row per logical operation; the INSERT is the guard.
def send_invoice_email(invoice_id: int):
    with db.transaction():
        inserted = db.execute(
            """INSERT INTO job_runs (job_key)
               VALUES (%s) ON CONFLICT (job_key) DO NOTHING""",
            (f"invoice-email:{invoice_id}",),
        ).rowcount
        if not inserted:
            return  # duplicate delivery — already done or in progress
        invoice = db.fetch_invoice(invoice_id)  # refetch fresh state
        mailer.send(invoice.email, render(invoice))

# Backoff with full jitter

python

import random

BASE_S = 30      # first retry ~30 s: transient blips resolve in seconds
FACTOR = 2
MAX_ATTEMPTS = 5 # ~30s..8min window; beyond that it's an outage, use the DLQ

def next_delay(attempt: int) -> float:
    return random.uniform(0, BASE_S * FACTOR ** attempt)  # full jitter

python

# Celery equivalent
@app.task(bind=True, max_retries=5, retry_backoff=30,
          retry_backoff_max=600, retry_jitter=True,
          autoretry_for=(TransientError,), time_limit=120)
def sync_account(self, account_id): ...

# Transactional enqueue (enqueue after commit)

python

# ❌ enqueue inside the transaction: worker may run before COMMIT
# ✅ enqueue on commit
with db.transaction() as tx:
    order_id = create_order(tx)
    tx.on_commit(lambda: queue.enqueue(process_order, order_id=order_id))

If the queue library has no on-commit hook, write the job to an outbox-style table in the same transaction and let a relay enqueue it.

# Chunked long-running job

python

PAGE = 500  # one page ≈ seconds of work: cheap to retry, no timeout risk

def reindex_products(cursor: int = 0):
    rows = db.fetch("SELECT id FROM products WHERE id > %s ORDER BY id LIMIT %s",
                    (cursor, PAGE))
    for r in rows:
        index(r.id)
    if len(rows) == PAGE:
        queue.enqueue(reindex_products, cursor=rows[-1].id)  # resume point

# Worker coordination with SKIP LOCKED

sql

-- Homemade queue table: safe concurrent pickup, no double-claim
UPDATE jobs SET state = 'running', locked_at = now()
WHERE id = (
  SELECT id FROM jobs
  WHERE state = 'pending' AND run_at <= now()
  ORDER BY run_at
  FOR UPDATE SKIP LOCKED
  LIMIT 1
)
RETURNING *;

# Distributed lock for scheduled jobs

python

# Prevents double-fire when two schedulers overlap.
def run_nightly_report():
    got = redis.set("lock:nightly-report:2026-08-05", worker_id,
                    nx=True, ex=3600)  # ex ≈ expected runtime + margin
    if not got:
        return
    ...

# Observability queries

python

# Emit per job type
metrics.timing(f"job.{name}.duration_ms", elapsed)
metrics.incr(f"job.{name}.{'ok' if success else 'failed'}")

Alert on: DLQ size > 0 (page), oldest pending message age > 5× expected latency (warn), failure rate > 5% over 10 min (warn).

# Gotchas

retry_jitter off by default in several libraries — synchronized retries stampede the dependency that just recovered.
Visibility timeout < job timeout (SQS-style queues) → the message reappears while the first worker still runs it; keep visibility ≥ job timeout + margin.
Enqueue-then-crash before commit → job references data that never existed; see transactional enqueue above.
Serialized enums/dataclasses break old in-flight jobs on deploy; IDs-only payloads (rule 2) sidestep the whole class of errors.
DLQ replays must go through the same idempotency guard — a replay is just one more duplicate delivery.
SELECT ... FOR UPDATE without SKIP LOCKED serializes all workers on one row: a queue with one effective consumer.