QForge — Job Queue & Workflow Orchestration Engine

A from-scratch background job system — a mini Celery/Sidekiq — built on Postgres: concurrent-safe job claiming, retries with exponential backoff, dead-letter handling, heartbeat-based recovery of crashed workers, and multi-step DAG workflows — all visible on a live dashboard.

A worker pool draining the queue, retry/backoff and dead-lettering, a DAG workflow, and a killed worker's job auto-recovered. (full-quality video)

The problem this is built around

Every backend eventually needs to run work outside the request path — send the email, process the upload, run the nightly report. Reaching for Celery/Sidekiq hides the two questions that make this a genuine distributed-systems problem, and that this project answers directly:

1. How do two workers avoid grabbing the same job? The classic "double processing" bug. Naive SELECT a pending job; UPDATE it to running has a race: two workers read the same row before either writes. The fix is SELECT ... FOR UPDATE SKIP LOCKED — each worker locks the row it's claiming, and other workers skip locked rows to the next available job. No double-claims, no blocking, no external lock service.
2. How does a job survive a worker crashing mid-execution? A claimed job would otherwise be stuck in running forever. Each worker heartbeats while it works; a monitor requeues any job whose heartbeat goes stale, so another worker finishes it.

On top of that: retries with exponential backoff, a dead-letter state after max retries, job priorities and delayed scheduling, and DAG workflows where a step runs only once its dependencies succeed.

Architecture

  enqueue job / submit DAG ──►  jobs table (Postgres)
                                 status, priority, scheduled_at, retry_count,
                                 max_retries, payload (jsonb), heartbeat_at,
                                 workflow_id, step_name, depends_on[]
                                          │
       ┌──────────────────────────────────┼──────────────────────────────────┐
       ▼                                   ▼                                   ▼
  Worker pool 1                       Worker pool 2                       Monitor
  claim via SELECT ... FOR            claims a DIFFERENT job —            requeues jobs
  UPDATE SKIP LOCKED,                 SKIP LOCKED guarantees no           whose heartbeat
  run handler, heartbeat              two workers share a row             went stale
       │                                                                  (crashed worker)
   ┌───┴────┐
   ▼        ▼
 success   failure
 mark      retry_count++, exponential backoff, requeue —
 done,     or dead-letter after max_retries.
 advance   On success, unblock DAG steps whose deps are met.
 DAG

The hardest decisions

1. FOR UPDATE SKIP LOCKED — the whole ballgame

The claim is a single atomic statement:

UPDATE jobs SET status='running', locked_by=$w, heartbeat_at=now(), attempts=attempts+1
WHERE id = (
  SELECT id FROM jobs
  WHERE status='pending' AND scheduled_at <= now() AND queue = ANY($queues)
  ORDER BY priority DESC, scheduled_at
  FOR UPDATE SKIP LOCKED           -- ← the line that makes concurrency correct
  LIMIT 1
) RETURNING *;

The inner SELECT locks exactly one claimable row and skips any already locked by another worker. The test suite proves it: 10 threads hammer a queue of 60 jobs and every job is claimed exactly once — no losses, no duplicates. Using Postgres for this (instead of a separate broker) means the queue is transactional with your data and needs no extra infrastructure.

2. Heartbeats vs. "is the worker alive?"

You can't reliably ask "is that worker still running?" across a network. So instead of asking, workers assert liveness: while a job runs, its worker updates heartbeat_at every couple of seconds. The monitor treats any running job with a stale heartbeat as orphaned and requeues it. A crash requeue counts as a failed attempt, so a job that repeatedly kills workers eventually dead-letters instead of looping forever.

3. Retries, backoff, and dead-letter

A failed attempt increments retry_count and reschedules with delay = min(base·2^n, cap) — exponential backoff so a flapping dependency isn't hammered. Past max_retries the job moves to the dead state (dead-letter) for inspection instead of retrying forever.

4. DAG workflows without a workflow engine

A workflow is just jobs that carry workflow_id, step_name, and depends_on[]. Steps with no dependencies start pending; the rest start blocked. When a step succeeds, the store unblocks any step whose dependencies are now all satisfied; when the last step finishes, the workflow is marked succeeded. If a step dead-letters, the remaining steps are cancelled and the workflow fails. Simple state on rows, no separate orchestrator.

Verifying it yourself

Tests run against a real Postgres (the guarantees are all about concurrency and SQL, so mocks would prove nothing):

python scripts/migrate.py
python -m pytest -q

They cover: concurrent claiming (10 threads, 60 jobs, each claimed exactly once), priority and delayed-schedule ordering, retry→backoff→dead-letter, DAG unblocking in dependency order and the failure-cancel cascade, and crash recovery (a stale-heartbeat job is requeued and completed by another worker).

Running locally

Requires Python 3.11+ and PostgreSQL (brew install postgresql@16 && brew services start postgresql@16).

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
python scripts/migrate.py
python scripts/devrun.py        # API + monitor + 2 worker pools
open http://localhost:8090      # dashboard
python scripts/demo.py          # enqueue a spread of jobs + a DAG workflow

See crash-recovery live: enqueue a long job (curl -s localhost:8090/jobs -H 'content-type: application/json' -d '{"task":"slow","payload":{"seconds":30}}'), find the worker that claimed it, kill -9 it, and watch the monitor requeue the job to the other worker within STALE_SECONDS.

API

# enqueue a job
curl -s localhost:8090/jobs -H 'content-type: application/json' \
  -d '{"task":"add","payload":{"a":2,"b":3},"priority":5,"max_retries":3}'

# submit a DAG workflow (deploy runs only after test AND lint, which run after build)
curl -s localhost:8090/workflows -H 'content-type: application/json' -d '{
  "name":"ci","steps":[
    {"step":"build","task":"echo"},
    {"step":"test","task":"echo","depends_on":["build"]},
    {"step":"lint","task":"echo","depends_on":["build"]},
    {"step":"deploy","task":"echo","depends_on":["test","lint"]}]}'

What I'd change at 10× scale

• LISTEN/NOTIFY instead of polling so idle workers wake instantly on new jobs rather than polling every 500 ms.
• Partition or archive the jobs table by day; move terminal jobs to a history table so the hot claimable set stays small.
• A separate scheduler for cron-style recurring jobs (the schema already supports delayed jobs via scheduled_at).
• Prometheus metrics (claim latency, queue depth, retry rate) and alerting on dead-letter growth — pair it with the observability pipeline (Project 6).
• Priority-fairness guards so a flood of high-priority jobs can't starve a queue.

License

MIT