event-driven-task-queue

A distributed, at-least-once task queue built on Redis Streams and PostgreSQL, with an auto-scaling worker pool, crash recovery, and a real end-to-end workload (an API-triggered stock report pipeline) proving the system under real conditions rather than simulated jobs.

Live demo: event-driven-task-queue-production.up.railway.app (API) · event-driven-task-queue-chi.vercel.app (frontend)

---

What this project demonstrates

This is the second in a series of three job-processing systems built at increasing sophistication, each demonstrating a different set of primitives:

1. postgres-job-queue — job dispatch using only database primitives: SELECT FOR UPDATE SKIP LOCKED for lock-free concurrent claiming, LISTEN/NOTIFY for push-based wakeup, window functions for analytics. No external broker.
2. This project — introduces an external broker (Redis Streams) specifically for delivery guarantees LISTEN/NOTIFY cannot provide, plus a worker pool that scales itself based on live queue depth rather than running a fixed number of workers.
3. kafka-clickstream-pipeline — a different broker (Kafka) chosen specifically to demonstrate partition-based ordering guarantees at a different scale point than Redis Streams offers.

The three are not "each one replaces the last" — they demonstrate different tradeoffs in message broker design, and the postgres-only version remains genuinely instructive for what you can build without any external dependency at all.

---

Architecture

                     ┌──────────────┐
   POST /reports ──▶ │   FastAPI    │
                     │   (api.py)   │
                     └──────┬───────┘
                            │ enqueue_job()
                            ▼
                  ┌──────────────────┐        ┌──────────────┐
                  │  Redis Stream    │◀──────▶│  Postgres    │
                  │     "jobs"       │        │  jobs table  │
                  └────────┬─────────┘        └──────▲───────┘
                           │ XREADGROUP                │
              ┌────────────┼────────────┐              │
              ▼            ▼            ▼              │
        ┌──────────┐ ┌──────────┐ ┌──────────┐         │
        │ Worker 1 │ │ Worker 2 │ │ Worker N │─────────┘
        └──────────┘ └──────────┘ └──────────┘
              ▲            ▲            ▲
              └────────────┴────────────┘
                           │
                   ┌───────┴────────┐
                   │    Monitor      │  watches stream lag + PEL size,
                   │  (monitor.py)   │  spawns/retires worker & reaper
                   └───────┬────────┘  processes via multiprocessing
                           │
                   ┌───────▼────────┐
                   │     Reaper      │  reclaims jobs abandoned by
                   │   (reaper.py)   │  crashed workers via XCLAIM
                   └────────────────┘

Components:

• producer.py — enqueue_job(payload): writes to Postgres and Redis in one logical operation, returns the Redis stream entry ID.
• worker.py — consume_jobs(): claims one job via XREADGROUP, marks it processing in Postgres, dispatches to the appropriate handler by task_type, writes the final result, and acknowledges the Redis message only after the Postgres write succeeds.
• reaper.py — cleanup_jobs(): scans the Pending Entries List for jobs whose idle time exceeds a threshold (meaning the worker that claimed them died before finishing), reclaims them via XCLAIM, and reprocesses them.
• monitor.py — the control plane. Polls Redis for stream lag (undelivered message count) and pending-entry count every 5 seconds, and spawns or lets idle worker/reaper processes exit to match the current load, using Python's multiprocessing.Process.
• stock.py — the demonstration workload. A pure function, generate_stock_report(ticker, email), that fetches market data, computes technical indicators, trains a lightweight classifier, generates a PDF, and emails it. Contains no database or queue logic — see Demo workload below for why that separation matters.
• api.py — FastAPI layer exposing POST /reports (enqueue, with a same-day cache check) and GET /reports/{id} (poll status).
• db.py — lazily-initialized psycopg2 connection pool, shared by every process that imports it.

---

Why Redis Streams instead of a Redis List or Pub/Sub

This is a deliberate choice worth being explicit about, since Redis offers three different primitives that could superficially serve as a queue:

Redis List (LPUSH/BRPOP) — a simple FIFO queue. Once a worker pops an item, it is gone; if that worker crashes mid-processing, the item is lost permanently. No concept of acknowledgment, no way to detect or recover an in-flight failure.

Redis Pub/Sub — fire-and-forget broadcast. A subscriber only receives messages published while it is connected; there is no persistence, no replay, and no way for a worker that was briefly disconnected to catch up. Also, every subscriber receives every message (broadcast semantics), which is the wrong shape for "exactly one worker processes each job."

Redis Streams — an append-only log, closer in spirit to Kafka than to a List. Entries are retained after being read. Consumer groups (XREADGROUP) let multiple workers cooperatively drain a stream with each entry going to exactly one consumer in the group. Delivered-but-unacknowledged entries sit in a Pending Entries List (PEL) and can be inspected (XPENDING) and reclaimed (XCLAIM) by another consumer — this is the mechanism that makes crash recovery possible at all, and it does not exist in a List or Pub/Sub.

Streams were chosen specifically because the PEL and XCLAIM give at-least-once delivery with recoverability, which a List cannot provide (no redelivery concept) and Pub/Sub actively cannot provide (no persistence).

---

Delivery guarantee: at-least-once, and why not exactly-once

This system provides at-least-once delivery: a job is guaranteed to be processed at least one time, but under specific failure windows it may be processed more than once. It does not provide exactly-once delivery.

Where the "at least" comes from: a worker calls XACK only after the Postgres write for the job's final state has committed successfully. If the worker crashes between finishing the Postgres write and calling XACK, the entry remains unacknowledged in the PEL, and the reaper will eventually reclaim and reprocess it — even though it was, in fact, already completed. This is the correct and unavoidable tradeoff of at-least-once systems: the alternative (acknowledging before the write, for "at-most-once") risks silently losing work if the process dies in that window, which is strictly worse for a job queue.

Why exactly-once wasn't attempted: true exactly-once delivery across two independent systems (Redis and Postgres) would require a distributed transaction or a two-phase commit protocol spanning both stores, which neither Redis Streams nor this architecture is designed to provide. Kafka has an exactly-once processing mode (transactional producers + idempotent writes within Kafka itself), but even that only guarantees exactly-once within Kafka's own state, not across an arbitrary external side effect like a Postgres write or, in this project's case, sending an email. In practice, systems that need exactly-once effects achieve it by combining at-least-once delivery with idempotent processing at the consumer — which is exactly what this project does instead.

Idempotency is implemented per-operation, not by the broker. The queue guarantees delivery; it says nothing about what "safe to process twice" means for a given job, because that meaning is entirely domain-specific. For the demo workload:

UPDATE jobs
SET status = 'processing', worker_id = %s, started_at = NOW(), attempts = attempts + 1
WHERE stream_id = %s AND status = 'pending';

This is a conditional claim: if rowcount == 0, another worker (or an earlier attempt by this same worker after a partial crash) already claimed or completed the job, so this attempt is a no-op — the message is simply acknowledged and dropped. This UPDATE ... WHERE status = 'pending' pattern is the queue's idempotency guard; a payment-processing job type, in contrast, would need a different guard entirely (an idempotency key checked against a charges table before calling a payment gateway), which is precisely why idempotency belongs in the handler, not in the transport layer.

---

Why the worker claims before executing, in two separate transactions

# Transaction 1: claim
UPDATE jobs SET status='processing', ... WHERE stream_id=%s AND status='pending';
conn.commit()   # lock released here

# --- job executes here, outside any open transaction ---

# Transaction 2: finalize
UPDATE jobs SET status=%s, completed_at=NOW(), error=%s WHERE stream_id=%s;
conn.commit()

If claiming and finalizing were a single transaction wrapped around the job's execution, the row (and, transitively, the Postgres connection) would be held for the entire duration of the work — which for the stock-report workload includes an external network call to Yahoo Finance, model training, PDF rendering, and an external call to Resend's email API. With SimpleConnectionPool(1, 10, ...), ten workers each holding a connection open for several seconds of unrelated I/O would exhaust the pool under any real concurrency. Splitting into two short transactions means each worker holds a Postgres connection only for the few milliseconds needed to run an UPDATE, and the pool serves many more workers than its configured maximum would otherwise support.

---

The monitor: why a custom autoscaler instead of a fixed worker pool

monitor.py is a control loop, not a worker itself. Every ADJUST_INTERVAL (5 seconds) it:

1. Reads stream lag (XINFO GROUPS → the lag field: entries in the stream not yet delivered to any consumer) and computes a target worker count: max(MIN_WORKERS, min(MAX_WORKERS, lag // JOBS_PER_WORKER)).
2. Reads pending count (XPENDING summary form) and computes a target reaper count using the same formula.
3. Spawns multiprocessing.Process instances up to the target, using p.is_alive() to prune any that have already exited.

Why multiprocessing.Process and not threading.Thread: the stock-report workload runs sklearn's LogisticRegression.fit(), which is CPU-bound. CPython's Global Interpreter Lock means only one thread executes Python bytecode at any instant regardless of thread count, so CPU-bound work does not parallelize across threads — it would parallelize across processes, since each process gets its own interpreter and its own GIL. Threads would have been sufficient for a purely I/O-bound workload (which is what the original placeholder time.sleep() jobs were), but since the real workload does real computation, processes are the correct choice.

Why workers self-terminate rather than being killed by the monitor: each worker calls consume_jobs(max_idle_polls=3) — after three consecutive empty polls, it returns and the process exits naturally. The monitor's next scaling pass sees p.is_alive() == False, prunes it from its tracked list, and (if load has genuinely dropped) simply doesn't replace it. This avoids the monitor ever needing to signal a worker mid-job — a worker only ever stops itself, between jobs, which means there is no scenario where a scale-down event interrupts in-progress work.

Known limitation — thundering herd on wakeup. When monitor.py spawns N workers simultaneously (or when Redis has a burst of new entries), every worker independently calls XREADGROUP, and only one can win the race for any given entry — the rest get an empty read and loop back to poll again. This wastes some CPU and generates redundant Redis round-trips under bursty load. It doesn't affect correctness (Redis's group-read semantics guarantee each entry goes to exactly one consumer regardless of how many ask at once) — it's a throughput inefficiency, not a bug. At this project's scale it's negligible; at production scale the standard fix is a shared multiprocessing.Queue or a semaphore that only wakes one idle worker per new entry rather than broadcasting to all of them.

---

Demo workload: Stock Report Generator

To exercise the pipeline with real work rather than time.sleep(), one task type (stock_report) triggers a genuine, non-trivial pipeline: fetch historical price data for a ticker via yfinance, compute a standard technical-analysis feature set (moving averages, MACD family, RSI-14, Bollinger Bands with %B and bandwidth, on-balance volume, stochastic oscillator, rolling returns, a rolling trend slope), train a LogisticRegression classifier on-the-fly against those features to predict next-session direction, render a multi-page PDF report with embedded matplotlib charts, and email it via Resend's API.

This is a demonstration workload, not a financial product. The prediction is produced by a simple classical model trained on a handful of thousand rows per request — it is not intended to produce reliable trading signals and carries an explicit disclaimer in both the PDF and the email body. Its purpose in this project is to give the queue a real workload with real failure modes (a bad ticker symbol, a network timeout against Yahoo Finance, an email API failure) rather than a workload that always succeeds after a fixed delay.

Why stock.py never opens a database connection. generate_stock_report() is a pure function: given a ticker and an email, it returns a result dict (status, predicted_direction, confidence, pdf_path, data, error) and has no knowledge that a queue or a database exists. worker.py is the only place that persists anything. This separation means the entire stock-report pipeline is independently testable and callable outside the queue context (see quick_test.py), and it keeps the queue's job-lifecycle logic (claim/execute/finalize/ack) in exactly one place rather than duplicated per task type. It also keeps the queue infrastructure generic rather than tied to stock reports: adding a new job type only requires implementing a new handler function and wiring it into the worker's task_type dispatch, while the queueing, retry, acknowledgment, and persistence logic remains unchanged.

Result caching. api.py checks for an existing {TICKER}_report_{date}.pdf before enqueuing — if today's report for that ticker already exists, it's emailed immediately with no queue round-trip. Report generation writes to a temp file and calls os.replace() to move it into its final path atomically, so a concurrent request that hits the cache mid-write never reads a partially-written file (the file only exists at its final path once it's completely written). monitor.py sweeps /tmp on every scaling pass and deletes any report file not stamped with today's date.

---

API

Method	Path	Description
POST	/reports	Body: {"ticker": str, "email": str}. Returns {"status": "queued", "stream_id": ...} or, on a same-day cache hit, {"status": "completed", "cached": true, "data": {...}} with the email sent immediately.
GET	/reports/{stream_id}	Poll job status by Redis stream ID. Returns {"status": ..., "error": ...}, plus ticker and data_url once completed.
GET	/reports/{ticker}/data	Returns the full computed indicator payload (all sections, plus a ~1-year history series) for a ticker/date, used by the frontend to render charts without re-fetching the PDF.
GET	/health	Liveness check.

---

Running locally

git clone https://github.com/SiddhantGupta3112/event-driven-task-queue.git
cd event-driven-task-queue
cp .env.example .env   # fill in RESEND_API_KEY at minimum
docker compose up

This starts Postgres (schema auto-applied on first boot via the mounted schema.sql), Redis, and the app container, which runs both the FastAPI server and the monitor loop as sibling processes inside one container (start.sh backgrounds uvicorn, then runs monitor.py in the foreground so the container's lifetime tracks the monitor's).

Enqueue a job directly:

curl -X POST http://localhost:8000/reports \
  -H "Content-Type: application/json" \
  -d '{"ticker": "AAPL", "email": "you@example.com"}'

Watch multiple workers scale up under load and confirm no job is double-processed:

docker compose exec postgres psql -U admin -d job_queue -c \
  "SELECT worker_id, status, COUNT(*) FROM jobs GROUP BY worker_id, status ORDER BY worker_id;"

---

Deployment

Railway (current live deployment) — Postgres and Redis provisioned as managed Railway services; the app container reads their connection details from Railway's auto-injected environment variables (PGHOST, DATABASE_URL, REDISHOST, REDIS_PASSWORD) rather than the local POSTGRES_HOST/REDIS_HOST names, switched via an IS_LOCAL flag read in both db.py and producer.py.

Terraform / Azure — terraform/ contains a complete, correct Terraform configuration targeting Azure: managed PostgreSQL Flexible Server, a Redis container as a sidecar inside an Azure Container Instance group (Azure's managed Redis Cache and Azure Container Registry were both attempted first but rejected by the specific student subscription's region policy — see below), and the container group itself. Deployment was blocked, not by a code or configuration error, but by an Azure for Students subscription policy that rejected resource creation (RequestDisallowedByAzure) for every tested resource type — Redis Cache, Container Registry, Container Instances, and PostgreSQL Flexible Server — across every region tried, including the account's home region. This was confirmed to be a subscription-level restriction, not a Terraform or region-selection error, by reproducing the same rejection directly via the Azure CLI outside of Terraform entirely. The Terraform code itself is complete and correct and would provision successfully on a standard (non-student) subscription; it's included in this repo as evidence of that work and is a talking point in its own right about diagnosing infrastructure failures that are policy-level rather than configuration-level.

---

Known limitations

• No authentication on Redis or Postgres in local/default configuration — acceptable for local development, not for a public deployment without additional hardening (Redis requirepass, network-level isolation). Railway's managed services provide network isolation and credentialed access by default, which mitigates this for the live deployment specifically.
• Thundering herd on worker wakeup — see The monitor above.
• Result caching keys only on ticker + calendar date, not time-of-day — a report requested at market open and one requested at market close on the same day will return the same cached data, which is correct given yfinance's data granularity (daily bars) but worth knowing as an explicit constraint.
• Resend's sandbox sender (onboarding@resend.dev) can only deliver to the account owner's verified email until a custom domain is verified — the live demo's email delivery is therefore limited to that one address; ticker/report generation itself is unrestricted.
• Email delivery via Resend, not the earlier Gmail SMTP approach — Gmail SMTP with app passwords was tried first and rejected in favor of a purpose-built transactional email API, which is also the more realistic production pattern (real applications generally do not send mail through a personal Gmail account's SMTP credentials).