<!--
Author: Simon-Pierre Boucher
Contact: contact@spboucher.ai
-->

# 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.