Distributed Task Queue System with Webhook Delivery

A production-grade, asynchronous task queue where clients submit jobs via a REST API, dedicated workers execute them reliably, and results are delivered to client endpoints via signed webhooks.

Tech Stack

Concern	Technology
API	FastAPI (async)
Queue	RabbitMQ via aio-pika
Cache / State	Redis via redis-py async
Database	MSSQL via SQLAlchemy 2 async
Migrations	Alembic
HTTP Client	httpx (async)
Logging	loguru with correlation IDs
Container	Docker + Docker Compose

Architecture

                  ┌──────────────────────────────┐
   Client ──────► │           FastAPI             │  POST /jobs
                  │  validate -> INSERT (PENDING) │  GET  /jobs/{id}
                  │  publish → jobs_exchange      │  POST /webhooks
                  └──────────────┬───────────────┘
                                 │  {job_id, job_type, payload}
                                 ▼
                        ┌─────────────────┐
                        │   jobs_exchange  │ (direct, durable)
                        └────────┬────────┘
                                 │ job.new
                                 ▼
                        ┌─────────────────┐
                        │    jobs_queue    │◄──── prefetch_count=1
                        └────────┬────────┘
                                 │
                    ┌────────────▼────────────┐
                    │     Job Worker (x2)     │  ack LAST after all work
                    │  1. Redis lock (TTL 5m) │
                    │  2. status → RUNNING    │
                    │  3. handler.handle()    │
                    └───┬──────────────┬──────┘
                        │ success      │ failure
                        ▼              ▼
             update MSSQL + Redis   incr attempts
                        │            wait = 2^attempt + jitter
                        │            republish to jobs_queue
                        ▼
        ┌─────────────────────────────┐
        │     webhook_exchange        │  exhausted → nack → DLX → DLQ
        └─────────────┬───────────────┘
                      │ webhook.deliver
                      ▼
        ┌─────────────────────────────┐
        │  Webhook Worker (x2)        │
        │  HMAC-SHA256 sign + POST    │
        │  record delivery_logs row   │
        │  retry w/ backoff on fail   │
        └─────────────────────────────┘

Folder Structure

app/
├── main.py                  FastAPI app, correlation-ID middleware, worker spawn
├── api/
│   ├── routes/{jobs,webhooks}.py
│   └── dependencies.py
├── workers/
│   ├── job_worker.py        consumes jobs_queue, executes handlers
│   └── webhook_worker.py    consumes webhook_queue, delivers signed webhooks
├── services/
│   ├── job_service.py       job CRUD + delivery-log reads
│   └── webhook_service.py   webhook registration + delivery logging
├── handlers/
│   ├── base_handler.py      abstract interface
│   ├── document_handler.py
│   └── notification_handler.py
├── models/{job,webhook}.py  SQLAlchemy async models
├── schemas/{job,webhook}.py Pydantic request/response models
├── core/
│   ├── config.py            pydantic-settings (reads .env)
│   ├── database.py          async engine + session factory
│   ├── redis.py             async Redis client
│   └── rabbitmq.py          topology + robust publishers
└── utils/
    ├── logger.py            loguru config + correlation-ID context
    ├── hmac.py              webhook signature helpers
    └── retry.py             exponential backoff + jitter

Data Flow

1. Client calls POST /jobs with {job_type, payload}.
2. FastAPI validates the job type, inserts a row to MSSQL with status PENDING, then publishes {job_id, job_type, payload} to jobs_exchange → jobs_queue.
3. Job Worker receives the full payload from the message (no DB read).
4. Worker tries to acquire the Redis lock lock:job:{job_id}; if another worker already holds it, the message is acknowledged and skipped.
5. Lock acquired → Redis status set to RUNNING → handler executes.
6. Success: Redis result cached, Redis/MSSQL status set to SUCCESS, and a webhook.deliver message is published to webhook_exchange.
7. Failure: Redis attempts:job:{job_id} is incremented; if below MAX_RETRIES the job is republished with backoff; otherwise the message is nacked (requeue=False) and dead-lettered to jobs_dlq.
8. Webhook Worker receives the completion event, looks up registered webhooks, and POSTs the job payload + result to each URL signed with HMAC-SHA256. Every attempt is written to delivery_logs.

RabbitMQ Topology

jobs_exchange (direct, durable)
  ├── jobs_queue          routing key: job.new
  │     ├── x-dead-letter-exchange: jobs_dlx
  │     └── x-dead-letter-routing-key: job.dead
  └── jobs_delay_queue    routing key: job.retry  (delayed retries via DLX)
jobs_dlx (direct, durable)
  └── jobs_dlq            routing key: job.dead
webhook_exchange (direct, durable)
  └── webhook_queue       routing key: webhook.deliver

Redis Keys

Key	Purpose	TTL
lock:job:{job_id}	Distributed worker lock	300s
status:job:{job_id}	Live job status	3600s
result:job:{job_id}	Cached job result	3600s
attempts:job:{job_id}	Job retry counter	3600s
attempts:webhook:{id}	Webhook retry counter	3600s
lock:webhook:{id}	Webhook delivery lock	300s

Job Status Flow

PENDING → RUNNING → SUCCESS
                  → FAILED → RETRYING → EXHAUSTED → DLQ

Retry Logic

wait = (2 ** attempt) + random.uniform(0, 1)

attempt 1 → ~2s
attempt 2 → ~4s
attempt 3 → ~8s
attempt 4 → ~16s
attempt 5 → EXHAUSTED → DLQ

API Endpoints

Method	Path	Description
POST	/jobs	Submit a new job (job_type, payload)
GET	/jobs/{id}	Get job status/result
POST	/webhooks	Register a webhook URL for a job
DELETE	/webhooks/{id}	Remove a webhook registration
GET	/jobs/{id}/logs	Delivery attempt history
GET	/health	Health check

Example — submit a job:

curl -X POST http://localhost:8000/jobs \
  -H 'Content-Type: application/json' \
  -d '{"job_type":"document","payload":{"template":"invoice","pages":3}}'

Every response includes an X-Request-ID header; pass it on subsequent calls to trace a job through the whole pipeline.

Webhook Signature Verification

Deliveries are signed with HMAC-SHA256 using the webhook's secret_key (returned once at registration). Headers:

• X-Task-Queue-Signature — hex(hmac_sha256(secret_key, raw_body))
• X-Task-Queue-Timestamp — epoch seconds

Receiver verification example:

import hashlib, hmac

def verify(secret_key: str, raw_body: bytes, signature: str) -> bool:
    expected = hmac.new(secret_key.encode(), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

Running with Docker Compose

cp .env.example .env        # adjust credentials if needed
docker-compose up --build   # starts mssql, redis, rabbitmq, api, workers

Services (with health checks on all of them):

Service	Port	Notes
api	8000	FastAPI
job_worker	—	Consumes jobs_queue
webhook_worker	—	Delivers signed webhooks
rabbitmq	5672	Management UI on 15672 (guest/guest)
redis	6379
mssql	1433

Run migrations (from the API container):

docker-compose exec api alembic upgrade head

Running Locally (without Docker)

Point the infrastructure hosts at localhost and run the API and workers:

python app/main.py                # API + spawns both workers
python -m app.workers.job_worker
python -m app.workers.webhook_worker

Architectural Decisions

1. Stateless workers. Workers keep no in-memory state between messages; all state lives in Redis (live) and MSSQL (final). This makes workers horizontally scalable — just add more replicas.
2. Full payload in the RabbitMQ message. Workers never read the DB to fetch a job, eliminating a read bottleneck and a source of inconsistency.
3. Redis = live state, MSSQL = final state. Redis gives fast status lookups and distributed locking; MSSQL is the durable source of truth for the audit trail, results, and delivery logs.
4. Hybrid at-least-once + explicit retries. Job failures are republished with exponential backoff + jitter (avoiding thundering-herd sync retries). Message acking happens last, after all work is done, so a crash never loses a message.
5. Dead-letter queue. After MAX_RETRIES attempts a job is nacked with requeue=False, and the DLX routes it to jobs_dlq for manual inspection or replay — nothing is silently dropped.
6. Distributed locking. Redis locks (TTL 300s) prevent two replicas from executing the same job concurrently. Locks are released after processing, and expired automatically on worker crash.
7. Signed webhooks. Every delivery carries an HMAC-SHA256 signature so clients can verify authenticity and integrity; a timestamp header enables replay-window checks.
8. Delivery logging. Each webhook attempt is persisted to delivery_logs (attempt number, HTTP status, timestamp, error) and exposed via GET /jobs/{id}/logs.
9. prefetch_count=1. Fair dispatch — a worker holds at most one message at a time, so a slow job never starves other consumers.
10. Correlation IDs. Loguru records are tagged with an X-Request-ID (propagated into workers as the job_id) so every log line across API, job worker, and webhook worker can be traced to a single request.
11. Durable exchanges/queues + persistent messages. Nothing survives a broker restart unless explicitly marked durable.
12. Health checks everywhere. Compose waits for RabbitMQ/Redis/MSSQL to be healthy before starting dependent services, and liveness checks run periodically on every container.
13. Configuration via environment. No hardcoded values; every knob (timeouts, retries, broker URLs) comes from .env through pydantic-settings.

Development Roadmap

Phase	Scope	Status
1	Foundation: compose, Dockerfile, /health	Done
2	Data layer: async models, Alembic migration	Done
3	API layer: jobs + webhooks endpoints	Done
4	Queue layer: publish jobs to RabbitMQ	Done
5	Worker layer: job execution + state updates	Done
6	Resilience: locks, backoff retries, DLX/DLQ	Done
7	Webhook layer: signed delivery + logging + retries	Done
8	Polish: correlation IDs, health checks, Postman, README	Done

Tests

pip install -r requirements-dev.txt
pytest tests/