# 🔁 Queue-Based Workers: Task Model, Backpressure, Rate Budget

By Lê Nguyễn Tuân

## From a Single Script to a Worker Service

By now the capstone runs as one process: it reads feed responses, normalizes media records, streams image bytes with chunked downloads and retries, assembles video fragments, hashes the result and writes it to the storage layout. That process is correct but fragile. One slow host stalls every other download, and a crash in the middle loses all in-flight work.

This lesson separates that pipeline into two roles that talk through a persistent queue, then adds the two controls every long-running crawler needs: backpressure and a per-source request budget.

After this lesson you can:

- Describe a durable task record and the states a task moves through.
- Split the capstone into a producer (discovery) and workers (network plus storage).
- Apply backpressure so a fast producer never buries slow downloads.
- Enforce a shared token-bucket budget for each source host.
- Size concurrency from measured latency instead of guessing.

```mermaid
flowchart TD
  Feed["📡<br/>Feed API"] --> Prod["🧭<br/>Producer"]
  Prod --> Q[("🗂️<br/>Persistent Queue")]
  Q --> W1["⚙️<br/>Worker 1"]
  Q --> W2["⚙️<br/>Worker 2"]
  W1 --> Store["💾<br/>Media Store"]
  W2 --> Store
  W1 --> Budget["🎟️<br/>Rate Budget"]
  W2 --> Budget
  Budget --> Hosts["🌐<br/>Source Hosts"]
```

## Producer, Worker, and the Queue Between Them

The producer is a small, fast loop. It fetches feed pages, normalizes records with the schema you built in Module 1, and enqueues tasks. It never downloads media bytes.

Workers are identical processes. Each one leases a task, performs the HTTP work, verifies the result (hash check from Module 2), writes to storage, then acknowledges the task. Because workers are interchangeable, scaling out means starting another process, nothing else.

The queue must be persistent: Redis Streams, a managed queue service, or a Postgres table read with `SELECT ... FOR UPDATE SKIP LOCKED`. A task is acknowledged only after the work is durable on disk. That single rule is what makes the capstone restartable.

## The Task Record: One Row, One Unit of Work

Keep the task small and self-describing, because a worker that receives it has no other context.

| Field | Purpose |
| --- | --- |
| `id` | Stable task identifier |
| `kind` | `image`, `video_manifest`, `video_fragment`, `verify` |
| `source` | `facebook`, `weibo`, `bilibili` |
| `payload` | JSON: URL, headers, referer, resolution, parent id |
| `attempts`, `max_attempts` | Retry control |
| `not_before` | Earliest eligible run time |
| `lease_until` | Visibility timeout for the current worker |
| `state` | `pending`, `leased`, `done`, `failed`, `dead` |

Two details matter more than the rest. First, `lease_until` is how a crashed worker gets recovered: when the lease expires the task becomes visible again and another worker takes it. Second, a large video is never one giant task. A `video_manifest` task fans out into `video_fragment` tasks, and a `verify` task joins them.

**Practice 3.1.** A video manifest lists 40 fragments. Sketch the task fan-out and say where the join happens.
**Check the answer**

Enqueue one `video_manifest` task. The worker that leases it parses the fragment list and enqueues 40 `video_fragment` tasks, each carrying the parent manifest id in `payload`. When the last fragment finishes, enqueue the `verify` task, which concatenates, hashes and compares against the expected size. Keeping the parent id makes that join possible, and lets you resume a half-finished video after a restart.

## Backpressure: Keeping the Producer Honest

A producer that paginates quickly can enqueue thousands of tasks per minute while workers manage a few dozen downloads. The queue grows without bound, memory and disk fill, and every task you enqueued is now stale.

Backpressure is the fix, and it is two-sided:

- Bound the queue. When depth crosses a high-water mark, the producer sleeps or polls a slower schedule. Resume below a low-water mark.
- Bound the worker. Each worker runs a fixed number of concurrent downloads (an `asyncio.Semaphore` or a thread pool) so it never opens more sockets than the host tolerates.

Size that concurrency with Little's law: in-flight requests ≈ target throughput × measured latency. If a thumbnail takes 400 ms and you want 5 per second, you need about 2 concurrent requests, not 20.

**Practice 3.2.** Queue depth climbs steadily while worker CPU stays near 5 percent. What are the two most likely causes?
**Check the answer**

Workers are blocked on network latency (per-source concurrency set too low), or the rate budget is exhausted and workers spend their time waiting for tokens. Check the token wait time metric first, then raise per-source concurrency only if the budget still has headroom.

## The Rate Budget: One Allowance per Source

Concurrency limits protect your own machine. The rate budget protects the relationship with each source host and keeps you inside whatever quota you are permitted to use. Because Facebook, Weibo and Bilibili behave differently, you keep one bucket per source.

A token bucket has two parameters: refill rate `r` (requests per second) and burst `b` (how many can go at once). Every HTTP request costs one token, including retries and manifest fetches. With multiple workers the bucket must live in shared storage, for example an atomic decrement in Redis with a refill timestamp, otherwise N workers each think they own the full budget.

A producer only _____ and never downloads media bytes itself.
Each task row carries an _____ counter and a _____ timestamp so a crashed worker's work can be reclaimed.
Backpressure means the producer _____ when queue depth crosses a high-water mark.
A per-source _____ caps requests per second across all workers because the budget is shared.
On HTTP 429 or repeated 5xx you should _____ instead of retrying immediately.

Treat a `429` as a signal, not an accident: halve that source's rate, add jittered exponential backoff, and let the rate recover slowly after a clean window.

**Practice 3.3.** Three workers each allow 4 requests per second. The Weibo budget allows 5 per second. What breaks and how do you fix it without changing worker code?
**Check the answer**

Local limits multiply to 12 requests per second, which overruns the shared budget. Move the limit out of worker memory into a shared bucket with `r = 5` and a small burst, and have each worker acquire a token before every request. Workers keep their own concurrency cap, but the source rate is now global.

## Recall Drill: Queue Terms and Limits

Front: What are the two roles in a queue-based crawler?
Back: A producer that discovers and enqueues tasks, and workers that do the network and storage work.

Front: What must the producer never do?
Back: Download media bytes; it only normalizes feed records and enqueues tasks.

Front: Why must the queue be persistent?
Back: Queued tasks and their state must survive process restarts and crashes.

Front: What does lease_until protect against?
Back: A worker dying mid-task: when the lease expires the task becomes visible and is reclaimed.

Front: Which task kinds does the capstone define?
Back: image, video_manifest, video_fragment and verify.

Front: How is a multi-fragment video modeled as tasks?
Back: One manifest task that fans out into per-fragment tasks, then a verify task joins them.

Front: Define backpressure in one line.
Back: Workers signal capacity and the producer slows down when queue depth or lag crosses a threshold.

Front: How does Little's law size worker concurrency?
Back: In-flight requests approximately equal throughput multiplied by latency, so concurrency follows measured latency.

Front: Why one token bucket per source?
Back: Each platform has different limits, and one global bucket either wastes capacity or overruns the strictest source.

Front: Where does the rate budget live when several workers run?
Back: In shared storage such as an atomic Redis counter, not in each process's memory.

Front: Correct response to HTTP 429 or repeated 5xx?
Back: Jittered exponential backoff plus a temporary reduction of that source's rate.

Front: What happens when attempts exceeds max_attempts?
Back: The task moves to a dead state for inspection instead of retrying forever.

Front: What does the not_before field buy you?
Back: Scheduled retries and polite spacing without holding a worker busy.

## Field Notes from Running Workers

- Acknowledge after the write, never before. An early ack turns a crash into silent data loss.
- Set the lease longer than your slowest plausible download, but short enough that recovery is quick. Ninety seconds to five minutes is a sane start.
- Do not let retries bypass the token bucket. A retry storm is still traffic.
- Log the token wait time next to each request. If waits dominate request time, the budget, not the code, is your bottleneck.
- Keep the producer dumb. Any logic that needs a downloaded byte belongs in a worker.
- The most common mistake: one global rate limit for all three sources, which starves the fast source and still overruns the slow one.

## What to Carry Forward

The capstone is now two roles: a producer that only enqueues, and workers that lease, download, verify and acknowledge, all governed by a shared per-source budget. Remember the four anchors: persistent queue, small self-describing tasks, backpressure on both sides, and one token bucket per source.

Next, we make those tasks safe to run twice: canonical URLs, cursors and dedupe, which is what keeps a queue-based crawler from downloading the same video again after a restart.

---

## Contents

Part of [🎞️ Media Crawler Architecture: Queue, Storage and Scaling](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling.md)

- **🧭 Foundations and Capstone Skeleton**
  - [🔍 How Facebook, Weibo and Bilibili Serve Media](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/media-serving-anatomy-three-platforms.md)
  - [🧱 Capstone Blueprint: Config, Storage Layout, Pipeline Modules](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/capstone-blueprint-config-storage-pipeline.md) (sign-in required)
  - [🧩 Capstone Step 1: Normalize Media Records from Feed Responses](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/capstone-normalize-media-records.md)
  - [✅ Checkpoint: Platform Anatomy and Project Skeleton](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/checkpoint-platform-anatomy-skeleton.md)
- **📥 Media Download Engineering**
  - [🌊 Streaming Image Downloads: Chunks, Resume, Retries, Timeouts](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/streaming-image-downloads-resume-retries.md)
  - [🎥 Video Downloads: Manifests, Fragments, Resolution Selection](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/video-downloads-manifests-fragments-resolution.md)
  - [🗄️ Capstone Step 2: Storage, Content Hashing and Integrity Checks](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/capstone-storage-hashing-integrity.md)
  - [✅ Checkpoint: Download Engineering](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/checkpoint-download-engineering.md)
- **🏗️ Architecture, Scheduling and Scaling**
  - [🔁 Queue-Based Workers: Task Model, Backpressure, Rate Budget](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/queue-workers-task-model-backpressure.md) — this lesson
  - [🚀 Capstone Step 3: Observability, Recovery and Scaling Out](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/capstone-observability-recovery-scaling.md) (premium content)
  - [✅ Checkpoint: Architecture and Scaling](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/checkpoint-architecture-scaling.md) (premium content)
- [Course cheatsheet](https://everwise.study/courses/media-crawler-architecture-queue-storage-and-scaling/cheatsheet.md) (premium content)