spb/ultra-sharp-agent-skills Public

1<!--2Author: Simon-Pierre Boucher3Contact: contact@spboucher.ai4-->56# Patterns — Background Jobs78## Contents9- Idempotency record10- Backoff with full jitter11- Transactional enqueue (enqueue after commit)12- Chunked long-running job13- Worker coordination with SKIP LOCKED14- Distributed lock for scheduled jobs15- Observability queries16- Gotchas1718## Idempotency record1920```python21# One row per logical operation; the INSERT is the guard.22def send_invoice_email(invoice_id: int):23    with db.transaction():24        inserted = db.execute(25            """INSERT INTO job_runs (job_key)26               VALUES (%s) ON CONFLICT (job_key) DO NOTHING""",27            (f"invoice-email:{invoice_id}",),28        ).rowcount29        if not inserted:30            return  # duplicate delivery — already done or in progress31        invoice = db.fetch_invoice(invoice_id)  # refetch fresh state32        mailer.send(invoice.email, render(invoice))33```3435## Backoff with full jitter3637```python38import random3940BASE_S = 30      # first retry ~30 s: transient blips resolve in seconds41FACTOR = 242MAX_ATTEMPTS = 5 # ~30s..8min window; beyond that it's an outage, use the DLQ4344def next_delay(attempt: int) -> float:45    return random.uniform(0, BASE_S * FACTOR ** attempt)  # full jitter46```4748```python49# Celery equivalent50@app.task(bind=True, max_retries=5, retry_backoff=30,51          retry_backoff_max=600, retry_jitter=True,52          autoretry_for=(TransientError,), time_limit=120)53def sync_account(self, account_id): ...54```5556## Transactional enqueue (enqueue after commit)5758```python59# ❌ enqueue inside the transaction: worker may run before COMMIT60# ✅ enqueue on commit61with db.transaction() as tx:62    order_id = create_order(tx)63    tx.on_commit(lambda: queue.enqueue(process_order, order_id=order_id))64```6566If the queue library has no on-commit hook, write the job to an67`outbox`-style table in the same transaction and let a relay enqueue it.6869## Chunked long-running job7071```python72PAGE = 500  # one page ≈ seconds of work: cheap to retry, no timeout risk7374def reindex_products(cursor: int = 0):75    rows = db.fetch("SELECT id FROM products WHERE id > %s ORDER BY id LIMIT %s",76                    (cursor, PAGE))77    for r in rows:78        index(r.id)79    if len(rows) == PAGE:80        queue.enqueue(reindex_products, cursor=rows[-1].id)  # resume point81```8283## Worker coordination with SKIP LOCKED8485```sql86-- Homemade queue table: safe concurrent pickup, no double-claim87UPDATE jobs SET state = 'running', locked_at = now()88WHERE id = (89  SELECT id FROM jobs90  WHERE state = 'pending' AND run_at <= now()91  ORDER BY run_at92  FOR UPDATE SKIP LOCKED93  LIMIT 194)95RETURNING *;96```9798## Distributed lock for scheduled jobs99100```python101# Prevents double-fire when two schedulers overlap.102def run_nightly_report():103    got = redis.set("lock:nightly-report:2026-08-05", worker_id,104                    nx=True, ex=3600)  # ex ≈ expected runtime + margin105    if not got:106        return107    ...108```109110## Observability queries111112```python113# Emit per job type114metrics.timing(f"job.{name}.duration_ms", elapsed)115metrics.incr(f"job.{name}.{'ok' if success else 'failed'}")116```117118Alert on: DLQ size > 0 (page), oldest pending message age > 5× expected119latency (warn), failure rate > 5% over 10 min (warn).120121## Gotchas122123- **`retry_jitter` off by default** in several libraries — synchronized124  retries stampede the dependency that just recovered.125- **Visibility timeout < job timeout** (SQS-style queues) → the message126  reappears while the first worker still runs it; keep visibility ≥ job127  timeout + margin.128- **Enqueue-then-crash before commit** → job references data that never129  existed; see transactional enqueue above.130- **Serialized enums/dataclasses** break old in-flight jobs on deploy;131  IDs-only payloads (rule 2) sidestep the whole class of errors.132- **DLQ replays must go through the same idempotency guard** — a replay is133  just one more duplicate delivery.134- **`SELECT ... FOR UPDATE` without `SKIP LOCKED`** serializes all workers135  on one row: a queue with one effective consumer.136