# FastAPI production checklist **Version:** 2.2 **Last updated:** 2026-08-05 **Site:** https://fastapi-production-checklist.grok.me **Creator:** [Amin Sharifi](http://moaminsharifi.com/) **Scope:** Production FastAPI services with durable background jobs **Broker policy:** RabbitMQ and/or Kafka for durable work. Redis is not the primary job broker. Use this as a pre-ship gate and ops review. Pair it with the guides under `docs/guides/` for pseudocode and examples. ## Why this exists FastAPI makes it easy to ship an API that works on a laptop. Production breaks on the boring edges: secrets in images, blocking calls in `async` routes, emails that only run in `BackgroundTasks`, and workers that share a container with the API. This checklist turns those failure modes into checkable items. The interactive site tracks progress in your browser. No account required. **Primary keywords:** FastAPI production checklist, FastAPI background jobs, FastAPI Celery RabbitMQ, FastAPI Kafka workers, event-driven FastAPI, DDD FastAPI. ## Contents 1. [Development](#1-development) 2. [Security](#2-security) 3. [Performance and scaling](#3-performance-and-scaling) 4. [Testing, errors, and observability](#4-testing-errors-and-observability) 5. [Containerization](#5-containerization) 6. [Deployment and CI/CD](#6-deployment-and-cicd) 7. [Maintenance](#7-maintenance) 8. [Background jobs and workers](#8-background-jobs-and-workers) 9. [RabbitMQ](#9-rabbitmq) 10. [Kafka](#10-kafka-if-used) 11. [Worker containers and deploy](#11-worker-containers-and-deploy) 12. [Anti-patterns gate](#12-anti-patterns-gate) 13. [Architecture selection guide](#13-architecture-selection-guide) 14. [Pre-production gate](#14-pre-production-gate) 15. [Reference layouts](#15-reference-layouts) 16. [Guides and references](#16-guides-and-references) 17. [FAQ](#17-faq) 18. [Events and DDD](#18-events-and-ddd) --- ## 1. Development ### Project structure - [ ] Domain-oriented layout (e.g. `auth/`, `users/`, `orders/`) each with `router`, `schemas`, `models`, `service`, `dependencies`, `exceptions` - [ ] Or modular **bounded contexts** (`contexts/billing/...`) when using DDD (see section 18) - [ ] Shared cross-cutting code isolated (`core/`, `db/`, `middleware/`, `workers/`) - [ ] Routers mounted via `APIRouter` with clear prefixes and tags - [ ] Separate Pydantic schemas (API I/O) from ORM models (persistence) - [ ] Background job code lives in a dedicated package, not inside route handlers ### Configuration (Pydantic Settings) - [ ] Use `pydantic-settings` `BaseSettings`. Do not hardcode secrets - [ ] Split settings by domain when large (`DatabaseSettings`, `AuthSettings`, `RabbitMQSettings`, `KafkaSettings`) - [ ] Load from env / secret store; optional `.env` only for local (never committed) - [ ] Broker URLs, queue/topic names, timeouts, and retry defaults come from settings ### Dependency injection - [ ] DB sessions, current user, settings, and clients via `Depends()` - [ ] Prefer async dependencies for I/O; avoid sync that blocks the event loop - [ ] Use `lifespan` for startup/shutdown (pools, broker connections) - [ ] Publish/enqueue clients injected as dependencies ### Code quality - [ ] Strict request/response models on every endpoint - [ ] No blocking calls inside `async def` routes - [ ] Offload long/retryable/business-critical work to the job queue - [ ] `BackgroundTasks` limited to non-critical, short, best-effort work only - [ ] Domain invariants not only in route validators (prefer aggregates / domain layer) **Guides:** [01 Async vs jobs](guides/01-async-vs-jobs.md) · [12 DDD](guides/12-ddd-with-fastapi.md) --- ## 2. Security - [ ] HTTPS only in production; HSTS at the edge - [ ] TLS for broker connections (AMQPS / Kafka TLS) in production - [ ] Explicit CORS allowlist; never `["*"]` with credentials in prod - [ ] AuthN/AuthZ via dependencies; protect or disable docs in production - [ ] Secrets only in env / vault, not images, logs, or git - [ ] Workers re-check authorization for sensitive side effects - [ ] Flower, RabbitMQ Management, Kafka UI: private network + auth + TLS --- ## 3. Performance and scaling - [ ] Request path non-blocking; heavy work enqueued - [ ] DB pool size fits `max_connections` across API and workers - [ ] Redis is cache/rate-limit only, not primary job broker - [ ] Heavy vs light work on separate queues/topics - [ ] Worker replicas scale independently of API --- ## 4. Testing, errors, and observability - [ ] `GET /health` and `GET /ready` - [ ] Structured JSON logs; propagate `job_id` / `event_id` into workers - [ ] Job metrics: enqueue, success %, retries, backlog, DLQ - [ ] Kafka consumer lag alerts when events are used - [ ] Tests for enqueue paths, handler idempotency, and event consumer dedupe - [ ] Domain unit tests for aggregate invariants when using DDD **Guide:** [10 Testing workers](guides/10-testing-workers.md) --- ## 5. Containerization - [ ] Multi-stage build; non-root; no secrets in the image - [ ] Same image for API and worker; different CMD - [ ] Reverse proxy terminates TLS --- ## 6. Deployment and CI/CD - [ ] CI: lint, typecheck, tests, security scan - [ ] Alembic as sole schema path; expand/contract for zero-downtime - [ ] Graceful shutdown for API and workers - [ ] Backward-compatible API **and** message/event schemas during rollout --- ## 7. Maintenance - [ ] Runbooks: rollback, scale workers, DLQ redrive, rotate broker credentials - [ ] Review event catalog and consumer owners quarterly --- ## 8. Background jobs and workers ### Architecture - [ ] Durable broker: RabbitMQ and/or Kafka for work that must survive deploys - [ ] Redis is not the primary job broker - [ ] API process ≠ worker process; same image, different CMD - [ ] Enqueue returns 202 + `job_id`; status from application DB - [ ] Commands vs events intentionally separated (see section 18) ### Job domain model - [ ] `jobs` table with status, type, attempts, timestamps, optional `idempotency_key` - [ ] States: `pending` → `running` → `succeeded` | `failed` - [ ] Small message payloads (IDs/refs) ### Outbox, inbox, task design - [ ] Critical paths: transactional outbox or insert + reconcile publish - [ ] Consumers: inbox/dedupe or natural unique keys - [ ] Idempotent handlers; finite retries; backoff; timeouts; DLQ - [ ] Error taxonomy: transient vs permanent vs unknown - [ ] JSON only; no pickle **Guides:** [03 Lifecycle](guides/03-job-lifecycle-and-schema.md) · [04 Outbox](guides/04-outbox-inbox-idempotency.md) · [09 Errors](guides/09-error-taxonomy-retries-dlq.md) ### Library selection | Need | Pick | |------|------| | Mature jobs, Beat, Flower | Celery + RabbitMQ | | Async-native FastAPI | Taskiq + RabbitMQ | | Simpler sync workers | Dramatiq + RabbitMQ | | Event/stream pipelines | FastStream + Kafka | | Jobs + domain events | Celery/Taskiq → RMQ + FastStream → Kafka | | Non-critical after-response | BackgroundTasks | --- ## 9. RabbitMQ - [ ] Durable queues + persistent messages for critical tasks - [ ] DLX and DLQ configured - [ ] Separate queues by workload (`jobs.default`, `jobs.high`, `jobs.heavy`, `jobs.io`) - [ ] Prefetch tuned; publisher confirms on critical enqueue - [ ] Alerts on queue depth, DLQ depth, consumer count --- ## 10. Kafka (if used) - [ ] Topics by domain; stable consumer groups; keys when order matters - [ ] Idempotent consumers; offset commit after side effects - [ ] Consumer lag as primary backlog SLO - [ ] Schema strategy (JSON Schema / Avro / Protobuf) - [ ] Prefer Kafka for multi-consumer domain events, not as a private single-worker job queue by default --- ## 11. Worker containers and deploy - [ ] Services: `api`, `worker`, `rabbitmq` and/or `kafka` - [ ] Graceful shutdown; scale on depth/lag - [ ] Beat not duplicated without leader election --- ## 12. Anti-patterns gate - [ ] ❌ Critical work only on `BackgroundTasks` - [ ] ❌ Redis-only job library while RMQ/Kafka is required - [ ] ❌ API and worker in one container without independent scale - [ ] ❌ Pickle serializer; multi-MB payloads; unlimited retries - [ ] ❌ Dual-write DB+broker with no outbox/reconciler - [ ] ❌ Broker as product source of truth for job status - [ ] ❌ Sync multi-service call chains for every side effect - [ ] ❌ Shared DB tables as the only integration between contexts - [ ] ❌ God events with entire aggregate graphs - [ ] ❌ Anemic domain with all rules only in FastAPI validators --- ## 13. Architecture selection guide | Bus | Use for | |-----|---------| | RabbitMQ | Commands, jobs, retries, DLQ | | Kafka | Domain/integration events, fan-out, replay | | Redis | Cache, rate limits, sessions | ```text Is the work business-critical or must survive deploys? NO → BackgroundTasks (keep tiny) YES → Command/job or domain event? COMMAND/JOB → RabbitMQ + Celery | Taskiq | Dramatiq DOMAIN EVENT → Kafka + FastStream (often after outbox) BOTH → RMQ for jobs + Kafka for events ``` **Guides:** [02 Brokers](guides/02-brokers-celery-redis-rabbitmq-kafka.md) · [11 Event-driven](guides/11-event-driven-system-design.md) --- ## 14. Pre-production gate 1. Settings via Pydantic Settings; secrets not in code 2. HTTPS + CORS + security headers 3. Durable RabbitMQ and/or Kafka for critical async work 4. Separate API vs worker deployables 5. Jobs table + 202 + `job_id`; idempotent consumers 6. Outbox or reconciler on critical dual-write paths 7. If multi-module: event catalog + bounded contexts documented 8. Backlog SLO (depth and/or lag); admin UIs private 9. JSON messages only; TLS to brokers --- ## 15. Reference layouts ### Modular monolith with DDD-ish contexts ```text app/ main.py core/settings.py contexts/ billing/ domain/ application/ adapters/api|persistence|messaging notifications/ ... workers/ # entrypoints calling application use cases docker-compose.yml docs/guides/ ``` ### Compose ```text api, worker, rabbitmq and/or kafka, db optional: beat, flower, redis (cache only) ``` --- ## 16. Guides and references | Guide | Topic | |-------|--------| | [01 Async vs jobs](guides/01-async-vs-jobs.md) | `async def` vs BackgroundTasks vs queue | | [02 Brokers](guides/02-brokers-celery-redis-rabbitmq-kafka.md) | Celery, Redis, RMQ, Kafka roles | | [03 Job lifecycle](guides/03-job-lifecycle-and-schema.md) | States, schema, 202 contract | | [04 Outbox and inbox](guides/04-outbox-inbox-idempotency.md) | Dual-write safety | | [05 Celery + RMQ](guides/05-celery-rabbitmq-example.md) | Pseudocode example | | [06 Taskiq + RMQ](guides/06-taskiq-rabbitmq-example.md) | Async example | | [07 Dramatiq + RMQ](guides/07-dramatiq-rabbitmq-example.md) | Actor example | | [08 FastStream + Kafka](guides/08-faststream-kafka-example.md) | Events example | | [09 Errors and DLQ](guides/09-error-taxonomy-retries-dlq.md) | Retry policy | | [10 Testing](guides/10-testing-workers.md) | Test strategy | | [11 Event-driven design](guides/11-event-driven-system-design.md) | Commands, events, catalog, topology | | [12 DDD with FastAPI](guides/12-ddd-with-fastapi.md) | Contexts, aggregates, use cases | | [REFERENCES](guides/REFERENCES.md) | Official docs links | Public LLM mirrors: `/llms.txt`, `/llms-full.txt`. --- ## 17. FAQ ### What is a FastAPI production checklist? A structured list of requirements that separate a demo API from a service you can operate: security, config, health, deploy, durable jobs, and optional event/DDD structure. ### When should I use Celery with FastAPI? When you need mature job features and mostly sync workers. Pair with RabbitMQ under this standard. ### Is Redis enough for background jobs? Not for business-critical durable work here. Use RabbitMQ and/or Kafka. Redis stays for cache and rate limits. ### How do I design an event-driven system? Separate commands from domain events, commit state then publish via outbox, give each consumer its own models and inbox, and keep an event catalog. Full walkthrough: guide 11. ### How do I use DDD with FastAPI? Bounded contexts as packages, invariants on aggregates, thin routes and workers calling application use cases, integrate with events not shared tables. Full walkthrough: guide 12. ### Commands vs events? **Command:** please do X (usually one handler, often RabbitMQ). **Event:** X happened (many consumers, often Kafka). ### Who created this? [Amin Sharifi](http://moaminsharifi.com/). --- ## 18. Events and DDD ### Event-driven system - [ ] Distinguish **commands** (do work) from **domain events** (something happened) - [ ] Prefer RabbitMQ for commands/jobs; Kafka for multi-consumer domain/integration events - [ ] Publish domain events only after successful aggregate commit (transactional outbox) - [ ] Consumers own local models/projections; use inbox/dedupe - [ ] Living **event catalog**: name, producer, bus, key, schema, consumers, failure mode - [ ] Version event types (`invoice.paid.v1`); payloads are contracts, not ORM dumps - [ ] Message keys when per-aggregate ordering matters **Guide:** [11 Event-driven system design](guides/11-event-driven-system-design.md) ### Domain-driven design - [ ] Bounded contexts documented; no cross-context domain model imports - [ ] Ubiquitous language aligned across code, API, and event names - [ ] Aggregates enforce invariants; HTTP adapters stay thin - [ ] Application use cases orchestrate repository + domain + outbox - [ ] Workers execute the same application commands as the API where possible - [ ] Policies/process managers turn events into new commands when workflows span contexts - [ ] Contexts do not share tables for integration **Guide:** [12 DDD with FastAPI](guides/12-ddd-with-fastapi.md) ### Minimal adoption path ```text 1. Name 2–3 bounded contexts 2. One aggregate with real invariants 3. One command path (API → job queue) 4. One domain event via outbox 5. One consumer with inbox in another context 6. Event catalog entry for that event 7. Only then split deployables if needed ``` --- ## Document history | Version | Notes | |---------|--------| | 1.0 | Core FastAPI production checklist | | 2.0 | Background jobs; RabbitMQ/Kafka required | | 2.1 | Job lifecycle, outbox, guides, FAQ | | 2.2 | Event-driven design + DDD guides and checklist section | Keep this file in version control. Update it when the broker topology, worker stack, domain boundaries, or event catalog changes. ## 19. Locks and rate limits ### Job locks - [ ] Document resources that need exclusive access and their lock keys - [ ] Use the smallest lock key that protects the invariant - [ ] Distributed locks (Redis/DB), token-safe release, TTL or renewal - [ ] On lock busy: retry with delay/jitter; combine with idempotency ### API rate limits - [ ] Global, per-IP, and per-user limits; stricter on auth and enqueue - [ ] Shared store across API replicas; 429 + Retry-After ### Job / vendor rate limits - [ ] Rate-limit outbound vendor calls across all workers - [ ] Dedicated low-concurrency queues where appropriate - [ ] Metrics for lock contention and rate-limit requeues **Guide:** [16 Locks and rate limits](guides/16-locks-and-rate-limits.md) --- # Guides (full text) ## FILE: docs/guides/01-async-vs-jobs.md # Async code vs background jobs **Audience:** FastAPI engineers choosing where work should run. **Reading time:** ~8 min **Keywords:** FastAPI async, BackgroundTasks, durable jobs, Celery, Taskiq ## TL;DR | Need | Put the work here | |------|-------------------| | Data for **this** response | `async def` (or sync route if intentionally blocking) | | Tiny best-effort after response | FastAPI `BackgroundTasks` | | Must **retry, scale, or survive deploys** | Durable job queue (RabbitMQ) or event bus (Kafka) | `async` only means “don’t block the event loop while waiting on I/O.” It does **not** mean durable, retriable, or independently scalable. ## Contents 1. [Three layers people mix up](#1-three-layers-people-mix-up) 2. [Decision tree](#2-decision-tree) 3. [What belongs in the request](#3-what-belongs-in-the-request) 4. [BackgroundTasks rules](#4-backgroundtasks-rules) 5. [Durable jobs](#5-durable-jobs) 6. [Pseudocode patterns](#6-pseudocode-patterns) 7. [Performance pitfalls](#7-performance-pitfalls) 8. [Checklist mapping](#8-checklist-mapping) --- ## 1. Three layers people mix up | Layer | Runs where | Survives crash/deploy? | Scales how? | Use when | |-------|------------|------------------------|-------------|----------| | `async def` in a request | Same request, event loop | No | With API replicas | Fast I/O needed for the response | | FastAPI `BackgroundTasks` | After response, **same process** | No | Stuck on that API pod | Tiny, non-critical cleanup | | Job queue + worker | Separate process | Yes (durable broker) | Worker replicas / queues | Must retry, scale, or outlive deploys | | Domain event + consumer | Separate process | Yes | Consumer groups / lag | Many services react to a fact | ## 2. Decision tree ```text Must the HTTP response body include this work's result? YES → do it in the request path (async I/O), or enqueue + return 202 and let the client poll job status NO → Would users/business care if this work is lost on deploy? NO → BackgroundTasks is acceptable (keep it tiny) YES → Durable queue (command on RabbitMQ) or event (Kafka) ``` ### Examples | Work | Placement | |------|-----------| | Load user + permissions for GET | Request path | | Send “welcome” email after signup | Job queue | | Write analytics pixel after 200 | BackgroundTasks OK | | Generate 50MB export | Job queue (`jobs.heavy`) | | Charge card + ledger entry | Request for charge API **or** command job; never BackgroundTasks | | Fan-out “order placed” to billing + email | Domain event (Kafka) after commit | ## 3. What belongs in the request Keep request handlers **short and predictable**: - Validate input (Pydantic) - Authorize - One transaction (or a clear saga start) - Return a response If work is slow but the user must wait, still avoid holding the connection for minutes—prefer **202 + job_id** and a status endpoint. ### Async vs sync routes ```python # Prefer async for concurrent I/O (DB, HTTP clients) @app.get("/users/{user_id}") async def get_user(user_id: int, db: Db = Depends(get_db)): return await db.fetch_user(user_id) # Sync is OK if the whole stack is sync and you understand the thread pool @app.get("/report-sync") def report_sync(): return heavy_cpu_report() # better: offload to worker ``` **Rule:** never call blocking I/O inside `async def` without `asyncio.to_thread` / a proper executor. Prefer async drivers (`asyncpg`, `httpx.AsyncClient`). ## 4. BackgroundTasks rules Allowed: - Delete a temp file - Best-effort metrics that you already accepted losing - Non-critical cache warm Forbidden for production-critical work: - Emails users expect - Payment webhooks - PDF invoices - Anything with SLAs or legal weight ```python # PSEUDOCODE — OK: best-effort @app.post("/events") async def track(event: EventIn, bg: BackgroundTasks): bg.add_task(log_event_best_effort, event) return {"ok": True} ``` ## 5. Durable jobs When work must survive: 1. Insert **job row** (`pending`) in your DB 2. Publish **command** to RabbitMQ (or outbox → broker) 3. Return **202** + `job_id` 4. Worker runs, updates status, retries with bounds 5. Client polls `GET /jobs/{id}` or receives a webhook See [03 Job lifecycle](03-job-lifecycle-and-schema.md) and [04 Outbox](04-outbox-inbox-idempotency.md). ## 6. Pseudocode patterns ### Request-only async ```python # PSEUDOCODE @app.get("/users/{user_id}") async def get_user(user_id: int, db: Db = Depends(get_db)): user = await db.fetch_user(user_id) if not user: raise HTTPException(404) return user ``` ### Enqueue durable work (preferred) ```python # PSEUDOCODE @app.post("/exports", status_code=202) async def start_export(body: ExportIn, db: Db, bus: JobBus, user=Depends(auth)): async with db.transaction(): job = await db.insert_job( type="export_csv", status="pending", entity_id=str(user.id), idempotency_key=body.idempotency_key, payload_ref=body.filters_ref, ) await bus.publish( queue="jobs.heavy", message={"job_id": str(job.id), "type": "export_csv"}, ) return {"job_id": job.id, "status": "pending"} ``` ### Wrong: critical work only on BackgroundTasks ```python # PSEUDOCODE — DO NOT DO THIS for mail users expect @app.post("/signup") async def signup(body: SignupIn, bg: BackgroundTasks): user = await create_user(body) bg.add_task(send_welcome_email, user.id) # lost on deploy/crash return user ``` ## 7. Performance pitfalls | Pitfall | Symptom | Fix | |---------|---------|-----| | Sync HTTP inside `async def` | Latency spikes, worker stalls | Async client or thread offload | | Huge work in request | Timeouts, user abandons | 202 + job | | Unbounded BackgroundTasks | Memory growth under load | Cap work; move to queue | | CPU in async worker without process pool | Event loop lag | Separate heavy queue / prefork | ## 8. Checklist mapping - Development: no blocking in async routes; BackgroundTasks limited - Jobs: durable broker for critical work; 202 + job_id - Anti-patterns: critical email only on BackgroundTasks **Next:** [02 Brokers and frameworks](02-brokers-celery-redis-rabbitmq-kafka.md) ## FILE: docs/guides/02-brokers-celery-redis-rabbitmq-kafka.md # Brokers and frameworks: Celery, Redis, RabbitMQ, Kafka **Audience:** Architects and leads picking a worker stack. **Policy:** Durable jobs use **RabbitMQ and/or Kafka**. Redis is not the primary job broker. ## TL;DR | Piece | Role here | |-------|-----------| | **RabbitMQ** | Commands / jobs (acks, routing, DLX) | | **Kafka** | Domain events / streams (fan-out, lag, replay) | | **Redis** | Cache, rate limits, optional Celery **result** backend | | **Celery / Taskiq / Dramatiq** | Job frameworks on AMQP | | **FastStream** | Event consumers/producers on Kafka (or RMQ) | | **ARQ / RQ** | Redis queues — not system of record under this checklist | ## Contents 1. [Roles](#1-roles) 2. [Why Redis is demoted](#2-why-redis-is-demoted-for-durable-jobs) 3. [Stack matrix](#3-stack-matrix) 4. [Topology patterns](#4-topology-patterns) 5. [Framework comparison](#5-framework-comparison) 6. [Config checklist](#6-config-checklist) 7. [Anti-patterns](#7-anti-patterns) --- ## 1. Roles Keep **transport** separate from **product state**: - Broker moves messages. - Your **jobs** (or projections) table is what the product UI reads. - Flower / RMQ UI / Kafka UI are ops tools, not user-facing status. ## 2. Why Redis is demoted for durable jobs Redis lists are fast. They are a weak default when you need: - Durable queues across broker restarts with mature ops tooling - Dead-letter exchanges and redrive - Prefetch / QoS for mixed job sizes - Clear separation of job traffic from cache traffic **OK uses of Redis:** session cache, rate limits, feature flags, optional Celery result backend. **Not OK as only home of payment email / invoice PDF:** business-critical commands belong on RabbitMQ (or an explicit Kafka design with care). ## 3. Stack matrix | Need | Pick | |------|------| | Classic jobs, Beat, Flower, team knows Celery | **Celery + RabbitMQ** | | Async FastAPI + DI-friendly tasks | **Taskiq + RabbitMQ** | | Simple reliable sync actors | **Dramatiq + RabbitMQ** | | Domain events, fan-out, replay | **FastStream + Kafka** | | Jobs + events | RMQ workers **+** Kafka consumers | | After-response fluff only | BackgroundTasks | ### Decision tree ```text Business-critical or must survive deploys? NO → BackgroundTasks (tiny) YES → Is it a command (one worker does work) or an event (many react)? COMMAND → RabbitMQ + Celery | Taskiq | Dramatiq EVENT → Kafka + FastStream (often after transactional outbox) BOTH → RMQ for jobs + Kafka for events ``` ## 4. Topology patterns ### Jobs only ```text API → RabbitMQ (jobs.*) → workers → DB / email / S3 ``` ### Events only ```text API/service → outbox → Kafka topic → consumers (own DBs) ``` ### Combined (recommended for growing products) ```text API ├─ command → RabbitMQ → job worker (export, email) └─ outbox event → Kafka → notifications, analytics, ledger ``` ## 5. Framework comparison | | Celery | Taskiq | Dramatiq | FastStream | |--|--------|--------|----------|------------| | Primary model | Tasks | Async tasks | Actors | Streams | | RabbitMQ | Excellent | Yes | Yes | Yes | | Kafka | Limited/community | Yes (broker) | No (use other) | Excellent | | Beat / schedule | Mature | Via schedules | Periodics | App-level | | Flower-like UI | Flower | Custom / metrics | Custom | Metrics | | Best fit | Existing Celery shops | Async FastAPI | Simple RMQ workers | Event pipelines | Deep dives: [05 Celery](05-celery-rabbitmq-example.md) · [06 Taskiq](06-taskiq-rabbitmq-example.md) · [07 Dramatiq](07-dramatiq-rabbitmq-example.md) · [08 FastStream](08-faststream-kafka-example.md) ## 6. Config checklist - [ ] Broker URL from Pydantic Settings (not hardcoded) - [ ] TLS in production (`amqps://`, Kafka SSL) - [ ] Separate vhost / cluster per environment - [ ] JSON serializers only (no pickle) - [ ] Queue/topic names versioned or stable + documented - [ ] DLQ / error topic defined - [ ] Worker and API as **different** deployables, same image version ## 7. Anti-patterns - Redis-only queue while the org requires RMQ/Kafka - Pickle serializers - Treating the broker UI as the product status API - One mega-queue for 50ms and 50-minute jobs - API and worker in one container without independent scale **Next:** [03 Job lifecycle](03-job-lifecycle-and-schema.md) ## FILE: docs/guides/03-job-lifecycle-and-schema.md # Job lifecycle and schema **Audience:** Backend engineers implementing user-visible async work. **Rule:** The broker is transport. **Product state lives in your database.** ## TL;DR 1. Insert job row → publish command → return **202 + job_id** 2. Worker: claim → run → succeed / fail / retry 3. Client polls `GET /jobs/{id}` (or webhook) ## Contents 1. [State machine](#1-state-machine) 2. [Schema](#2-schema) 3. [Enqueue contract](#3-enqueue-contract) 4. [Worker contract](#4-worker-contract) 5. [Idempotency keys](#5-idempotency-keys) 6. [Payload design](#6-payload-design) 7. [Observability](#7-observability) 8. [API sketch](#8-api-sketch) --- ## 1. State machine ```text pending ──► running ──► succeeded │ ├──► failed (terminal after max attempts) ├──► cancelled (optional) └──► pending/retrying (scheduled retry; keep attempts++) ``` Keep statuses few. Map framework retries onto `attempts` + timestamps rather than inventing ten enums. ### Terminal vs non-terminal | Status | Terminal? | Client meaning | |--------|-----------|----------------| | pending | No | Queued or waiting for retry | | running | No | Worker claimed it | | succeeded | Yes | Done; result ready | | failed | Yes | Give up; show error | | cancelled | Yes | User or system aborted | ## 2. Schema ```sql -- PSEUDOCODE DDL CREATE TABLE jobs ( id UUID PRIMARY KEY, type TEXT NOT NULL, -- send_email | export_csv | ... status TEXT NOT NULL, -- pending|running|succeeded|failed|cancelled idempotency_key TEXT, -- UNIQUE when present payload_ref TEXT, -- small JSON or object-storage key result_ref TEXT, -- optional output location entity_id TEXT, -- user_id / order_id for listing attempts INT NOT NULL DEFAULT 0, max_attempts INT NOT NULL DEFAULT 5, last_error TEXT, created_at TIMESTAMPTZ NOT NULL, updated_at TIMESTAMPTZ NOT NULL, started_at TIMESTAMPTZ, finished_at TIMESTAMPTZ ); CREATE UNIQUE INDEX jobs_idempotency_uidx ON jobs (idempotency_key) WHERE idempotency_key IS NOT NULL; CREATE INDEX jobs_status_created_idx ON jobs (status, created_at); CREATE INDEX jobs_entity_idx ON jobs (entity_id, created_at DESC); ``` ## 3. Enqueue contract ```http POST /jobs/exports Idempotency-Key: client-generated-or-body → 202 Accepted { "job_id": "...", "status": "pending" } ``` ```python # PSEUDOCODE async def enqueue_export(cmd, db, bus): existing = await db.find_by_idempotency(cmd.idempotency_key) if existing: return existing # no double publish side effects async with db.transaction(): job = await db.insert_job(type="export_csv", status="pending", ...) await bus.publish("jobs.heavy", {"job_id": str(job.id)}) # or write outbox row instead of direct publish return job ``` ### Failure modes on enqueue | Failure | Response | Recovery | |---------|----------|----------| | Validation | 400 | Client fixes input | | Duplicate key | 202 with existing job | None | | Broker down, no outbox | 503 | Client retry with same key | | Broker down, with outbox | 202 | Relay publishes later | ## 4. Worker contract ```python # PSEUDOCODE async def handle_export(message): job_id = message["job_id"] job = await db.get_job(job_id) if job.status in ("succeeded", "cancelled"): return # idempotent ack claimed = await db.try_mark_running(job_id, expected=("pending",)) if not claimed: return try: result = await do_export(job) await db.mark_succeeded(job_id, result_ref=result) except TransientError as e: await db.bump_attempt(job_id, error=str(e)) raise # framework retries except PermanentError as e: await db.mark_failed(job_id, error=str(e)) # do not retry ``` ## 5. Idempotency keys - Client sends key for “create export” / “send receipt” - Server stores unique key on job - Replays return the **same** job_id Worker-side: natural keys (`invoice_id` + action) or inbox table — see [04 Outbox/inbox](04-outbox-inbox-idempotency.md). ## 6. Payload design | Do | Don't | |----|-------| | Pass IDs and small refs | Pass multi-MB blobs on the bus | | Version message schema | Dump ORM instances | | Keep secrets out of messages | Log full PII payloads | Large inputs/outputs → object storage; job row holds keys. ## 7. Observability Log and metric: - enqueue rate, success rate, attempts, runtime - time-to-start, time-to-complete SLOs - backlog (queue depth) Status for users always from **DB**, not Flower. ## 8. API sketch ```text POST /jobs/{type} 202 + job_id GET /jobs/{id} status + result_ref when ready GET /jobs?entity=… list for UI POST /jobs/{id}/cancel optional ``` **Next:** [04 Outbox and inbox](04-outbox-inbox-idempotency.md) ## FILE: docs/guides/04-outbox-inbox-idempotency.md # Outbox, inbox, and idempotency **Audience:** Engineers shipping reliable dual-writes (DB + broker). **Guarantee:** Brokers deliver **at least once**. Networks fail between commit and publish. ## TL;DR | Problem | Pattern | |---------|---------| | Commit then publish can drop the publish | **Transactional outbox** | | At-least-once delivery double-runs handlers | **Inbox / natural keys** | | Client double-clicks submit | **Idempotency-Key** on enqueue | ## Contents 1. [Dual-write problem](#1-dual-write-problem) 2. [Transactional outbox](#2-transactional-outbox) 3. [Relay design](#3-relay-design) 4. [Inbox (consumer dedupe)](#4-inbox-consumer-dedupe) 5. [Idempotent handlers](#5-idempotent-handlers) 6. [When outbox is optional](#6-when-outbox-is-optional) 7. [Pseudocode end-to-end](#7-pseudocode-end-to-end) 8. [Ops](#8-ops) --- ## 1. Dual-write problem ```text A) INSERT job OK → publish FAIL → stuck pending, client may retry B) publish OK → INSERT FAIL → worker has message, no job row ``` Exactly-once **across** DB and broker without an extra pattern is a myth for most stacks. Design for **at-least-once + idempotency**. ## 2. Transactional outbox Write business state and “message to send” in **one DB transaction**. ```sql -- PSEUDOCODE CREATE TABLE outbox ( id UUID PRIMARY KEY, aggregate_id TEXT NOT NULL, destination TEXT NOT NULL, -- queue or topic headers JSONB, payload JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL, published_at TIMESTAMPTZ, attempts INT NOT NULL DEFAULT 0 ); CREATE INDEX outbox_unpublished_idx ON outbox (created_at) WHERE published_at IS NULL; ``` ```python # PSEUDOCODE — API async with db.transaction(): job = await db.insert_job(...) await db.insert_outbox( destination="jobs.default", payload={"job_id": str(job.id), "type": job.type}, ) return job # 202 even if broker is down; relay catches up ``` ## 3. Relay design ```python # PSEUDOCODE — single-leader or SKIP LOCKED workers async def outbox_relay_tick(): rows = await db.fetch_unpublished(limit=100, for_update_skip_locked=True) for row in rows: try: await broker.publish(row.destination, row.payload, headers=row.headers) await db.mark_published(row.id) except TransientBrokerError: await db.bump_outbox_attempt(row.id) ``` Rules: - Prefer **one leader** or safe concurrent pollers with `SKIP LOCKED` - Metric: unpublished age p95 - Alert if outbox lag exceeds SLO ## 4. Inbox (consumer dedupe) ```sql CREATE TABLE inbox ( event_id TEXT PRIMARY KEY, processed_at TIMESTAMPTZ NOT NULL, consumer TEXT NOT NULL ); ``` ```python # PSEUDOCODE async def on_message(event): if await db.inbox_exists(event["event_id"], consumer="billing"): return # already done await apply_side_effect(event) await db.inbox_insert(event["event_id"], consumer="billing") ``` Order matters: for some side effects you need **insert inbox first** with a unique constraint and only then external I/O, or use a unique business key (`invoice_id` paid). ## 5. Idempotent handlers | Strategy | Example | |----------|---------| | Natural unique key | `UPDATE payments SET status='paid' WHERE id=? AND status='issued'` | | Inbox table | Store `event_id` | | Job status guard | Skip if `succeeded` | | Upsert projection | `INSERT … ON CONFLICT UPDATE` | Handlers must tolerate **replay** of the same message. ## 6. When outbox is optional Direct publish after insert can be OK if: - Job is non-critical, or - You have a **reconciler** that re-publishes `pending` jobs older than N seconds, and - Product accepts brief delay Still implement **idempotent** workers. ## 7. Pseudocode end-to-end ```text Client POST /pay → TX: mark invoice issued→paid intent + outbox event invoice.paid.v1 → 200/202 Relay → Kafka Notifications consumer → inbox → send email command → RabbitMQ Email worker → send → mark job succeeded ``` ## 8. Ops - Dashboards: outbox lag, inbox size growth, duplicate suppress rate - Redrive unpublished outbox after broker outage - Never delete outbox rows before retention policy (audit optional) **See also:** [11 Event-driven design](11-event-driven-system-design.md) · [09 Errors](09-error-taxonomy-retries-dlq.md) ## FILE: docs/guides/05-celery-rabbitmq-example.md # Example: Celery + RabbitMQ + FastAPI **Audience:** Teams standardizing on Celery for durable jobs. **Broker:** RabbitMQ (AMQP). Redis only as optional result backend. ## TL;DR layout ```text app/ core/settings.py api/routes/jobs.py workers/celery_app.py workers/tasks/email.py workers/tasks/exports.py Dockerfile # same image, different CMD docker-compose.yml # api, worker, rabbitmq, db, flower? ``` ## Contents 1. [Settings](#1-settings) 2. [Celery app](#2-celery-app) 3. [Task definition](#3-task-definition) 4. [FastAPI enqueue](#4-fastapi-enqueue) 5. [Queues and routing](#5-queues-and-routing) 6. [Retries and time limits](#6-retries-and-time-limits) 7. [Run commands](#7-run-commands) 8. [Ops notes](#8-ops-notes) --- ## 1. Settings ```python # PSEUDOCODE — pydantic-settings class Settings(BaseSettings): database_url: str celery_broker_url: str = "amqps://user:pass@rabbitmq:5671//" celery_result_backend: str | None = None # optional Redis/DB; status still in app DB environment: str = "local" model_config = SettingsConfigDict(env_file=".env", extra="ignore") ``` ## 2. Celery app ```python # PSEUDOCODE — app/workers/celery_app.py from celery import Celery from app.core.settings import settings celery_app = Celery("app", broker=settings.celery_broker_url) celery_app.conf.update( task_serializer="json", accept_content=["json"], result_serializer="json", task_acks_late=True, worker_prefetch_multiplier=1, # fair for long jobs task_default_queue="jobs.default", task_routes={ "app.workers.tasks.exports.*": {"queue": "jobs.heavy"}, "app.workers.tasks.email.*": {"queue": "jobs.io"}, }, broker_connection_retry_on_startup=True, ) celery_app.autodiscover_tasks(["app.workers.tasks"]) ``` **Never enable pickle.** ## 3. Task definition ```python # PSEUDOCODE — app/workers/tasks/email.py from app.workers.celery_app import celery_app from app.db import session_scope from app.services.mail import send_email @celery_app.task( bind=True, name="app.workers.tasks.email.send_receipt", max_retries=5, autoretry_for=(TransientMailError,), retry_backoff=True, retry_backoff_max=600, retry_jitter=True, soft_time_limit=30, time_limit=45, ) def send_receipt(self, job_id: str): with session_scope() as db: job = db.get_job(job_id) if job is None or job.status == "succeeded": return db.mark_running(job_id) try: send_email(job.payload_ref) # sync client OK in Celery prefork db.mark_succeeded(job_id) except PermanentMailError as e: db.mark_failed(job_id, str(e)) raise except TransientMailError as e: db.bump_attempt(job_id, str(e)) raise self.retry(exc=e) ``` ## 4. FastAPI enqueue ```python # PSEUDOCODE @router.post("/receipts", status_code=202) async def enqueue_receipt(body: ReceiptIn, db: Db = Depends()): job = await db.create_job(type="send_receipt", payload_ref=..., idempotency_key=body.key) send_receipt.delay(str(job.id)) # or apply_async(queue="jobs.io") return {"job_id": job.id, "status": "pending"} ``` Prefer **job_id only** on the wire; load details from DB in the worker. ## 5. Queues and routing | Queue | Work | |-------|------| | `jobs.default` | General | | `jobs.high` | User-visible latency | | `jobs.heavy` | CPU / large files | | `jobs.io` | Email / HTTP | | `jobs.dlq` | Dead letters (via DLX) | Separate worker deployments can subscribe to different queues. ## 6. Retries and time limits - `acks_late=True` so crash redelivers - Soft + hard time limits - Retry only **transient** errors - After max retries → mark failed + DLQ See [09 Error taxonomy](09-error-taxonomy-retries-dlq.md). ## 7. Run commands ```bash # API uvicorn app.main:app --host 0.0.0.0 --port 8000 # Worker celery -A app.workers.celery_app.celery_app worker -Q jobs.default,jobs.io -c 4 # Heavy pool celery -A app.workers.celery_app.celery_app worker -Q jobs.heavy -c 1 # Optional Beat (single leader) celery -A app.workers.celery_app.celery_app beat # Optional Flower (private + auth) celery -A app.workers.celery_app.celery_app flower ``` ## 8. Ops notes - Memory: recycle workers (`worker_max_tasks_per_child`) - Graceful shutdown: stop consuming, finish in-flight within termination grace - Metrics: task runtime, retries, queue depth via RabbitMQ - Flower is **not** the product job status UI **Related:** [06 Taskiq](06-taskiq-rabbitmq-example.md) if you want async-native workers. **Also:** rate limits + entity locks end-to-end — [17 Celery RL + lock pipeline](17-celery-rate-limit-lock-pipeline.md) ## FILE: docs/guides/06-taskiq-rabbitmq-example.md # Example: Taskiq + RabbitMQ + FastAPI **Audience:** Async-first FastAPI teams. **Why Taskiq:** Native async tasks, broker plugins for RabbitMQ (and Kafka), dependency injection friendly. ## TL;DR ```text API process: create job row → task.kiq(...) Worker process: taskiq worker app.workers.broker:broker Broker: RabbitMQ ``` ## Contents 1. [Broker setup](#1-broker-setup) 2. [Task definition](#2-task-definition) 3. [FastAPI integration](#3-fastapi-integration) 4. [Concurrency model](#4-concurrency-model) 5. [Retries and middleware](#5-retries-and-middleware) 6. [Run](#6-run) 7. [When to prefer Celery instead](#7-when-to-prefer-celery-instead) --- ## 1. Broker setup ```python # PSEUDOCODE — app/workers/broker.py from taskiq_aio_pika import AioPikaBroker from app.core.settings import settings broker = AioPikaBroker(settings.rabbitmq_url) # optional result backend if you need it; product status still in DB ``` ## 2. Task definition ```python # PSEUDOCODE — app/workers/tasks/exports.py from app.workers.broker import broker from app.db import get_session from app.services.export import build_export @broker.task( task_name="export_csv", queue_name="jobs.heavy", retry_on_error=True, max_retries=5, ) async def export_csv(job_id: str) -> None: async with get_session() as db: job = await db.get_job(job_id) if job is None or job.status == "succeeded": return await db.mark_running(job_id) try: ref = await build_export(job) # async I/O await db.mark_succeeded(job_id, result_ref=ref) except PermanentError as e: await db.mark_failed(job_id, str(e)) raise # transient: let Taskiq retry policy re-raise ``` ## 3. FastAPI integration ```python # PSEUDOCODE @router.post("/exports", status_code=202) async def start_export(body: ExportIn, db: Db = Depends()): job = await db.create_job(type="export_csv", ...) await export_csv.kiq(str(job.id)) return {"job_id": job.id, "status": "pending"} ``` Startup: ensure broker is started/stopped in lifespan if the library requires it for the API process (publish only). ## 4. Concurrency model | Work type | Guidance | |-----------|----------| | Many small HTTP jobs | Higher async concurrency | | CPU-heavy export | Low concurrency queue `jobs.heavy` | | Mixed | Split queues; never starve light jobs behind heavy ones | **Rule:** no blocking `requests` / sync ORM inside async tasks without thread offload. ## 5. Retries and middleware - Configure max retries + backoff at task or broker level - Log `task_id` + `job_id` in middleware - Propagate trace headers if OTel is enabled ## 6. Run ```bash taskiq worker app.workers.broker:broker # API separately uvicorn app.main:app --host 0.0.0.0 --port 8000 ``` Compose services: `api`, `worker`, `rabbitmq`, `db`. ## 7. When to prefer Celery instead - You need Battle-tested Beat + Flower today - Team already operates Celery at scale - Mostly sync libraries and prefork comfort Otherwise Taskiq fits async FastAPI cleanly. **Related:** [05 Celery](05-celery-rabbitmq-example.md) · [07 Dramatiq](07-dramatiq-rabbitmq-example.md) ## FILE: docs/guides/07-dramatiq-rabbitmq-example.md # Example: Dramatiq + RabbitMQ + FastAPI **Audience:** Teams wanting simpler actors than Celery on AMQP. **Model:** Sync actors, RabbitMQ broker, middleware for retries. ## TL;DR ```text @dramatiq.actor → RabbitMQ → dramatiq worker FastAPI enqueues via actor.send(job_id) ``` ## Contents 1. [Broker and actor](#1-broker-and-actor) 2. [Enqueue from FastAPI](#2-enqueue-from-fastapi) 3. [Retries and time limits](#3-retries-and-time-limits) 4. [Queues](#4-queues) 5. [Run](#5-run) 6. [Pros and cons](#6-pros-and-cons) --- ## 1. Broker and actor ```python # PSEUDOCODE — app/workers/dramatiq_app.py import dramatiq from dramatiq.brokers.rabbitmq import RabbitmqBroker from app.core.settings import settings broker = RabbitmqBroker(url=settings.rabbitmq_url) dramatiq.set_broker(broker) @dramatiq.actor( queue_name="jobs.io", max_retries=5, min_backoff=1_000, max_backoff=600_000, time_limit=45_000, # ms ) def send_receipt(job_id: str): job = get_job(job_id) if job is None or job.status == "succeeded": return mark_running(job_id) try: do_send(job) mark_succeeded(job_id) except PermanentError as e: mark_failed(job_id, str(e)) raise ``` ## 2. Enqueue from FastAPI ```python # PSEUDOCODE @router.post("/receipts", status_code=202) def enqueue_receipt(body: ReceiptIn, db: Session = Depends()): job = create_job(db, type="send_receipt", ...) send_receipt.send(str(job.id)) return {"job_id": job.id, "status": "pending"} ``` Use `send` / `send_with_options` for queue overrides. ## 3. Retries and time limits Dramatiq retries via middleware. Map: - Transient errors → raise and retry - Permanent → mark failed and use `max_retries=0` path or catch without re-raise after mark Dead letters: configure RabbitMQ DLX for the queue. ## 4. Queues Same split as Celery: `jobs.default`, `jobs.io`, `jobs.heavy` with separate worker processes if needed. ## 5. Run ```bash dramatiq app.workers.dramatiq_app # or module path containing actors ``` ## 6. Pros and cons | Pros | Cons | |------|------| | Simple mental model | Less ecosystem than Celery | | Solid RabbitMQ support | Not async-native | | Good defaults for retries | Kafka not the focus | Prefer **Taskiq** if your workers are heavily async; prefer **Celery** if you need Beat/Flower maturity. **Related:** [05 Celery](05-celery-rabbitmq-example.md) · [06 Taskiq](06-taskiq-rabbitmq-example.md) ## FILE: docs/guides/08-faststream-kafka-example.md # Example: FastStream + Kafka (events) **Audience:** Services emitting or consuming domain/integration events. **Policy:** Kafka for multi-consumer events; RabbitMQ remains for commands/jobs. ## TL;DR ```text Producer (API or worker): after commit → outbox → Kafka topic Consumer: FastStream subscriber → inbox → local projection / new command ``` ## Contents 1. [When Kafka vs RabbitMQ](#1-when-kafka-vs-rabbitmq) 2. [Topic design](#2-topic-design) 3. [Producer pseudocode](#3-producer-pseudocode) 4. [Consumer pseudocode](#4-consumer-pseudocode) 5. [Ordering and keys](#5-ordering-and-keys) 6. [Offsets and lag](#6-offsets-and-lag) 7. [Schema evolution](#7-schema-evolution) 8. [Run](#8-run) --- ## 1. When Kafka vs RabbitMQ | Use Kafka | Use RabbitMQ | |-----------|--------------| | Many consumers of the same fact | One worker should do a job | | Replay / log history | Per-message ack work queue | | Stream processing | Classic task queue, DLX jobs | ## 2. Topic design ```text billing.events # domain events from billing billing.invoice.paid # optional finer topics ``` Envelope: ```json { "event_id": "uuid", "type": "invoice.paid.v1", "occurred_at": "2026-08-05T12:00:00Z", "payload": { "invoice_id": "...", "customer_id": "...", "amount_cents": 1200 } } ``` ## 3. Producer pseudocode ```python # PSEUDOCODE — prefer outbox, not dual-write async with db.transaction(): invoice.pay() await db.save(invoice) await db.outbox_add( topic="billing.events", key=invoice.customer_id, payload=envelope, ) ``` Relay publishes to Kafka with the key for partition affinity. ## 4. Consumer pseudocode ```python # PSEUDOCODE — FastStream-style from faststream import FastStream from faststream.kafka import KafkaBroker broker = KafkaBroker(settings.kafka_bootstrap) app = FastStream(broker) @broker.subscriber("billing.events", group_id="notifications") async def on_billing_event(message: dict): if message["type"] != "invoice.paid.v1": return if await inbox_seen(message["event_id"]): return await enqueue_email_job(message["payload"]["invoice_id"]) await inbox_insert(message["event_id"]) ``` ## 5. Ordering and keys - Set **key** = aggregate id when order per entity matters - Different keys → parallel partitions - Do not assume global order across keys ## 6. Offsets and lag - Commit offsets **after** successful side effects (or use inbox so reprocessing is safe) - Alert on **consumer lag** as the primary backlog SLO - Scale consumers up to partition count ## 7. Schema evolution - Version in `type` (`invoice.paid.v1` → `v2`) - Additive fields first; consumers ignore unknown keys - Document in the event catalog ([11](11-event-driven-system-design.md)) ## 8. Run ```bash # consumer process faststream run app.workers.stream:app # or uvicorn if mounted; keep consumer separate from API when possible ``` **Related:** [11 Event-driven design](11-event-driven-system-design.md) · [04 Outbox](04-outbox-inbox-idempotency.md) ## FILE: docs/guides/09-error-taxonomy-retries-dlq.md # Error taxonomy, retries, and DLQ **Audience:** Anyone configuring worker retries. **Goal:** Retry what heals; fail fast on poison; never infinite loops. ## TL;DR | Error class | Action | |-------------|--------| | **Transient** | Retry with backoff + jitter | | **Permanent** | Mark failed; do not retry | | **Unknown** | Limited retries, then DLQ + alert | | **Poison message** | DLQ after max attempts; fix and redrive | ## Contents 1. [Taxonomy](#1-taxonomy) 2. [Retry policy](#2-retry-policy) 3. [Timeouts](#3-timeouts) 4. [Dead-letter queues](#4-dead-letter-queues) 5. [Pseudocode](#5-pseudocode) 6. [Alerting](#6-alerting) 7. [Anti-patterns](#7-anti-patterns) --- ## 1. Taxonomy | Class | Examples | Retry? | |-------|----------|--------| | Transient network | 503, timeout, connection reset | Yes | | Transient overload | 429, broker full | Yes (honor Retry-After) | | Dependency down | DB failover in progress | Yes (bounded) | | Validation / schema | Bad payload, missing field | No | | Business rule | Invoice already void | No (or no-op success) | | Auth config | Wrong API key | No until fixed | | Bug / NPE | Unexpected exception | Limited then DLQ | Encode classes as exceptions or error codes your middleware understands. ## 2. Retry policy Recommended defaults (tune with data): ```text max_attempts: 5 backoff: exponential base: 1s cap: 10m jitter: full or equal jitter ``` ```python # PSEUDOCODE delay = min(cap, base * 2 ** attempt) * random(0.5, 1.5) ``` **Do not** retry non-idempotent permanent side effects without a guard. ## 3. Timeouts Every external call needs a timeout: - HTTP client: connect + read - DB statement timeout - Soft/hard task time limits (Celery/Dramatiq) A task without timeouts becomes a stuck consumer (prefetch blocked). ## 4. Dead-letter queues ### RabbitMQ - Queue with DLX → `jobs.dlq` - Reject/nack without requeue after max attempts - Redrive tool for operators ### Kafka - Error topic `billing.events.errors` - Or stop-the-world + alert on poison (document choice) ## 5. Pseudocode ```python # PSEUDOCODE async def handle(job_id: str): try: await run(job_id) await mark_succeeded(job_id) except PermanentError as e: await mark_failed(job_id, str(e)) # ack / no retry except TransientError as e: attempts = await bump(job_id, str(e)) if attempts >= max_attempts: await mark_failed(job_id, str(e)) await publish_dlq(job_id) return raise Retry(delay=backoff(attempts)) except Exception as e: attempts = await bump(job_id, "unknown:" + str(e)) if attempts >= max_attempts: await mark_failed(job_id, str(e)) await publish_dlq(job_id) await alert("poison_or_bug", job_id) return raise Retry(delay=backoff(attempts)) ``` ## 6. Alerting | Signal | Severity | |--------|----------| | DLQ depth > 0 for critical queues | High | | Retry rate spike | Medium | | Success rate drop | High | | Time-to-complete SLO breach | Medium | | Zero consumers | Critical | ## 7. Anti-patterns - `except Exception: retry` forever - No jitter (thundering herd) - Retrying after non-idempotent charge without guard - Silent drop of poison messages - Using DLQ as a black hole with no runbook **Related:** [03 Lifecycle](03-job-lifecycle-and-schema.md) · [10 Testing](10-testing-workers.md) ## FILE: docs/guides/10-testing-workers.md # Testing workers and async jobs **Audience:** Engineers writing CI for API + workers. **Goal:** Confidence without flaky full-stack broker tests on every PR. ## TL;DR pyramid ```text Domain / handler unit tests (many, fast) API enqueue tests with fakes (many) Contract tests for messages (some) Integration with real broker (few, CI job or nightly) Staging smoke e2e (per deploy) ``` ## Contents 1. [What to test](#1-what-to-test) 2. [Unit: handlers](#2-unit-handlers) 3. [API: enqueue](#3-api-enqueue) 4. [Idempotency tests](#4-idempotency-tests) 5. [Integration](#5-integration) 6. [Staging smoke](#6-staging-smoke) 7. [Fixtures and fakes](#7-fixtures-and-fakes) 8. [CI layout](#8-ci-layout) --- ## 1. What to test | Layer | Assert | |-------|--------| | Domain | Invariants (pay twice, void paid) | | Handler | Status transitions, retries classification | | API | 202, job row, publish called with routing key | | Consumer | Inbox prevents double side effect | | Smoke | One real job completes end-to-end | ## 2. Unit: handlers ```python # PSEUDOCODE def test_export_skips_if_already_succeeded(): job = Job(status="succeeded") db = FakeDb(job) handle_export(job.id, db=db, bus=FakeBus()) assert db.export_calls == 0 def test_transient_error_bumps_attempt(): ... ``` No real RabbitMQ required. ## 3. API: enqueue ```python # PSEUDOCODE def test_post_export_returns_202_and_publishes(client, fake_bus): r = client.post("/exports", json={...}, headers={"Idempotency-Key": "k1"}) assert r.status_code == 202 assert fake_bus.messages[0]["queue"] == "jobs.heavy" assert db.jobs[0].status == "pending" def test_duplicate_idempotency_no_second_publish(client, fake_bus): client.post(..., headers={"Idempotency-Key": "k1"}) client.post(..., headers={"Idempotency-Key": "k1"}) assert len(fake_bus.messages) == 1 ``` ## 4. Idempotency tests - Double delivery of same `event_id` → one email - `try_mark_running` race → only one worker proceeds - Outbox relay publishes each row once ## 5. Integration Use Testcontainers or compose in CI: - Postgres + RabbitMQ - Run worker process - Enqueue via API - Poll job until succeeded (timeout) Mark as optional/slow job if PR feedback must stay fast. ## 6. Staging smoke ```text POST /exports → poll GET /jobs/{id} → expect succeeded within SLO ``` Run after every deploy. Alert on failure. ## 7. Fixtures and fakes ```python # PSEUDOCODE class FakeBus: def __init__(self): self.messages = [] async def publish(self, queue, payload): self.messages.append({"queue": queue, "payload": payload}) ``` Override `Depends(get_bus)` in FastAPI tests. ## 8. CI layout ```text lint → typecheck → unit+api → (integration optional) → build image ``` Never require Flower or production broker credentials for unit tests. **Related:** [09 Errors](09-error-taxonomy-retries-dlq.md) · [03 Lifecycle](03-job-lifecycle-and-schema.md) ## FILE: docs/guides/11-event-driven-system-design.md # Designing an event-driven system (with FastAPI) **Audience:** Teams moving beyond a single API + DB into async reactions. **Pairs with:** RabbitMQ for commands, Kafka for domain events (this checklist). ## TL;DR 1. Name **bounded contexts** 2. Separate **commands** (do work) from **events** (something happened) 3. Commit state → **outbox** → bus 4. Consumers own **local models** + **inbox** 5. Keep a living **event catalog** ## Contents 1. [Message kinds](#1-message-kinds) 2. [Design steps](#2-design-steps) 3. [Topology](#3-topology) 4. [Event catalog](#4-event-catalog) 5. [Naming and versioning](#5-naming-and-versioning) 6. [Delivery guarantees](#6-delivery-guarantees) 7. [Pseudocode](#7-pseudocode) 8. [Process managers](#8-process-managers) 9. [Anti-patterns](#9-anti-patterns) 10. [Rollout plan](#10-rollout-plan) --- ## 1. Message kinds | Kind | Meaning | Bus | Example | |------|---------|-----|---------| | Command | Please do X | RabbitMQ | `GenerateExport` | | Domain event | X happened in our model | Kafka (or RMQ topic) | `InvoicePaid` | | Integration event | Cross-service contract | Kafka | `billing.invoice_paid.v1` | Do not treat them as interchangeable. ## 2. Design steps 1. **Bounded contexts** — who owns language and data? 2. **Trigger path** — API writes state, then emits command/event 3. **Name events** — past tense, versioned 4. **Pick guarantees** — outbox, inbox, keys for order 5. **Catalog** — producers, consumers, failure modes 6. **Observe** — lag, DLQ, publish rate ## 3. Topology ```text Client ──HTTP──► FastAPI (context) │ ├── commands ──► RabbitMQ ──► job workers │ └── outbox ──► Kafka ──► consumer A (own DB) └─► consumer B (own DB) ``` ## 4. Event catalog For every event: | Field | Example | |-------|---------| | Name | `invoice.paid.v1` | | Producer | Billing API / worker | | Bus / topic | Kafka `billing.events` | | Key | `customer_id` | | Schema | JSON Schema / Pydantic | | Consumers | Notifications, Ledger | | Failure | retry → error topic | Without a catalog, event-driven systems become tribal knowledge. ## 5. Naming and versioning - Past tense for facts: `OrderPlaced`, not `PlaceOrder` - Include version: `.v1` - Payload = contract (not ORM dump) - Stable `event_id` for inbox ## 6. Delivery guarantees | Need | Pattern | |------|---------| | User job status | Job row + RMQ command | | Multi-service reaction | Domain event on Kafka | | No lost event after commit | Transactional outbox | | At-least-once | Idempotent consumers + inbox | | Order per entity | Kafka key = aggregate id | ## 7. Pseudocode ### Publish after state change ```python # PSEUDOCODE — Billing async def mark_invoice_paid(invoice_id, db, outbox): async with db.transaction(): invoice = await db.lock_invoice(invoice_id) if invoice.status == "paid": return invoice invoice.status = "paid" await db.save(invoice) await outbox.add( topic="billing.events", key=invoice.customer_id, payload={ "event_id": new_uuid(), "type": "invoice.paid.v1", "invoice_id": invoice.id, "customer_id": invoice.customer_id, "amount_cents": invoice.amount_cents, "occurred_at": utcnow_iso(), }, ) return invoice ``` ### Consumer ```python # PSEUDOCODE — Notifications async def on_invoice_paid(event, db): if await db.inbox_seen(event["event_id"]): return await send_receipt_email(event["customer_id"], event["invoice_id"]) await db.inbox_insert(event["event_id"]) ``` ### Combined with a job ```text POST export → job command (RMQ) → worker finishes → domain event export.completed.v1 (Kafka) → analytics consumer updates dashboard projection ``` ## 8. Process managers Long workflows spanning contexts: ```text OrderPlaced → reserve inventory command → InventoryReserved → charge payment command → PaymentCaptured → ship command ``` Implement as a **policy** service that listens to events and sends commands. Keep state in its own table; do not share DBs. ## 9. Anti-patterns | Avoid | Prefer | |-------|--------| | God event with full aggregate graph | Small purpose-built payloads | | Shared DB as integration | Events + local read models | | Sync HTTP chains for every side effect | Commands/events + retries | | One undifferentiated topic forever | Clear topics + catalog | | Consumer querying producer DB | Consumer owns projection | ## 10. Rollout plan 1. One context, one command queue, one event type 2. Outbox for that event 3. One consumer with inbox 4. Metrics: publish rate, lag, DLQ 5. Expand catalog before the fifth event type **Related:** [12 DDD](12-ddd-with-fastapi.md) · [04 Outbox](04-outbox-inbox-idempotency.md) · [08 FastStream](08-faststream-kafka-example.md) ## FILE: docs/guides/12-ddd-with-fastapi.md # Using DDD with FastAPI (and workers) **Audience:** Teams whose FastAPI apps outgrew “fat routes + shared models.” **DDD is:** language, boundaries, and ownership — not a folder religion and not “event sourcing required.” ## TL;DR | Idea | In this stack | |------|----------------| | Bounded context | Package (or service) with own model | | Aggregate | Transaction boundary + invariants | | Application service | Use case called by HTTP **and** workers | | Domain event | Outbox → Kafka after commit | | Infrastructure | SQLAlchemy, RMQ, SMTP as adapters | ## Contents 1. [Why DDD helps](#1-why-ddd-helps) 2. [Core ideas](#2-core-ideas) 3. [Suggested layout](#3-suggested-layout) 4. [Aggregates](#4-aggregates) 5. [Application use cases](#5-application-use-cases) 6. [Mapping to jobs and events](#6-mapping-to-jobs-and-events) 7. [Contexts vs packages](#7-contexts-vs-packages) 8. [Tactical patterns](#8-tactical-patterns) 9. [Testing](#9-testing) 10. [Anti-patterns](#10-anti-patterns) 11. [Adoption path](#11-adoption-path) --- ## 1. Why DDD helps Without boundaries, FastAPI codebases often grow: - Fat routers with business rules - One `models.py` for the company - Jobs that UPDATE five “domains” in one transaction DDD pushes **invariants next to the data**, **explicit integration**, and **shared language**. ## 2. Core ideas | Idea | Meaning | |------|---------| | Ubiquitous language | Same words in code, API, events, tickets | | Bounded context | Model boundary; words can differ across contexts | | Aggregate | Cluster with one root; change via root; one TX | | Domain event | Fact from a meaningful state change | | Application service | Orchestrates load → domain → save → outbox | | Infrastructure | Frameworks and I/O adapters | ## 3. Suggested layout ```text app/ contexts/ billing/ domain/ # entities, value objects, domain events, errors application/ # use cases adapters/ api/ # FastAPI routers + Pydantic persistence/ # ORM, repositories messaging/ # publish commands/events notifications/ domain/ application/ adapters/ workers/ # process entrypoints → application use cases main.py ``` Start as a **modular monolith**. Split deployables when scale or team ownership demands it. ## 4. Aggregates ```python # PSEUDOCODE — domain (no FastAPI imports) @dataclass class Invoice: id: str status: str # draft | issued | paid | void amount_cents: int def pay(self, at: datetime) -> list[DomainEvent]: if self.status == "paid": return [] if self.status != "issued": raise DomainError("only issued invoices can be paid") self.status = "paid" return [InvoicePaid(self.id, self.amount_cents, at)] ``` Rules live **on the aggregate**, not only in route validators. ## 5. Application use cases ```python # PSEUDOCODE async def pay_invoice(cmd: PayInvoice, repo: InvoiceRepo, outbox: Outbox) -> None: async with unit_of_work(): invoice = await repo.get(cmd.invoice_id) events = invoice.pay(utcnow()) await repo.save(invoice) for e in events: await outbox.add(e.to_integration_event()) ``` Thin route: ```python @router.post("/invoices/{invoice_id}/pay", status_code=204) async def pay(invoice_id: str, svc=Depends(get_billing_app)): await svc.pay_invoice(PayInvoice(invoice_id=invoice_id)) ``` Workers call the **same** use case (or a command variant) with a messaging adapter. ## 6. Mapping to jobs and events | DDD | Implementation | |-----|----------------| | User command | HTTP POST or RabbitMQ job message | | Domain event | Aggregate returns events → outbox → Kafka | | Policy / process manager | Worker listening to events, sending commands | | Read model | Projection table updated by consumer | | Anti-corruption layer | Translate external payloads into your language | ```text POST /invoices/{id}/pay → Billing application service → Invoice.pay() → outbox: invoice.paid.v1 → Kafka → Notifications policy → command SendReceiptEmail → RabbitMQ → email worker ``` ## 7. Contexts vs packages - Do **not** import `billing.domain` from `notifications.domain` - Notifications may depend on **integration event schemas**, not Billing ORM - Shared kernel only for true shared primitives (`Money`, `TenantId`) if agreed ## 8. Tactical patterns | Pattern | Use when | |---------|----------| | Modular monolith | Default | | Light CQRS | Heavy/different read models | | Event sourcing | Hard requirement to rebuild from events (costly) | | Sagas / process managers | Multi-step cross-context workflows | | Ports & adapters | Test without DB/broker | You can practice DDD **without** event sourcing. ## 9. Testing 1. Domain tests — pure aggregate rules 2. Application tests — fake repo + fake outbox 3. Adapter tests — HTTP → command; message → command 4. Contract tests — event payload schema ## 10. Anti-patterns | Smell | Fix | |-------|-----| | Anemic models + all logic in routes | Move invariants to aggregates | | One company-wide models module | Split by context | | Job SQL-updates five contexts | Events + local updates | | Shared session across contexts “for convenience” | Explicit messaging | | Language only in docs | Rename code and events | ## 11. Adoption path 1. Name 2–3 contexts on one page 2. One aggregate with real invariants 3. Thin the route; rules on the aggregate 4. One domain event via outbox 5. One consumer + inbox in another context 6. Only then split deployables **Related:** [11 Event-driven](11-event-driven-system-design.md) · [04 Outbox](04-outbox-inbox-idempotency.md) ## FILE: docs/guides/13-laravel-queues-to-fastapi.md # Laravel-style queues → FastAPI (RabbitMQ / Kafka) **Audience:** Teams who know Laravel queues and are building the same ideas with FastAPI. **Policy reminder:** Durable jobs use **RabbitMQ** (commands) and/or **Kafka** (events). Redis is not the primary job broker in this checklist. ## TL;DR Laravel gives you one opinionated queue API (`ShouldQueue`, `dispatch`, middleware, batches, Horizon). In FastAPI you assemble the same **behaviors** from: | Laravel idea | FastAPI-side approach | |--------------|------------------------| | `ShouldQueue` job class | Task/actor function + **job row** in your DB | | `dispatch()` | Publish to RabbitMQ (Celery / Taskiq / Dramatiq) | | `queue:work` | Separate **worker** process/container | | Connections vs queues | Broker URL vs queue/topic names | | Job middleware | Task decorators / custom wrappers / middleware | | Unique / without overlapping | Idempotency key + cache/DB lock | | `after_commit` | Transactional **outbox** (preferred) | | Failed jobs table | `jobs` status=`failed` + DLQ | | Horizon | Flower / RMQ management / Kafka UI + your metrics | | Batches / chains | Orchestrator job, saga, or workflow table | Laravel is a **framework product**. FastAPI is a **web library** — you own the job domain model. ## Contents 1. [Mental model](#1-mental-model) 2. [Connections vs queues](#2-connections-vs-queues) 3. [Creating jobs](#3-creating-jobs) 4. [Dispatching](#4-dispatching) 5. [Middleware equivalents](#5-middleware-equivalents) 6. [Retries, timeouts, failed jobs](#6-retries-timeouts-failed-jobs) 7. [Workers and deploy](#7-workers-and-deploy) 8. [Batches, chains, unique, debounce](#8-batches-chains-unique-debounce) 9. [Testing](#9-testing) 10. [Feature map (Laravel → stack)](#10-feature-map-laravel--stack) 11. [What not to copy blindly](#11-what-not-to-copy-blindly) --- ## 1. Mental model ### Laravel ```text Controller → ProcessPodcast::dispatch($model) Worker (queue:work) → handle() failed_jobs / Horizon ``` ### FastAPI (this checklist) ```text Route → insert jobs row (pending) → publish command (RabbitMQ) Worker process → load job by id → run → succeeded|failed Client → GET /jobs/{id} (product status from DB, not broker UI) ``` | Laravel | You implement | |---------|----------------| | Framework serializes job class | JSON message `{ "job_id", "type" }` only | | Eloquent model on queue | Pass **IDs**; reload in worker | | `failed_jobs` migration | Your `jobs` table + optional DLQ | | `Queue::fake()` | Fake bus / mock publish in tests | See [03 Job lifecycle](03-job-lifecycle-and-schema.md). --- ## 2. Connections vs queues Laravel: - **Connection** = backend (redis, sqs, database, …) in `config/queue.php` - **Queue name** = stack on that connection (`emails`, `high`, …) FastAPI / Celery-style: | Concept | Example | |---------|---------| | Connection / broker | `amqps://…@rabbitmq//` (Pydantic Settings) | | Queue name | `jobs.default`, `jobs.high`, `jobs.heavy`, `jobs.io` | | Kafka “connection” | Bootstrap servers + topic `billing.events` | ```python # PSEUDOCODE — settings class Settings(BaseSettings): rabbitmq_url: str default_job_queue: str = "jobs.default" ``` ```python # PSEUDOCODE — route to named queue (Celery-like) send_receipt.apply_async(args=[job_id], queue="jobs.io") # Taskiq await send_receipt.kicker().with_queue("jobs.io").kiq(job_id) ``` Worker priority (Laravel `--queue=high,default`): ```bash # Celery: separate workers or -Q high,default celery -A app worker -Q jobs.high,jobs.default ``` --- ## 3. Creating jobs ### Laravel job class ```php class ProcessPodcast implements ShouldQueue { public function handle(AudioProcessor $processor): void { ... } } ``` ### FastAPI equivalent shape ```python # PSEUDOCODE — prefer IDs on the bus @celery_app.task(name="podcasts.process") def process_podcast(job_id: str) -> None: job = load_job(job_id) if job.status == "succeeded": return podcast = load_podcast(job.entity_id) # reload; don't trust stale snapshot AudioProcessor().run(podcast) mark_succeeded(job_id) ``` | Laravel habit | FastAPI habit | |---------------|---------------| | Pass Eloquent model into job | Pass `podcast_id` / `job_id` | | Container injects into `handle` | Explicit deps or DI in Taskiq | | Huge serialized relations | Forbidden — small JSON only | | Binary on queue | Object storage + ref | --- ## 4. Dispatching | Laravel | FastAPI pattern | |---------|-----------------| | `ProcessPodcast::dispatch($p)` | Create job row + `task.delay(job_id)` / `.kiq` / `.send` | | `dispatch()->delay(...)` | ETA/countdown (Celery), schedule, or delayed exchange | | `dispatch_sync` | Call use-case function in-process (tests/admin only) | | `Bus::batch([...])` | Batch table + child jobs (you build it) | | `Bus::chain([...])` | Chain in message / next_step field / saga | | `after_commit()` | **Outbox** in same DB transaction ([04](04-outbox-inbox-idempotency.md)) | ### after_commit ≈ outbox (important) Laravel `after_commit` avoids dispatching if the HTTP transaction rolls back. In FastAPI, do **not** rely on “publish after await commit” alone under load — use: ```python # PSEUDOCODE async with db.transaction(): job = await insert_job(...) await insert_outbox(queue="jobs.default", payload={"job_id": job.id}) # relay publishes after commit ``` ### 202 response (Laravel often still returns 200 after dispatch) ```python # PSEUDOCODE @router.post("/podcasts/{id}/process", status_code=202) async def process(id: str, db: Db, bus: Bus): job = await enqueue(db, bus, type="podcast.process", entity_id=id) return {"job_id": job.id, "status": "pending"} ``` --- ## 5. Middleware equivalents Laravel job middleware (rate limit, without overlapping, throttle exceptions) map to: | Laravel middleware | FastAPI approach | |--------------------|------------------| | `RateLimited` | Redis token bucket in task wrapper; per-queue worker concurrency; broker prefetch | | `WithoutOverlapping` | Redis/DB lock keyed by `user_id` / `order_id` before work | | `ThrottlesExceptions` | Error taxonomy + backoff ([09](09-error-taxonomy-retries-dlq.md)) | | `release($seconds)` | Retry with countdown / nack + delay plugin | | Skip job | Early `return` if `job.status == succeeded` | ```python # PSEUDOCODE — WithoutOverlapping-style async def handle(job_id: str, entity_id: str): lock = await redis.lock(f"job:podcast:{entity_id}", ttl=300) if not await lock.acquire(blocking=False): raise Retry(countdown=30) # or release equivalent try: await do_work(job_id) finally: await lock.release() ``` Unique jobs (`ShouldBeUnique`): ```python # PSEUDOCODE — unique by product # 1) UNIQUE(idempotency_key) on jobs table, or # 2) cache lock before publish key = f"unique:reindex:{product_id}" if not await cache.set_nx(key, "1", ttl=3600): return existing_job ``` Debounced jobs: store “latest wins” token in Redis; worker checks token still current before running (or use a short delay + version number). Encrypted jobs: encrypt payload fields yourself or put secrets in vault and pass IDs only (preferred). --- ## 6. Retries, timeouts, failed jobs | Laravel | FastAPI / Celery-like | |---------|------------------------| | `--tries=3` | `max_retries` / task config | | `--backoff` / `backoff()` | Exponential backoff + jitter | | `--timeout` | Soft/hard time limits; HTTP client timeouts | | `retry_after` (visibility) | RabbitMQ consumer timeout / ack rules; don’t set visibility < runtime | | `failed_jobs` table | `jobs.status=failed` + `last_error` | | `queue:retry` | Admin redrive from DLQ or `status=pending` + re-publish | | `failed()` method | `mark_failed` + alert hook | | `DeleteWhenMissingModels` | Catch not-found → mark cancelled/failed without retry | ```python # PSEUDOCODE — failed hook except PermanentError as e: await mark_failed(job_id, str(e)) await notify_ops(job_id, e) ``` DLQ: RabbitMQ dead-letter exchange → `jobs.dlq` (see [09](09-error-taxonomy-retries-dlq.md)). --- ## 7. Workers and deploy | Laravel | FastAPI stack | |---------|----------------| | `php artisan queue:work` | `celery worker` / `taskiq worker` / `dramatiq` | | Supervisor `numprocs` | K8s replicas / Compose `worker` service scale | | `queue:restart` | Rolling restart; graceful SIGTERM; finish in-flight | | Maintenance mode skips jobs | Your feature flag or stop workers | | Horizon (Redis) | Flower (Celery) + RMQ UI; **not** user-facing status | | `--queue=high,default` | `-Q jobs.high,jobs.default` or dedicated deployments | ```text # compose api: uvicorn worker: celery/taskiq/dramatiq rabbitmq: db: # optional flower (private network + auth) ``` Graceful shutdown: termination grace **>** longest job (same idea as Supervisor `stopwaitsecs`). --- ## 8. Batches, chains, unique, debounce Laravel has first-class **batches** and **chains**. In FastAPI you model them explicitly: ### Chain ```python # PSEUDOCODE — next step on success async def step_a(job_id): await do_a() await enqueue_child(type="step_b", parent_id=job_id) ``` Or a single orchestrator message with `steps: ["a","b","c"]` and cursor. ### Batch ```sql -- PSEUDOCODE batch(id, status, total, done, failed) batch_jobs(batch_id, job_id) ``` When `done + failed == total`, mark batch complete / run `then` callback job. ### SQS FIFO / fair queues On RabbitMQ: separate queues + careful prefetch; per-tenant queues if needed. On Kafka: key by tenant for ordering. True SQS FIFO is an AWS concept — use RMQ/Kafka patterns instead unless you actually run SQS from Python. --- ## 9. Testing | Laravel | FastAPI | |---------|---------| | `Queue::fake()` | `FakeBus` dependency override | | `Queue::assertPushed` | Assert `fake_bus.messages` | | `Bus::assertChained` | Assert ordered publishes / child rows | | Run `handle` in unit test | Call task function with fakes | ```python # PSEUDOCODE def test_dispatch_export(client, fake_bus): r = client.post("/exports", json={...}) assert r.status_code == 202 assert fake_bus.messages[0]["queue"] == "jobs.heavy" ``` Full guide: [10 Testing workers](10-testing-workers.md). --- ## 10. Feature map (Laravel → stack) | Laravel queues doc section | Closest implementation | |----------------------------|-------------------------| | Introduction / why queues | [01 Async vs jobs](01-async-vs-jobs.md) | | Connections vs queues | Settings + queue names (this guide §2) | | Creating job classes | Celery/Taskiq/Dramatiq tasks + job table | | Unique jobs | Idempotency key + lock | | Debounced jobs | Redis version / delay + “latest token” | | Encrypted jobs | Don’t put secrets on bus; encrypt if you must | | Job middleware | Wrappers / locks / rate limits | | Delayed dispatch | ETA / delayed message | | Sync dispatch | In-process call | | Bulk dispatch | Loop + bulk insert jobs + multi-publish | | Jobs & DB transactions | **Outbox** [04](04-outbox-inbox-idempotency.md) | | Job chaining | Saga / next_step / chain messages | | Max attempts / timeout | Task config + client timeouts [09](09-error-taxonomy-retries-dlq.md) | | Queue failover | Multi-AZ broker; not “second Redis” as source of truth | | Error handling | Taxonomy + DLQ | | Job batching | Batch tables you own | | Queueing closures | Avoid; named tasks only (debuggable) | | `queue:work` | Worker containers | | Supervisor | K8s/systemd/Compose restart policy | | Failed jobs | `jobs` failed + DLQ redrive | | Clearing queues | RMQ purge (ops only; dangerous) | | Monitoring | Depth/lag metrics + alerts | | Testing | Fakes + smoke [10](10-testing-workers.md) | | Job events | Logging middleware / OpenTelemetry spans | ### Framework pick (Laravel “one way” → your choice) | If you liked… | Prefer | |---------------|--------| | Jobs + Horizon-ish ops + sync PHP style | **Celery + RabbitMQ** (+ Flower private) | | Modern async, DI | **Taskiq + RabbitMQ** | | Small actors, simple API | **Dramatiq + RabbitMQ** | | Events / pub-sub like Laravel events at scale | **FastStream + Kafka** (+ outbox) | --- ## 11. What not to copy blindly | Laravel convenience | Risk if copied naively | |---------------------|-------------------------| | Redis as default queue | This checklist wants **RabbitMQ/Kafka** for durable business jobs | | Eloquent models on the queue | Huge payloads, stale relations — use IDs | | Closures on the queue | Opaque, hard to ops | | Horizon as product status | Users need **your** job API | | `dispatch` without job row | No user-visible status / audit | | Infinite retries | Poison loops — bound attempts + DLQ | --- ## Pseudocode: “Laravel dispatch” in one place ```python # PSEUDOCODE — application service used by FastAPI routes async def dispatch_job( *, type: str, entity_id: str, idempotency_key: str | None, queue: str, db: Db, bus: Bus, ) -> Job: if idempotency_key: existing = await db.find_job_by_key(idempotency_key) if existing: return existing async with db.transaction(): job = await db.insert_job( type=type, status="pending", entity_id=entity_id, idempotency_key=idempotency_key, ) await db.insert_outbox( destination=queue, payload={"job_id": str(job.id), "type": type}, ) return job ``` That is the portable core of Laravel’s “push a job and forget” — with production-safe dual-write behavior. ## See also - [01 Async vs jobs](01-async-vs-jobs.md) - [02 Brokers](02-brokers-celery-redis-rabbitmq-kafka.md) - [03 Job lifecycle](03-job-lifecycle-and-schema.md) - [04 Outbox / inbox](04-outbox-inbox-idempotency.md) - [05 Celery](05-celery-rabbitmq-example.md) · [06 Taskiq](06-taskiq-rabbitmq-example.md) · [07 Dramatiq](07-dramatiq-rabbitmq-example.md) - [09 Errors / DLQ](09-error-taxonomy-retries-dlq.md) - [10 Testing](10-testing-workers.md) ## External reading - [Laravel queues (official)](https://laravel.com/docs/queues) — conceptual checklist for features to re-implement - Not a runtime dependency of this FastAPI standard **Creator note:** Patterns above are for FastAPI production systems; Laravel remains the reference for *feature completeness*, not the deploy target of this site. ## FILE: docs/guides/14-celery-beat-scheduling.md # Celery Beat scheduling (with FastAPI) **Audience:** Teams running periodic work (cron-like) next to FastAPI + Celery + RabbitMQ. **Scope:** Celery Beat as the scheduler; workers still execute tasks. Beat does **not** run the business logic. ## TL;DR | Piece | Role | |-------|------| | **Celery Beat** | Emits “run this task” messages on a schedule | | **Celery workers** | Consume RabbitMQ and execute tasks | | **Your DB `jobs` table** | Optional but recommended for user-visible / auditable runs | | **Rule** | Run **exactly one** Beat (or use a leader-elected scheduler) | ```text Beat (single) ──schedule tick──► RabbitMQ ──► workers ──► side effects / job rows ``` Without single-leader discipline, every Beat replica will **duplicate** every periodic task. ## Contents 1. [When to use Beat](#1-when-to-use-beat) 2. [Architecture](#2-architecture) 3. [Schedule styles](#3-schedule-styles) 4. [Configuration pseudocode](#4-configuration-pseudocode) 5. [Periodic task design](#5-periodic-task-design) 6. [One Beat only (HA)](#6-one-beat-only-ha) 7. [Timezone and DST](#7-timezone-and-dst) 8. [Missed runs and catch-up](#8-missed-runs-and-catch-up) 9. [Observability](#9-observability) 10. [Alternatives](#10-alternatives) 11. [Anti-patterns](#11-anti-patterns) 12. [Checklist](#12-checklist) --- ## 1. When to use Beat | Fit | Example | |-----|---------| | Fixed interval | Every 5 minutes: sync stale cache, poll vendor | | Cron calendar | Nightly invoice settle at 02:15 | | Business day rules | Weekdays 09:00 report (careful with TZ) | | Not a fit | “Run once when user clicks” → normal enqueue | | Not a fit | Multi-consumer domain facts → Kafka events | Prefer **enqueue a job** from Beat (small message) over putting heavy work inside the Beat process. Beat should stay thin. --- ## 2. Architecture ```text ┌─────────────┐ schedule ┌──────────┐ tasks ┌──────────┐ │ Celery Beat │ ────────────────► │ RabbitMQ │ ─────────────► │ workers │ │ (1 leader) │ │ jobs.* │ │ N pods │ └─────────────┘ └──────────┘ └──────────┘ │ │ │ never shares process with API │ ▼ ▼ separate container/service updates jobs / DB ``` | Deployable | Count | |------------|-------| | API | N | | Worker | N (scale on depth) | | Beat | **1** logical leader | | Flower | 0–1 (private) | Same Docker image, different `CMD`: ```bash # worker celery -A app.workers.celery_app.celery_app worker -Q jobs.default,jobs.io -c 4 # beat (single replica) celery -A app.workers.celery_app.celery_app beat -l info ``` --- ## 3. Schedule styles ### Interval ```python # every 300 seconds "sync-vendor": { "task": "app.workers.tasks.sync.pull_vendor", "schedule": 300.0, } ``` ### Crontab ```python from celery.schedules import crontab "nightly-settle": { "task": "app.workers.tasks.billing.settle_day", "schedule": crontab(hour=2, minute=15), "options": {"queue": "jobs.heavy"}, } ``` ### Solar / custom schedules Rare; prefer crontab + explicit TZ. Custom schedules must be serializable if you use persistent schedulers. ### Static vs database schedule | Approach | Pros | Cons | |----------|------|------| | `beat_schedule` in code | Simple, reviewed in PRs | Deploy to change | | `django-celery-beat` style DB | Ops can edit without deploy | Extra tables; still need one Beat | | RedBeat (Redis) | Dynamic, HA-friendly lock | Redis becomes scheduler dependency | For this checklist (RabbitMQ jobs, Redis not primary job broker), **code schedule + one Beat** is the default. RedBeat is acceptable if you already operate Redis for cache/locks and want multi-instance Beat with locking. --- ## 4. Configuration pseudocode ```python # PSEUDOCODE — app/workers/celery_app.py from celery import Celery from celery.schedules import crontab from app.core.settings import settings celery_app = Celery("app", broker=settings.celery_broker_url) celery_app.conf.update( timezone="UTC", # store and think in UTC enable_utc=True, task_serializer="json", accept_content=["json"], beat_schedule={ "reconcile-outbox-every-minute": { "task": "app.workers.tasks.outbox.relay_tick", "schedule": 60.0, "options": {"queue": "jobs.default"}, }, "daily-prune-jobs": { "task": "app.workers.tasks.maintenance.prune_old_jobs", "schedule": crontab(hour=3, minute=0), # 03:00 UTC "options": {"queue": "jobs.default"}, }, "weekday-report": { "task": "app.workers.tasks.reports.daily_summary", "schedule": crontab(hour=8, minute=0, day_of_week="1-5"), "args": (), # prefer no fat args "options": {"queue": "jobs.io"}, }, }, beat_max_loop_interval=5, # how often Beat wakes to check schedule ) ``` ```python # PSEUDOCODE — settings class Settings(BaseSettings): celery_broker_url: str celery_beat_enabled: bool = True # disable in most containers ``` Only the Beat service should run the `beat` command. Do not start Beat inside the API process. --- ## 5. Periodic task design ### Thin scheduled tasks ```python # PSEUDOCODE — good: schedule enqueues unit of work @celery_app.task(name="app.workers.tasks.billing.settle_day") def settle_day(): # optional: create a jobs row for audit/UI job_id = create_job(type="settle_day", status="pending") run_settle(job_id) # or split: only enqueue children ``` Better for large work: ```python # PSEUDOCODE @celery_app.task(name="billing.settle_day_fanout") def settle_day_fanout(): for merchant_id in list_merchants_due(): settle_merchant.delay(merchant_id) # many small tasks ``` ### Idempotency (mandatory) Schedules **will** double-fire under restarts, clock skew, or accidental multi-Beat. ```python # PSEUDOCODE @celery_app.task(name="reports.daily_summary") def daily_summary(): day = utc_today_iso() key = f"daily_summary:{day}" if not acquire_once(key, ttl=36 * 3600): # Redis lock or UNIQUE job key return # already ran generate_summary(day) ``` Or natural key: ```sql UNIQUE (type, period_key) -- type='daily_summary', period_key='2026-08-05' ``` ### Don’t pass huge args on the schedule Beat serializes schedule args into the message. Pass **IDs or dates**, load data in the worker. ### Queue routing Heavy periodic jobs → `jobs.heavy`. Don’t block `jobs.high` with nightly analytics. --- ## 6. One Beat only (HA) | Strategy | How | |----------|-----| | **Single replica** | K8s `Deployment` replicas=1, or Compose one `beat` service | | **Leader election** | RedBeat / custom lock: only leader sends | | **External cron** | K8s `CronJob` or system cron HTTP-calls `POST /internal/tick` (auth!) | ```yaml # PSEUDOCODE k8s idea # Deployment beat: replicas: 1 # PodDisruptionBudget optional; prefer schedule tolerance over multi-beat ``` **Never** set `replicas: 3` on Beat without a distributed lock. You will get triple charges, triple emails, triple settles. ### API-triggered “schedule” For some teams, replace Beat with: ```text Cloud scheduler / CronJob → POST /internal/schedules/daily-summary → enqueue task (same as Beat would) ``` Still enforce idempotency keys. Protect the endpoint (mTLS, network policy, shared secret). --- ## 7. Timezone and DST - Prefer **`timezone="UTC"`** and `enable_utc=True` - Convert in the product UI for humans - Cron in local time + DST is a common footgun (“runs twice” / “skips an hour”) - Document “03:00 UTC = X local” in runbooks ```python # PSEUDOCODE — business local window computed inside task @celery_app.task def market_open_tick(): if not is_market_open(zone="America/New_York"): return ... ``` --- ## 8. Missed runs and catch-up If Beat was down during a tick: | Behavior | Notes | |----------|-------| | Default Celery Beat | Generally does **not** replay all missed crons like a wall-clock catch-up engine | | Design for drift | Tasks should be safe if run late or skipped | | Critical periods | Reconcile job: “ensure day D settled” runnable ad hoc | ```python # PSEUDOCODE — reconcilable periodic work def settle_day(day: str | None = None): day = day or utc_yesterday() if already_settled(day): return do_settle(day) ``` Ops can call `settle_day.delay("2026-08-04")` after an outage. --- ## 9. Observability | Signal | Why | |--------|-----| | Beat process up | Liveness; alert if down > 2 minutes | | Last tick timestamp | Custom metric/heartbeat task every minute | | Periodic task success rate | Per `task_name` | | Duplicate suppress count | Lock / unique key hits | | Queue depth after fan-out | Nightly storms | Heartbeat pattern: ```python # PSEUDOCODE — schedule every 60s @celery_app.task(name="ops.beat_heartbeat") def beat_heartbeat(): metrics.gauge("beat_heartbeat_unixtime", time.time()) ``` Alert if `now - beat_heartbeat_unixtime > 180s`. --- ## 10. Alternatives | Tool | Use when | |------|----------| | **Celery Beat** | Already on Celery + need in-process schedules | | **RedBeat** | Need dynamic schedules + multi-instance Beat with Redis lock | | **K8s CronJob** | Platform schedules HTTP/enqueue; fewer moving parts in app | | **Taskiq scheduler** | Taskiq stack instead of Celery | | **APScheduler in API** | Avoid for multi-replica APIs (duplicate fires) | | **Kafka + time windows** | Stream processing, not simple cron | APScheduler embedded in each API replica is the FastAPI analogue of “accidentally multi-Beat.” --- ## 11. Anti-patterns | Anti-pattern | Fix | |--------------|-----| | Beat + worker + API in one container | Split processes | | `replicas: 3` on Beat, no lock | Single leader or RedBeat | | Heavy CPU inside Beat process | Beat only dispatches | | Non-idempotent daily charge task | Unique period key + lock | | Local TZ crons without DST review | UTC + explicit business rules | | Silent miss after Beat outage | Reconcile commands + alerts | | Schedule args with secrets | Settings/vault in worker | --- ## 12. Checklist - [ ] Beat is a **separate** deployable from API and workers - [ ] Exactly one leader (replicas=1 or distributed lock) - [ ] `timezone` documented (prefer UTC) - [ ] Periodic tasks **idempotent** (period key / lock) - [ ] Heavy work fanned out to workers/queues - [ ] JSON tasks only; no pickle - [ ] Heartbeat metric + alert if Beat silent - [ ] Runbook: re-run missed day (`settle_day(day=...)`) - [ ] Graceful deploy: workers drain; Beat restart acceptable with idempotency - [ ] Flower/RMQ UI private; schedules reviewed in code review or controlled admin --- ## Pseudocode: full minimal setup ```python # PSEUDOCODE — tasks/maintenance.py @celery_app.task(name="maintenance.prune_old_jobs") def prune_old_jobs(days: int = 30): deleted = db.execute( "DELETE FROM jobs WHERE status IN ('succeeded','failed') " "AND finished_at < now() - interval '%s days'" % days ) # use bound params in real code log.info("pruned", extra={"deleted": deleted}) # celery_app.conf.beat_schedule entry points at this task daily ``` ```bash # production processes celery -A app.workers.celery_app.celery_app worker -Q jobs.default,jobs.io,jobs.heavy celery -A app.workers.celery_app.celery_app beat -l info ``` --- ## See also - [05 Celery + RabbitMQ](05-celery-rabbitmq-example.md) - [03 Job lifecycle](03-job-lifecycle-and-schema.md) - [09 Errors / retries](09-error-taxonomy-retries-dlq.md) - [13 Laravel queues map](13-laravel-queues-to-fastapi.md) (schedule ≈ cron / Task scheduling) ## External docs - [Celery periodic tasks (Beat)](https://docs.celeryq.dev/en/stable/userguide/periodic-tasks.html) - [crontab schedules](https://docs.celeryq.dev/en/stable/userguide/periodic-tasks.html#crontab-schedules) ## FILE: docs/guides/15-observability-metrics-flower.md # Observability, metrics, and Flower (FastAPI + Celery) **Audience:** Teams shipping FastAPI + Celery + RabbitMQ who need production visibility. **Golden rule:** Flower is an **ops** tool. **Product job status** always comes from your application **`jobs` table** (or equivalent), never from Flower or the broker UI. ## TL;DR | Layer | What to do | |-------|------------| | Health | `GET /health` (liveness) + `GET /ready` (readiness) on the API | | Logs | Structured **JSON** to stdout; propagate `job_id` / `event_id` / request id | | Metrics | Enqueue rate, success %, retries, runtime, **queue depth** (backlog) | | SLOs | **Time-to-start** and **time-to-complete** for critical job types | | Flower | Optional Celery dashboard — **private network + auth + TLS** | | Alerts | Zero consumers, DLQ depth, success drop, retry spike, SLO breach | ```text Client ──► FastAPI ──► jobs row + publish ──► RabbitMQ ──► workers │ │ │ logs/metrics │ logs/metrics + Flower (ops) ▼ ▼ GET /jobs/{id} ◄── product truth ── application DB ``` ## Contents 1. [API health endpoints](#1-api-health-endpoints) 2. [Structured logging and correlation](#2-structured-logging-and-correlation) 3. [Core job metrics](#3-core-job-metrics) 4. [SLOs](#4-slos) 5. [Celery + Flower setup](#5-celery--flower-setup) 6. [Securing admin UIs](#6-securing-admin-uis) 7. [What to alert on](#7-what-to-alert-on) 8. [Worker and API instrumentation pseudocode](#8-worker-and-api-instrumentation-pseudocode) 9. [Dashboard sketch](#9-dashboard-sketch) 10. [Anti-patterns](#10-anti-patterns) 11. [Checklist](#11-checklist) --- ## 1. API health endpoints | Endpoint | Purpose | Typical checks | |----------|---------|----------------| | `GET /health` | Liveness | Process up; return 200 quickly | | `GET /ready` | Readiness | DB reachable; (optional) can publish or settings loaded | **Usually do not** make API readiness depend on “workers are healthy.” Scale and restart workers independently. Optionally expose a **worker** readiness that checks broker connectivity for the worker deployment only. ```python # PSEUDOCODE @app.get("/health") async def health(): return {"status": "ok"} @app.get("/ready") async def ready(db: Db = Depends()): await db.execute("SELECT 1") return {"status": "ready"} ``` --- ## 2. Structured logging and correlation Use **JSON logs** to stdout (platform scrapes them). ### Fields to include | Field | Where | |-------|--------| | `timestamp`, `level`, `message` | Always | | `service` (`api` / `worker` / `beat`) | Always | | `request_id` | API middleware | | `job_id` | Enqueue + every worker log line | | `task_id` | Celery task id when present | | `event_id` | Domain/event consumers | | `queue`, `task_name` | Workers | | `attempts` | On retry/fail | ```python # PSEUDOCODE — API enqueue log.info( "job_enqueued", extra={ "job_id": str(job.id), "job_type": job.type, "queue": "jobs.io", "request_id": request_id, }, ) # PSEUDOCODE — worker log.info( "job_started", extra={"job_id": job_id, "task_id": self.request.id, "attempts": self.request.retries}, ) ``` Propagate ids: HTTP header `X-Request-Id` → log context → message headers → worker logging context. Optional: OpenTelemetry trace context in message headers for API → worker spans. --- ## 3. Core job metrics Track at least: | Metric | Type | Notes | |--------|------|-------| | `jobs_enqueued_total` | Counter | By `type`, `queue` | | `jobs_succeeded_total` | Counter | By `type` | | `jobs_failed_total` | Counter | By `type`, `reason` | | `jobs_retries_total` | Counter | By `type` | | `job_runtime_seconds` | Histogram | By `type` | | `job_time_to_start_seconds` | Histogram | `started_at - created_at` | | `job_time_to_complete_seconds` | Histogram | `finished_at - created_at` | | `rabbitmq_queue_depth` | Gauge | Backlog per queue | | `rabbitmq_dlq_depth` | Gauge | Per DLQ | | `rabbitmq_consumers` | Gauge | Per queue | **Success rate** ≈ `succeeded / (succeeded + failed)` over a window (exclude still-running). **Backlog** = queue depth (and/or lag if you also use Kafka). Export via RabbitMQ exporter or management API scraper. --- ## 4. SLOs Define per **critical job type** (not one global number): | SLO | Example | |-----|---------| | Time-to-start | p95 < 30s for `send_receipt` | | Time-to-complete | p95 < 2m for `send_receipt`; p95 < 30m for `export_csv` | Measure from the **`jobs` table** timestamps so product and ops share the same truth. ```text time_to_start = started_at - created_at time_to_complete = finished_at - created_at ``` --- ## 5. Celery + Flower setup ### Workers and Beat (reminder) ```bash celery -A app.workers.celery_app.celery_app worker -Q jobs.default,jobs.io -c 4 celery -A app.workers.celery_app.celery_app beat -l info # single leader ``` ### Flower ```bash # PSEUDOCODE — never expose publicly without auth celery -A app.workers.celery_app.celery_app flower \ --port=5555 \ --basic_auth=ops_user:strong_password ``` Compose sketch: ```yaml # PSEUDOCODE services: flower: image: your-app:tag command: celery -A app.workers.celery_app.celery_app flower --port=5555 # no public ports in prod; private network only environment: CELERY_BROKER_URL: ${CELERY_BROKER_URL} ``` ### What Flower is good for - See active tasks, workers online, basic task history - Debug “is any worker connected?” - Inspect failures during incidents ### What Flower is **not** - Not the customer “is my export done?” API - Not the source of truth for SLAs (use DB timestamps + metrics) - Not a substitute for RabbitMQ queue depth / DLQ alerts --- ## 6. Securing admin UIs **Mandatory for production:** | UI | Controls | |----|----------| | Flower | Private network / VPN / mesh; **auth**; **TLS** at proxy | | RabbitMQ Management | Same | | Kafka UI | Same | Checklist: - [ ] No public `0.0.0.0` without auth - [ ] SSO or strong basic auth + network policy - [ ] TLS terminate at reverse proxy - [ ] Separate credentials from app DB users - [ ] Audit who can access ops UIs --- ## 7. What to alert on | Signal | Severity | Notes | |--------|----------|-------| | **Zero consumers** on a critical queue | **Critical** | Nothing is processing | | **DLQ depth > 0** (critical queues) | **High** | Poison or repeated failure | | **Success rate drop** | **High** | Compare to baseline window | | **Retry rate spike** | **Medium** | Upstream flapping / overload | | **Time-to-complete SLO breach** | **Medium** | p95/p99 over SLO | | Beat heartbeat missing | **High** | Schedules stopped ([14](14-celery-beat-scheduling.md)) | | API 5xx / latency | **High** | Standard RED metrics | Page on Critical/High; ticket or daytime on Medium unless prolonged. --- ## 8. Worker and API instrumentation pseudocode ```python # PSEUDOCODE — enqueue async def enqueue(...): job = await db.insert_job(...) await bus.publish(queue, {"job_id": str(job.id)}) metrics.incr("jobs_enqueued_total", tags={"type": job.type, "queue": queue}) log.info("job_enqueued", extra={"job_id": str(job.id), "request_id": rid}) return job # PSEUDOCODE — worker @celery_app.task(bind=True) def send_receipt(self, job_id: str): t0 = time.perf_counter() log.info("job_started", extra={"job_id": job_id, "task_id": self.request.id}) try: mark_running(job_id) do_work(job_id) mark_succeeded(job_id) metrics.incr("jobs_succeeded_total", tags={"type": "send_receipt"}) except TransientError: metrics.incr("jobs_retries_total", tags={"type": "send_receipt"}) raise except Exception: mark_failed(job_id, ...) metrics.incr("jobs_failed_total", tags={"type": "send_receipt"}) raise finally: metrics.observe("job_runtime_seconds", time.perf_counter() - t0, tags={"type": "send_receipt"}) ``` Export Prometheus (or your vendor) from API and workers; scrape RabbitMQ exporter for depth/consumers. --- ## 9. Dashboard sketch **Row 1 — Traffic:** enqueue rate, success rate, retry rate **Row 2 — Latency:** time-to-start p95, time-to-complete p95 (by type) **Row 3 — Broker:** depth per queue, DLQ depth, consumer count **Row 4 — Workers:** process count, task runtime histogram **Row 5 — Dependencies:** DB errors, SMTP/HTTP client errors Link runbooks: scale workers, redrive DLQ, restart Beat. --- ## 10. Anti-patterns | Anti-pattern | Fix | |--------------|-----| | Users poll Flower for status | `GET /jobs/{id}` from DB | | Flower on the public internet | Private + auth + TLS | | Only metrics, no `job_id` in logs | Structured correlation | | Alert on depth alone with no consumer check | Alert **zero consumers** too | | One SLO for all job types | SLOs per critical type | | Health check that requires workers | Keep API liveness independent | --- ## 11. Checklist ### Observability - [ ] `GET /health` and `GET /ready` on API - [ ] Structured JSON logs on API, worker, Beat - [ ] `request_id` / `job_id` / `event_id` propagated end-to-end - [ ] Counters: enqueue, success, fail, retries - [ ] Histograms: runtime, time-to-start, time-to-complete - [ ] Gauges: queue depth, DLQ depth, consumer count - [ ] SLOs defined for critical job types ### Flower and admin UIs - [ ] Flower optional; command documented - [ ] Flower on private network - [ ] Authentication enabled - [ ] TLS at the edge - [ ] Same controls for RabbitMQ Management / Kafka UI - [ ] Documented: Flower ≠ product status ### Alerts - [ ] Zero consumers → Critical - [ ] DLQ depth > 0 (critical queues) → High - [ ] Success rate drop → High - [ ] Retry spike → Medium - [ ] Time-to-complete SLO breach → Medium --- **Deep dive:** [18 Monitoring Celery with Flower](18-monitoring-celery-flower.md) **Prometheus:** [19 Integrate Flower with Prometheus](19-flower-prometheus.md) ## See also - [05 Celery + RabbitMQ](05-celery-rabbitmq-example.md) - [09 Errors, retries, DLQ](09-error-taxonomy-retries-dlq.md) - [14 Celery Beat](14-celery-beat-scheduling.md) - [03 Job lifecycle](03-job-lifecycle-and-schema.md) (status in DB) - [10 Testing](10-testing-workers.md) ## External - [Flower](https://flower.readthedocs.io/) - [Celery monitoring](https://docs.celeryq.dev/en/stable/userguide/monitoring.html) - [OpenTelemetry](https://opentelemetry.io/docs/) ## FILE: docs/guides/16-locks-and-rate-limits.md # Job locks and rate limits (FastAPI + workers) **Audience:** Teams that need exclusive processing, API abuse protection, and fair use of third-party APIs from workers. **Stack:** FastAPI API + Celery/Taskiq/Dramatiq workers + RabbitMQ (jobs) + Redis as **lock/rate-limit store** (not the job broker). ## TL;DR | Need | Tool | |------|------| | Only one worker runs work for key `X` | **Distributed lock** (Redis/DB) or unique job key | | Protect HTTP endpoints | **API rate limit** (gateway or app middleware) | | Limit how fast jobs hit a vendor API | **Job rate limit** (token bucket / Celery rate / middleware) | | “Latest wins” for noisy updates | Debounce + lock (see also unique jobs) | ```text API rate limit ──► enqueue (maybe) ──► RabbitMQ │ job rate limit + lock ──► side effect ``` Redis here is for **locks and counters**. Durable job transport stays on **RabbitMQ**. ## Contents 1. [When you need a lock](#1-when-you-need-a-lock) 2. [Lock patterns](#2-lock-patterns) 3. [Unique jobs vs locks](#3-unique-jobs-vs-locks) 4. [API rate limiting](#4-api-rate-limiting) 5. [Job rate limiting](#5-job-rate-limiting) 6. [Pseudocode catalog](#6-pseudocode-catalog) 7. [Failure modes](#7-failure-modes) 8. [Metrics and alerts](#8-metrics-and-alerts) 9. [Checklist](#9-checklist) 10. [Anti-patterns](#10-anti-patterns) --- ## 1. When you need a lock | Situation | Without lock | With lock | |-----------|--------------|-----------| | Update credit score for `user_id` | Concurrent overwrites | One job at a time per user | | Reindex product `sku` | Duplicate full reindexes | Skip or wait | | Ledger post for `account_id` | Double post risk | Serialize per account | | Multi-step import for `import_id` | Parallel steps corrupt state | Exclusive run | **Do not** lock the entire global queue. Lock the **smallest key** that protects the invariant (`user_id`, `order_id`, `tenant_id+resource`). --- ## 2. Lock patterns ### A. Redis atomic lock (common) ```python # PSEUDOCODE — SET key value NX EX ttl async def acquire_lock(redis, key: str, token: str, ttl_sec: int) -> bool: return await redis.set(key, token, nx=True, ex=ttl_sec) async def release_lock(redis, key: str, token: str) -> None: # Lua: delete only if value == token (avoid deleting someone else's lock) await redis.eval(RELEASE_LUA, keys=[key], args=[token]) ``` ```python # PSEUDOCODE — worker @celery_app.task(bind=True, max_retries=10) def update_score(self, job_id: str, user_id: str): token = new_uuid() key = f"lock:score:{user_id}" if not acquire_lock(redis, key, token, ttl_sec=120): # another worker holds it — retry later (job rate / overlap) raise self.retry(countdown=15) try: mark_running(job_id) do_update(user_id) mark_succeeded(job_id) finally: release_lock(redis, key, token) ``` **TTL must exceed** worst-case work time, or use renewal (watchdog) for long jobs. ### B. Database lock / advisory lock ```sql -- PSEUDOCODE — Postgres advisory lock SELECT pg_try_advisory_lock(hashtext(:user_id)); ``` Good when Redis is unavailable and DB is already the system of record. Prefer short critical sections. ### C. “Claim” row in jobs table ```sql -- PSEUDOCODE UPDATE jobs SET status = 'running', locked_by = :worker, locked_at = now() WHERE id = :id AND status = 'pending' RETURNING *; ``` Serializes **that job**; does not by itself prevent two different jobs for the same `user_id`. Combine with a uniqueness key or entity lock. ### D. WithoutOverlapping (Laravel-style) Same as Redis lock keyed by entity. Release or retry with delay when not acquired (see [13 Laravel map](13-laravel-queues-to-fastapi.md)). --- ## 3. Unique jobs vs locks | Mechanism | Prevents | Does not prevent | |-----------|----------|------------------| | **Unique / idempotency key** at enqueue | Two identical jobs on the queue | Two different job types on same entity | | **Lock while running** | Concurrent execution for a key | Duplicate enqueues (unless also unique) | | **Both** | Best for “one reindex per product” | — | ```python # PSEUDOCODE — unique enqueue key = f"reindex:{product_id}" job = await db.insert_job_if_absent(idempotency_key=key, ...) if job.already_existed: return job # 202 with existing id await bus.publish(...) ``` ```python # PSEUDOCODE — lock while running (even if two slipped through) lock_key = f"lock:reindex:{product_id}" ``` --- ## 4. API rate limiting Protect the **HTTP edge** before work is enqueued. | Layer | Example | |-------|---------| | API gateway / reverse proxy | nginx, Envoy, cloud WAF | | App middleware | slowapi, custom Redis limiter | | Per-route | Stricter on `/auth/login`, `/jobs` enqueue | ```python # PSEUDOCODE — Redis fixed window (simple) async def allow(redis, key: str, limit: int, window_sec: int) -> bool: n = await redis.incr(key) if n == 1: await redis.expire(key, window_sec) return n <= limit # key examples: # ratelimit:ip:{ip} # ratelimit:user:{user_id}:enqueue ``` ```python # PSEUDOCODE — FastAPI dependency async def rate_limit_enqueue(user=Depends(auth), redis=Depends(get_redis)): ok = await allow(redis, f"rl:enqueue:{user.id}", limit=30, window_sec=60) if not ok: raise HTTPException(429, "Too many job submissions") ``` **Multi-worker API:** shared Redis (or gateway) so limits are global, not per process. **Response:** `429` + `Retry-After` when possible. --- ## 5. Job rate limiting Workers must not stampede third parties (email, SMS, payment, OpenAI, etc.). ### Strategies | Strategy | How | |----------|-----| | **Token bucket / leaky bucket in Redis** | Before side effect, take token or `retry(countdown=…)` | | **Celery `rate_limit`** | e.g. `10/m` on the task decorator | | **Dedicated slow queue + low concurrency** | `jobs.io` with `-c 2` | | **Prefetch = 1** | Fairness for long tasks | | **Per-tenant keys** | `rl:vendor:tenant:{id}` so one tenant cannot starve others | ```python # PSEUDOCODE — Celery built-in (process-local-ish; prefer Redis for multi-worker fairness) @celery_app.task(rate_limit="30/m") def send_email(job_id: str): ... ``` ```python # PSEUDOCODE — Redis token bucket shared across workers async def take_token(redis, key: str, rate_per_sec: float, burst: int) -> bool: # classic token bucket Lua or redis-cell / redis-py bucket ... @celery_app.task(bind=True, max_retries=20) def call_vendor(self, job_id: str, tenant_id: str): if not take_token(redis, f"rl:vendor:{tenant_id}", rate_per_sec=5, burst=10): raise self.retry(countdown=2) do_http_call(...) ``` ### Enqueue vs execute limits | Limit | Protects | |-------|----------| | API enqueue RL | Your DB/broker from floods | | Job execute RL | Downstream vendors | | Both | Production default for public “start job” APIs | --- ## 6. Pseudocode catalog ### Lock + rate limit together ```python # PSEUDOCODE @celery_app.task(bind=True, max_retries=25) def sync_account(self, job_id: str, account_id: str): if not take_token(redis, "rl:sync-global", rate_per_sec=20, burst=40): raise self.retry(countdown=1) token = new_uuid() lock_key = f"lock:account:{account_id}" if not acquire_lock(redis, lock_key, token, ttl_sec=300): raise self.retry(countdown=10) try: if job_already_succeeded(job_id): return mark_running(job_id) run_sync(account_id) mark_succeeded(job_id) except TransientError as e: mark_retry(job_id, str(e)) raise self.retry(countdown=backoff(self.request.retries)) except PermanentError as e: mark_failed(job_id, str(e)) finally: release_lock(redis, lock_key, token) ``` ### Safe lock release (Lua sketch) ```lua -- RELEASE_LUA if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end ``` ### Lock renewal for long jobs ```python # PSEUDOCODE async def with_lock(key, ttl=60): token = new_uuid() if not acquire(key, token, ttl): raise LockBusy() renew = asyncio.create_task(renew_loop(key, token, ttl)) try: yield finally: renew.cancel() release(key, token) ``` --- ## 7. Failure modes | Failure | Symptom | Mitigation | |---------|---------|------------| | Lock TTL too short | Two workers in critical section | TTL > max runtime or renew | | Lock TTL too long + crash | Key stuck, work stalls | TTL bound; fencing token; admin unlock runbook | | Retry without backoff on lock busy | Hot loop | `countdown` + jitter | | API RL only in-memory | Ineffective with N API pods | Shared Redis / gateway | | Celery `rate_limit` only | Uneven multi-worker | Redis bucket for global fairness | | Lock key too coarse | Throughput collapse | Narrower keys | | Lock key missing tenant | Cross-tenant blocking | Include `tenant_id` in key | --- ## 8. Metrics and alerts | Metric | Use | |--------|-----| | `lock_acquire_fail_total` | Contention | | `lock_hold_seconds` | TTL tuning | | `rate_limit_requeue_total` | Job RL pressure | | `api_429_total` | Client abuse / limits too tight | | `job_time_to_start` | Locks/RL delaying start | Alert when lock fail rate or RL requeues spike with backlog growth (possible under-capacity or deadlocks). --- ## 9. Checklist ### Locks - [ ] Identify resources that need exclusive access (document keys) - [ ] Use **smallest** lock key that protects the invariant - [ ] Redis/DB lock with **token-safe release** - [ ] TTL ≥ work time or active renewal - [ ] On lock busy: **retry with delay**, not busy-spin - [ ] Combine with **idempotent** handlers - [ ] Unique/idempotency key at enqueue where duplicates are useless ### API rate limits - [ ] Global and per-IP (and per-user) limits on public APIs - [ ] Stricter limits on auth and **enqueue** endpoints - [ ] Shared store across API replicas - [ ] `429` + `Retry-After` where applicable ### Job rate limits - [ ] Per-vendor / per-tenant limits for outbound calls - [ ] Shared limiter across all workers (Redis) - [ ] Low-concurrency queues for heavy/IO-bound classes - [ ] Prefetch tuned so rate limits are not bypassed by huge in-flight batches ### Ops - [ ] Metrics for lock contention and RL requeues - [ ] Runbook: clear stuck lock (with fencing discipline) - [ ] Redis HA if locks/RL depend on it; app degrades safely if Redis down (fail closed on enqueue RL) --- ## 10. Anti-patterns | Avoid | Prefer | |-------|--------| | Global `lock:all-jobs` | Per-entity keys | | Infinite wait for lock | Bounded retries → fail/DLQ | | In-process locks only | Distributed locks for multi-worker | | Rate limit only at API, workers blast vendor | Job RL too | | Using RabbitMQ as a lock service | Redis/DB locks | | Storing lock state only in memory of one worker | Shared store | --- ## Celery implementation Step-by-step with FastAPI + Celery + Redis helpers: **[17 Celery RL + lock pipeline](17-celery-rate-limit-lock-pipeline.md)** ## See also - [09 Errors / retries / DLQ](09-error-taxonomy-retries-dlq.md) - [03 Job lifecycle](03-job-lifecycle-and-schema.md) - [13 Laravel queues map](13-laravel-queues-to-fastapi.md) (WithoutOverlapping, RateLimited) - [15 Observability & Flower](15-observability-metrics-flower.md) - [05 Celery example](05-celery-rabbitmq-example.md) ## External concepts - Redis `SET NX EX` / Redlock (know the tradeoffs) - Token bucket rate limiting - Postgres advisory locks - Celery [rate limits](https://docs.celeryq.dev/en/stable/userguide/tasks.html#Task.rate_limit) ## FILE: docs/guides/17-celery-rate-limit-lock-pipeline.md # Implement with Celery: API RL → enqueue → RabbitMQ → job RL + lock → side effect **Audience:** FastAPI + Celery + RabbitMQ teams implementing the standard pipeline. **Related:** [16 Locks and rate limits](16-locks-and-rate-limits.md) · [05 Celery example](05-celery-rabbitmq-example.md) ## Pipeline ```text Client │ ▼ FastAPI ── API rate limit (Redis) ──► 429 or continue │ ├── insert jobs row (pending) └── Celery apply_async ──► RabbitMQ queue │ ▼ Celery worker │ ┌─────────────┼─────────────┐ ▼ ▼ ▼ job rate limit entity lock load job (Redis) (Redis) (DB) │ │ │ └─────────────┴──────► side effect │ mark succeeded / failed ``` | Stage | Where | Store | |-------|--------|--------| | API rate limit | FastAPI dependency / middleware | Redis | | Enqueue | FastAPI route | Postgres `jobs` + RabbitMQ via Celery | | Job rate limit | Celery task (before work) | Redis (shared across workers) | | Entity lock | Celery task | Redis `SET NX EX` + token release | | Side effect | Celery task | Vendor / DB | | Product status | Always | **`jobs` table**, not Flower | --- ## 1. Dependencies and settings ```python # PSEUDOCODE — requirements # celery[redis] # redis extra only if you use Redis result backend; broker is RabbitMQ # redis # fastapi, pydantic-settings, sqlalchemy/asyncpg, ... # PSEUDOCODE — settings class Settings(BaseSettings): database_url: str celery_broker_url: str = "amqps://user:pass@rabbitmq:5671//" redis_url: str = "redis://redis:6379/0" # locks + rate limits only api_enqueue_limit: int = 30 api_enqueue_window_sec: int = 60 vendor_rate_per_sec: float = 5.0 vendor_burst: int = 10 lock_ttl_sec: int = 120 ``` Broker = **RabbitMQ**. Redis is **not** the job broker. --- ## 2. Redis helpers (locks + rate limits) ```python # PSEUDOCODE — app/core/redis_limits.py import time import uuid from redis import Redis RELEASE_LUA = """ if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end """ # Simple token bucket (production: use a well-tested Lua bucket) BUCKET_LUA = """ local key = KEYS[1] local rate = tonumber(ARGV[1]) local burst = tonumber(ARGV[2]) local now = tonumber(ARGV[3]) local data = redis.call("hmget", key, "tokens", "ts") local tokens = tonumber(data[1]) local ts = tonumber(data[2]) if tokens == nil then tokens = burst ts = now end local delta = math.max(0, now - ts) tokens = math.min(burst, tokens + delta * rate) local ok = 0 if tokens >= 1 then tokens = tokens - 1 ok = 1 end redis.call("hmset", key, "tokens", tokens, "ts", now) redis.call("expire", key, 3600) return ok """ def get_redis(url: str) -> Redis: return Redis.from_url(url, decode_responses=True) def api_allow(r: Redis, key: str, limit: int, window_sec: int) -> bool: n = r.incr(key) if n == 1: r.expire(key, window_sec) return n <= limit def take_token(r: Redis, key: str, rate_per_sec: float, burst: int) -> bool: return r.eval(BUCKET_LUA, 1, key, rate_per_sec, burst, time.time()) == 1 def acquire_lock(r: Redis, key: str, token: str, ttl_sec: int) -> bool: return bool(r.set(key, token, nx=True, ex=ttl_sec)) def release_lock(r: Redis, key: str, token: str) -> None: r.eval(RELEASE_LUA, 1, key, token) ``` Use **sync Redis** in Celery prefork workers; use async Redis only if your worker model is async (Taskiq) or you wrap calls carefully. --- ## 3. Celery app (RabbitMQ) ```python # PSEUDOCODE — app/workers/celery_app.py from celery import Celery from app.core.settings import settings celery_app = Celery("app", broker=settings.celery_broker_url) celery_app.conf.update( task_serializer="json", accept_content=["json"], result_serializer="json", task_acks_late=True, worker_prefetch_multiplier=1, task_default_queue="jobs.default", task_routes={ "app.workers.tasks.sync.*": {"queue": "jobs.io"}, }, broker_connection_retry_on_startup=True, ) celery_app.autodiscover_tasks(["app.workers.tasks"]) ``` --- ## 4. FastAPI: API rate limit → enqueue ```python # PSEUDOCODE — app/api/deps.py from fastapi import Depends, HTTPException, Request from app.core.redis_limits import get_redis, api_allow from app.core.settings import settings def get_r(): return get_redis(settings.redis_url) def rate_limit_enqueue( request: Request, user=Depends(get_current_user), r=Depends(get_r), ): # per-user (and optionally also per-IP) key = f"rl:api:enqueue:user:{user.id}" if not api_allow(r, key, settings.api_enqueue_limit, settings.api_enqueue_window_sec): raise HTTPException( status_code=429, detail="Too many job submissions", headers={"Retry-After": str(settings.api_enqueue_window_sec)}, ) ``` ```python # PSEUDOCODE — app/api/routes/sync.py from fastapi import APIRouter, Depends from app.workers.tasks.sync import sync_account from app.api.deps import rate_limit_enqueue, get_db router = APIRouter(prefix="/sync", tags=["sync"]) @router.post("/accounts/{account_id}", status_code=202, dependencies=[Depends(rate_limit_enqueue)]) def start_sync(account_id: str, db=Depends(get_db), user=Depends(get_current_user)): # optional: unique key so double-click does not double-publish idem = f"sync-account:{account_id}:{user.id}" existing = db.find_job_by_idempotency(idem) if existing: return {"job_id": existing.id, "status": existing.status} job = db.insert_job( type="sync_account", status="pending", entity_id=account_id, idempotency_key=idem, ) # Celery → RabbitMQ sync_account.apply_async( args=[str(job.id), account_id], queue="jobs.io", ) return {"job_id": job.id, "status": "pending"} ``` Prefer **outbox** if you need commit-safe publish ([04](04-outbox-inbox-idempotency.md)); `apply_async` right after insert is the simple path. --- ## 5. Celery task: job rate limit + entity lock → side effect ```python # PSEUDOCODE — app/workers/tasks/sync.py import uuid from celery.exceptions import MaxRetriesExceededError from app.workers.celery_app import celery_app from app.core.settings import settings from app.core.redis_limits import ( get_redis, take_token, acquire_lock, release_lock, ) from app.db import session_scope from app.services.vendor import call_vendor_sync @celery_app.task( bind=True, name="app.workers.tasks.sync.sync_account", max_retries=25, acks_late=True, soft_time_limit=90, time_limit=120, ) def sync_account(self, job_id: str, account_id: str) -> None: r = get_redis(settings.redis_url) # --- job / vendor rate limit (shared across all workers) --- rl_key = f"rl:vendor:sync:{account_id}" # or global: rl:vendor:sync if not take_token( r, rl_key, rate_per_sec=settings.vendor_rate_per_sec, burst=settings.vendor_burst, ): # re-queue with delay; does not burn forever if max_retries set raise self.retry(countdown=2 + (self.request.retries % 5)) # --- entity lock (one sync per account at a time) --- token = str(uuid.uuid4()) lock_key = f"lock:account:{account_id}" if not acquire_lock(r, lock_key, token, ttl_sec=settings.lock_ttl_sec): raise self.retry(countdown=10 + (self.request.retries % 10)) try: with session_scope() as db: job = db.get_job(job_id) if job is None: return if job.status == "succeeded": return # idempotent db.mark_running(job_id) try: call_vendor_sync(account_id) # side effect db.mark_succeeded(job_id) except PermanentVendorError as e: db.mark_failed(job_id, str(e)) # do not retry permanent errors return except TransientVendorError as e: db.bump_attempt(job_id, str(e)) raise self.retry( exc=e, countdown=min(600, 2 ** self.request.retries), ) finally: release_lock(r, lock_key, token) ``` ### Why this order? 1. **Rate limit first** — avoid holding a lock while waiting on a vendor quota. 2. **Lock second** — exclusive critical section only while doing real work. 3. **Always release** in `finally` with token check. Optional: take a **global** vendor token and a **per-tenant** token (two `take_token` calls). --- ## 6. Celery `rate_limit` vs Redis bucket | Mechanism | Scope | Use | |-----------|--------|-----| | `@task(rate_limit="30/m")` | Per-worker process (approximate) | Extra safety net | | Redis token bucket | **All workers** | Real multi-worker fairness | ```python # Optional second line of defense (not enough alone with many workers) @celery_app.task(bind=True, rate_limit="30/m") def sync_account(self, job_id: str, account_id: str): ... ``` Prefer **Redis** for production multi-replica workers. --- ## 7. Run processes ```bash # API uvicorn app.main:app --host 0.0.0.0 --port 8000 # Workers consuming the IO queue celery -A app.workers.celery_app.celery_app worker -Q jobs.io,jobs.default -c 4 # Optional Flower (private + auth) — not product status celery -A app.workers.celery_app.celery_app flower --basic_auth=user:pass ``` Compose services: `api`, `worker`, `rabbitmq`, `redis`, `db`. --- ## 8. Testing sketch ```python # PSEUDOCODE def test_enqueue_rate_limited(client, redis): for _ in range(30): assert client.post("/sync/accounts/a1").status_code == 202 assert client.post("/sync/accounts/a1").status_code == 429 def test_task_retries_when_lock_held(redis): acquire_lock(redis, "lock:account:a1", "other", 60) with pytest.raises(Retry): sync_account.push_request(retries=0).run("job-1", "a1") # or mock retry def test_success_releases_lock(redis, db): sync_account.run(str(job.id), "a1") assert redis.get("lock:account:a1") is None assert db.get_job(job.id).status == "succeeded" ``` --- ## 9. Production checklist (Celery-specific) - [ ] Broker URL is RabbitMQ (`amqps://` in prod) - [ ] Redis used only for RL + locks (and optional result backend) - [ ] API enqueue dependency returns **429** when over limit - [ ] `apply_async` / `.delay` after job row insert (or outbox) - [ ] Task uses **acks_late**, JSON, routed queue - [ ] Redis **job** rate limit before side effect - [ ] Entity lock with **token-safe** release in `finally` - [ ] Lock busy / RL miss → `self.retry(countdown=…)` with **max_retries** - [ ] Permanent errors → `mark_failed`, no retry - [ ] Status for users from **`GET /jobs/{id}`** - [ ] Metrics: `api_429`, `lock_busy_retry`, `rl_requeue`, runtime --- ## 10. Minimal file layout ```text app/ core/settings.py core/redis_limits.py api/deps.py api/routes/sync.py workers/celery_app.py workers/tasks/sync.py db/... ``` --- ## See also - [16 Locks and rate limits](16-locks-and-rate-limits.md) — concepts - [05 Celery + RabbitMQ](05-celery-rabbitmq-example.md) — app wiring - [09 Errors / DLQ](09-error-taxonomy-retries-dlq.md) - [15 Observability & Flower](15-observability-metrics-flower.md) - [03 Job lifecycle](03-job-lifecycle-and-schema.md) ## External - [Celery tasks](https://docs.celeryq.dev/en/stable/userguide/tasks.html) - [Celery retry](https://docs.celeryq.dev/en/stable/userguide/tasks.html#retrying) - [Celery rate_limit](https://docs.celeryq.dev/en/stable/userguide/tasks.html#Task.rate_limit) - [Redis SET NX](https://redis.io/commands/set/) ## FILE: docs/guides/18-monitoring-celery-flower.md # Monitoring Celery with Flower **Audience:** Ops and backend engineers running Celery workers on RabbitMQ. **Golden rule:** Flower is an **ops dashboard**. Product job status always comes from your **`jobs` table** / `GET /jobs/{id}` — never from Flower. ## TL;DR | Topic | Guidance | |-------|----------| | What Flower shows | Workers online, active/reserved/scheduled tasks, basic history, rates | | What it does **not** replace | Queue depth/DLQ alerts, app metrics, user-facing job status | | How to run | `celery -A ... flower` as a **separate** process | | Security | **Private network + authentication + TLS** (mandatory in prod) | | Broker | Works with RabbitMQ (this checklist); Redis broker not our job default | ```text Celery workers ◄── events / inspect ──► Flower (ops UI) │ └── still emit metrics/logs to Prometheus/your stack ``` ## Contents 1. [What Flower is for](#1-what-flower-is-for) 2. [What Flower is not for](#2-what-flower-is-not-for) 3. [Install and run](#3-install-and-run) 4. [Essential configuration](#4-essential-configuration) 5. [Security (mandatory)](#5-security-mandatory) 6. [What to watch in the UI](#6-what-to-watch-in-the-ui) 7. [Enable Celery events](#7-enable-celery-events) 8. [Flower + metrics (better together)](#8-flower--metrics-better-together) 9. [Deploy sketch](#9-deploy-sketch) 10. [Troubleshooting](#10-troubleshooting) 11. [Checklist](#11-checklist) --- ## 1. What Flower is for Flower is a real-time web monitor for Celery: - List **workers** (alive, concurrency, queues) - See **active**, **reserved**, **scheduled** tasks - Inspect **task success/failure** history (when events are enabled) - Basic **rates** and task runtime views - Optional **revoke** / shutdown controls (treat as dangerous in prod) Use it during incidents: “Are workers connected? Is a task stuck active? Did failures spike?” --- ## 2. What Flower is not for | Not this | Use this instead | |----------|------------------| | Customer “is my export done?” | `GET /jobs/{id}` from application DB | | Sole source of queue backlog SLOs | RabbitMQ depth / consumer count metrics | | Public status page | Never expose Flower publicly without auth | | Long-term analytics warehouse | Prometheus + logs + job table | | Replace structured logging | JSON logs with `job_id` / `task_id` | See [15 Observability](15-observability-metrics-flower.md). --- ## 3. Install and run ```bash # same app env as workers pip install flower # PSEUDOCODE — module path to Celery app instance celery -A app.workers.celery_app.celery_app flower \ --port=5555 \ --basic_auth=ops_user:strong_password ``` ```python # PSEUDOCODE — app/workers/celery_app.py must be importable from celery import Celery celery_app = Celery("app", broker=settings.celery_broker_url) ``` Compose / process list: ```text api | worker | beat (1) | flower (1, private) | rabbitmq | redis | db ``` --- ## 4. Essential configuration | Flag / setting | Purpose | |----------------|---------| | `--port=5555` | HTTP port (behind internal proxy) | | `--basic_auth=user:pass` | Built-in basic auth (or put auth at proxy/SSO) | | `--broker_api=` | Optional RabbitMQ management API URL for more broker insight | | `--persistent=True` | Persist task state to Flower’s DB (optional; know disk use) | | `--db=flower.db` | Path when persistent | | `--max_tasks=10000` | Cap history size | | `--xheaders` | Trust proxy headers when TLS terminates upstream | | `--url_prefix=flower` | If mounted under a path | ```bash # PSEUDOCODE — richer RabbitMQ view (management plugin + credentials) celery -A app.workers.celery_app.celery_app flower \ --broker_api=https://user:pass@rabbitmq:15672/api/ ``` Environment variables (common): ```bash CELERY_BROKER_URL=amqps://... FLOWER_BASIC_AUTH=ops_user:strong_password FLOWER_PORT=5555 ``` --- ## 5. Security (mandatory) **Never** put Flower on the public internet without controls. | Control | Requirement | |---------|-------------| | Network | Private VPC / cluster network / VPN / mesh only | | Auth | Basic auth **or** SSO at reverse proxy (OAuth2 proxy, etc.) | | TLS | Terminate TLS at ingress/proxy | | Authorization | Ops-only roles; not all developers by default | | Actions | Disable or restrict revoke/shutdown if your policy requires | | Secrets | Flower creds ≠ DB creds; rotate | ```nginx # PSEUDOCODE — internal reverse proxy sketch # listen only on internal LB location /flower/ { auth_request /oauth2/auth; # or htpasswd proxy_pass http://flower:5555/; proxy_set_header Host $host; proxy_set_header X-Forwarded-Proto https; } ``` Same rules for **RabbitMQ Management UI**. --- ## 6. What to watch in the UI ### Workers tab | Signal | Meaning | |--------|---------| | Worker missing | Process crash, deploy, wrong broker URL | | Concurrency | Prefork pool size vs load | | Queues | Is worker bound to `jobs.io` / `jobs.heavy`? | ### Tasks | State | Meaning | |-------|---------| | Active | Running now | | Reserved | Prefetched (with `prefetch_multiplier=1` this stays small) | | Succeeded / Failed | Needs **task events** for reliable history | ### During incidents 1. Any workers online? 2. Active task stuck for too long? (compare to soft_time_limit) 3. Failure rate jumping? 4. Then check RabbitMQ depth/DLQ and app metrics — not Flower alone --- ## 7. Enable Celery events Flower relies on Celery **events** for live task visibility. ```python # PSEUDOCODE — celery config (workers) celery_app.conf.update( worker_send_task_events=True, # workers emit events task_send_sent_event=True, # optional: task-sent events ) ``` ```bash # workers must not disable events celery -A app.workers.celery_app.celery_app worker -E -Q jobs.default,jobs.io # -E is --task-events ``` Without events, Flower may show workers but sparse/empty task history. --- ## 8. Flower + metrics (better together) | Need | Flower | Metrics / logs | |------|--------|----------------| | “Who is online?” | Excellent | Worker up gauge | | Queue depth / DLQ | Weak / secondary | **RabbitMQ exporter** (primary) | | p95 time-to-complete | Approximate | **jobs table** + histograms | | Alerting | Not an alerter | Alertmanager / PagerDuty | | Trace one job_id | Search if present | Structured logs + OTel | **Minimum production set:** 1. Flower (private) for humans 2. Prometheus metrics: enqueue, success, fail, runtime, depth, consumers 3. Alerts: zero consumers, DLQ > 0, success drop, SLO breach 4. JSON logs with `job_id` + `task_id` --- ## 9. Deploy sketch ```yaml # PSEUDOCODE — docker-compose fragment services: flower: image: your-app:${TAG} command: > celery -A app.workers.celery_app.celery_app flower --port=5555 --basic_auth=${FLOWER_BASIC_AUTH} environment: CELERY_BROKER_URL: ${CELERY_BROKER_URL} # no public ports in production — only attach to internal network networks: [internal] depends_on: [rabbitmq] restart: unless-stopped ``` Kubernetes: - `Deployment` replicas: **1** is enough for Flower - `Service` ClusterIP only - Ingress with auth + TLS, or no Ingress (port-forward / VPN) - Resource limits modest (CPU/memory grow with `--persistent` and task volume) --- ## 10. Troubleshooting | Symptom | Checks | |---------|--------| | Empty workers | Broker URL; workers running; same vhost; network policy | | Empty tasks | `-E` / `worker_send_task_events`; clock skew | | Flower OOM | Lower `--max_tasks`; disable or bound persistence | | Stale state | Restart Flower; check broker connectivity | | Auth loops | `url_prefix`, proxy headers, cookie paths | | “It works in Flower but user status wrong” | You’re using Flower as product truth — fix API/DB | --- ## 11. Checklist - [ ] Flower runs as its **own** process/container (not inside API) - [ ] Same Celery app module / broker as workers - [ ] Task events enabled on workers (`-E` / config) - [ ] **Private** network only - [ ] **Authentication** enabled (basic or SSO) - [ ] **TLS** at the edge - [ ] Documented: Flower ≠ user job status - [ ] RabbitMQ depth/DLQ/consumer **metrics + alerts** still configured - [ ] Optional `--broker_api` only over TLS with locked-down creds - [ ] Revoke/admin actions limited by policy - [ ] Runbook link from dashboard (scale workers, redrive DLQ) --- ## Quick commands ```bash # start celery -A app.workers.celery_app.celery_app flower --port=5555 --basic_auth=user:pass # worker with events celery -A app.workers.celery_app.celery_app worker -E -Q jobs.default,jobs.io -c 4 # never: expose 5555 on 0.0.0.0 to the internet without auth ``` --- **Prometheus:** [19 Integrate Flower with Prometheus](19-flower-prometheus.md) ## See also - [15 Observability, metrics, and Flower](15-observability-metrics-flower.md) — health, SLOs, alerts - [05 Celery + RabbitMQ](05-celery-rabbitmq-example.md) - [14 Celery Beat](14-celery-beat-scheduling.md) - [17 Celery RL + lock pipeline](17-celery-rate-limit-lock-pipeline.md) - [09 Errors / DLQ](09-error-taxonomy-retries-dlq.md) ## External - [Flower documentation](https://flower.readthedocs.io/) - [Celery monitoring](https://docs.celeryq.dev/en/stable/userguide/monitoring.html) - [Celery events](https://docs.celeryq.dev/en/stable/userguide/monitoring.html#events) ## FILE: docs/guides/19-flower-prometheus.md # Integrate Flower with Prometheus **Audience:** Teams running Celery + Flower who want scrapeable metrics and alerts. **Policy:** Flower UI stays **private**. Prometheus scrapes an internal metrics endpoint. Product job status still comes from the **`jobs` table**. ## TL;DR | Piece | Role | |-------|------| | **Flower** `/metrics` | Celery-centric Prometheus metrics (workers, tasks) | | **Prometheus** | Scrapes Flower (+ RabbitMQ exporter + app metrics) | | **Grafana / Alertmanager** | Dashboards and pages | | **Not enough alone** | Queue depth/DLQ → **RabbitMQ exporter**; SLOs → **app/job metrics** | ```text Celery workers ──events──► Flower ──/metrics──► Prometheus ──► Grafana / Alertmanager RabbitMQ ──exporter──► rabbitmq-exporter ──/metrics──┘ API/workers ──custom counters/histograms──────────┘ ``` ## Contents 1. [Architecture](#1-architecture) 2. [Enable Flower metrics](#2-enable-flower-metrics) 3. [Prometheus scrape config](#3-prometheus-scrape-config) 4. [Useful Flower metrics](#4-useful-flower-metrics) 5. [What else to scrape](#5-what-else-to-scrape) 6. [Example PromQL](#6-example-promql) 7. [Alert rules](#7-alert-rules) 8. [Grafana dashboard ideas](#8-grafana-dashboard-ideas) 9. [Security](#9-security) 10. [Compose sketch](#10-compose-sketch) 11. [Troubleshooting](#11-troubleshooting) 12. [Checklist](#12-checklist) --- ## 1. Architecture ```text ┌────────────┐ ┌─────────────┐ ┌────────────┐ │ workers │ │ Flower │ │ Prometheus │ │ (-E) │────►│ :5555 │────►│ scrape │ └────────────┘ │ /metrics │ └─────┬──────┘ └─────────────┘ │ ┌────────────┐ ┌─────────────┐ │ │ RabbitMQ │────►│ RMQ exporter│───────────┤ └────────────┘ └─────────────┘ │ ┌────────────┐ ┌─────────────┐ │ │ API/worker │────►│ app /metrics│───────────┘ │ (custom) │ └─────────────┘ └────────────┘ ``` Flower answers: worker liveness, task counts/rates from Celery’s view. RabbitMQ answers: **depth**, **consumers**, **DLQ**. Your app answers: **enqueue**, **job SLOs**, **business outcomes**. --- ## 2. Enable Flower metrics Flower exposes Prometheus metrics when the **Prometheus client** is available and the metrics endpoint is enabled (Flower 1.0+ / 2.x). ```bash pip install flower prometheus-client ``` ```bash # PSEUDOCODE — run Flower (internal only) celery -A app.workers.celery_app.celery_app flower \ --port=5555 \ --basic_auth="${FLOWER_BASIC_AUTH}" \ --address=0.0.0.0 ``` Metrics URL (cluster-internal): ```text http://flower:5555/metrics ``` Workers still need **task events** so Flower sees task activity: ```bash celery -A app.workers.celery_app.celery_app worker -E -Q jobs.default,jobs.io ``` ```python celery_app.conf.update( worker_send_task_events=True, task_send_sent_event=True, ) ``` > **Note:** Metric **names** can vary slightly by Flower version. Hit `/metrics` once and copy the actual series names into dashboards. Below uses common `flower_*` patterns — adjust if your build differs. ### Optional: scrape without basic-auth friction Prefer **network policy** (Prometheus in the same mesh scrapes Flower Service) over disabling auth. If Prometheus must scrape past basic auth: ```yaml # PSEUDOCODE — prometheus.yml scrape with basic auth basic_auth: username: ops_user password: strong_password ``` Or terminate auth at the UI path only and leave `/metrics` on a separate internal listener (if you customize the proxy). Simplest: **ClusterIP + basic_auth in scrape config**. --- ## 3. Prometheus scrape config ```yaml # PSEUDOCODE — prometheus.yml global: scrape_interval: 15s scrape_configs: - job_name: flower metrics_path: /metrics static_configs: - targets: ["flower:5555"] labels: service: celery-flower env: production basic_auth: username: ops_user password_file: /etc/prometheus/flower_password - job_name: rabbitmq static_configs: - targets: ["rabbitmq-exporter:9419"] - job_name: api static_configs: - targets: ["api:8000"] metrics_path: /metrics - job_name: worker-app # if workers expose a small HTTP metrics port static_configs: - targets: ["worker-metrics:9100"] ``` Kubernetes ServiceMonitor sketch: ```yaml # PSEUDOCODE apiVersion: monitoring.coreos.com/v1 kind: ServiceMonitor metadata: name: flower spec: selector: matchLabels: app: flower endpoints: - port: http path: /metrics interval: 15s basicAuth: username: name: flower-auth key: user password: name: flower-auth key: password ``` --- ## 4. Useful Flower metrics Inspect live output: ```bash curl -u ops_user:pass http://flower:5555/metrics | head ``` Common series (verify names on your version): | Metric (typical) | Meaning | |------------------|---------| | `flower_worker_online` / worker up gauges | Worker presence | | `flower_events_total` / task event counters | Event throughput | | `flower_task_runtime_seconds` (histogram/summary) | Task runtime | | Task succeeded/failed counters by `task` label | Outcome rates | | Active task gauges | In-flight work | **Relabel** noisy labels (full args, ids) if cardinality explodes — prefer `task` name, `queue`, `worker`, `state`. --- ## 5. What else to scrape Flower alone **misses** broker truth: | Source | Why | |--------|-----| | **rabbitmq_exporter** / RMQ prometheus plugin | `queue_messages`, consumers, DLQ depth | | **App metrics** on API | `jobs_enqueued_total`, 429s | | **App metrics** on workers | `jobs_succeeded_total`, lock/RL requeues | | **postgres** (optional) | Job table lag queries via exporter/sql exporter | ```text # PSEUDOCODE — app counters (prometheus_client) jobs_enqueued_total{type=} jobs_succeeded_total{type=} jobs_failed_total{type=} job_runtime_seconds{type=} job_time_to_start_seconds{type=} job_time_to_complete_seconds{type=} ``` See [15 Observability](15-observability-metrics-flower.md). --- ## 6. Example PromQL ```promql # PSEUDOCODE — adapt metric names to your /metrics output # Task success rate (5m) by task name sum(rate(flower_task_succeeded_total[5m])) by (task) / clamp_min( sum(rate(flower_task_succeeded_total[5m])) by (task) + sum(rate(flower_task_failed_total[5m])) by (task), 1e-9 ) # Workers online (if gauge exists) sum(flower_worker_online) # RabbitMQ backlog (exporter) sum(rabbitmq_queue_messages{queue=~"jobs\\..*"}) by (queue) # Consumers sum(rabbitmq_queue_consumers{queue=~"jobs\\..*"}) by (queue) # App SLO: p95 time to complete histogram_quantile(0.95, sum(rate(job_time_to_complete_seconds_bucket[10m])) by (le, type)) ``` --- ## 7. Alert rules ```yaml # PSEUDOCODE — alert rules (tune names/thresholds) groups: - name: celery-flower rules: - alert: CeleryWorkersDown expr: sum(flower_worker_online) == 0 for: 2m labels: severity: critical annotations: summary: No Celery workers visible to Flower - alert: RabbitMQZeroConsumers expr: sum(rabbitmq_queue_consumers{queue="jobs.default"}) == 0 for: 2m labels: severity: critical annotations: summary: Zero consumers on jobs.default - alert: RabbitMQDLQNotEmpty expr: sum(rabbitmq_queue_messages{queue=~".*dlq.*"}) > 0 for: 5m labels: severity: high annotations: summary: DLQ has messages - alert: CeleryTaskFailureRateHigh expr: | ( sum(rate(flower_task_failed_total[5m])) / clamp_min(sum(rate(flower_task_succeeded_total[5m])) + sum(rate(flower_task_failed_total[5m])), 1e-9) ) > 0.05 for: 10m labels: severity: high annotations: summary: Task failure rate > 5% - alert: JobQueueBacklogHigh expr: sum(rabbitmq_queue_messages{queue="jobs.default"}) > 1000 for: 10m labels: severity: medium annotations: summary: jobs.default depth high ``` Map to checklist severities: zero consumers **Critical**, DLQ **High**, success drop **High**, SLO **Medium** ([15](15-observability-metrics-flower.md)). --- ## 8. Grafana dashboard ideas **Row 1 — Celery (Flower):** workers online, task success/fail rate, runtime p95 **Row 2 — Broker (RMQ):** depth by queue, consumers, DLQ **Row 3 — App:** enqueue rate, time-to-start/complete p95, 429s **Row 4 — Dependencies:** DB errors, vendor latency Link runbooks: scale workers, redrive DLQ, check Beat, open Flower UI (VPN). --- ## 9. Security | Risk | Control | |------|---------| | Metrics leak task names / hostnames | Internal scrape only; careful labels | | Flower UI exposure | Private + auth + TLS ([18](18-monitoring-celery-flower.md)) | | Scrape credentials | K8s Secret / password_file; not in git | | High cardinality | Drop `task_id` / args from metric labels | Prometheus should scrape **ClusterIP** Flower, not a public Ingress `/metrics`. --- ## 10. Compose sketch ```yaml # PSEUDOCODE services: flower: image: your-app:${TAG} command: > celery -A app.workers.celery_app.celery_app flower --port=5555 --basic_auth=${FLOWER_BASIC_AUTH} environment: CELERY_BROKER_URL: ${CELERY_BROKER_URL} networks: [internal] # expose 5555 only on internal network prometheus: image: prom/prometheus:latest volumes: - ./deploy/prometheus.yml:/etc/prometheus/prometheus.yml:ro - ./deploy/flower_password:/etc/prometheus/flower_password:ro networks: [internal] rabbitmq-exporter: image: kbudde/rabbitmq-exporter:latest environment: RABBIT_URL: http://rabbitmq:15672 RABBIT_USER: monitor RABBIT_PASSWORD: ${RMQ_MONITOR_PASSWORD} networks: [internal] ``` --- ## 11. Troubleshooting | Symptom | Fix | |---------|-----| | `/metrics` 404 | Upgrade Flower; ensure `prometheus-client` installed; check Flower version docs | | Empty/zero series | Workers not using `-E`; Flower not connected to broker | | 401 on scrape | Add `basic_auth` to scrape_config | | Cardinality explosion | Remove high-cardinality labels; lower history | | Workers up in Flower UI but bad alerts | Prefer RMQ consumer metrics as source of truth for “can we drain?” | | Success rate wrong | Combine with app `jobs_*` counters from DB-backed outcomes | --- ## 12. Checklist - [ ] `prometheus-client` installed in Flower environment - [ ] Flower `/metrics` reachable **internally** - [ ] Prometheus scrape job for Flower (auth if required) - [ ] Workers send task events (`-E`) - [ ] RabbitMQ exporter (or plugin) for depth/consumers/DLQ - [ ] App metrics for enqueue + job SLOs - [ ] Alert rules: workers down, zero consumers, DLQ, failure rate, backlog - [ ] Grafana dashboard with Flower + RMQ + app rows - [ ] Flower UI still private; metrics not public - [ ] Document metric name mapping for your Flower version --- ## Quick verify ```bash # internal curl -u ops_user:pass http://flower:5555/metrics | grep -E 'flower_|celery_' | head # prometheus targets UI: flower state = UP ``` --- ## See also - [18 Monitoring Celery with Flower](18-monitoring-celery-flower.md) - [15 Observability, metrics, Flower](15-observability-metrics-flower.md) - [05 Celery + RabbitMQ](05-celery-rabbitmq-example.md) - [09 Errors / DLQ](09-error-taxonomy-retries-dlq.md) ## External - [Flower documentation](https://flower.readthedocs.io/) - [Prometheus scrape config](https://prometheus.io/docs/prometheus/latest/configuration/configuration/) - [Celery monitoring](https://docs.celeryq.dev/en/stable/userguide/monitoring.html) - [RabbitMQ Prometheus plugin](https://www.rabbitmq.com/docs/prometheus) / community exporters - Alternatives if Flower metrics are insufficient: dedicated **celery-exporter** sidecars (still pair with RMQ + app metrics) ## FILE: docs/guides/REFERENCES.md # References Curated links for the FastAPI production checklist and guides. Prefer official docs over random blog posts when implementing. ## FastAPI and Python - [FastAPI documentation](https://fastapi.tiangolo.com/) - [FastAPI Advanced Middleware](https://fastapi.tiangolo.com/tutorial/middleware/) - [FastAPI Background Tasks](https://fastapi.tiangolo.com/tutorial/background-tasks/) - [Pydantic Settings](https://docs.pydantic.dev/latest/concepts/pydantic_settings/) - [SQLAlchemy 2.0 asyncio](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html) - [Alembic](https://alembic.sqlalchemy.org/) ## ASGI and deploy - [Uvicorn](https://www.uvicorn.org/) - [Gunicorn](https://docs.gunicorn.org/) - [Twelve-Factor App](https://12factor.net/) ## Brokers - [RabbitMQ documentation](https://www.rabbitmq.com/docs) - [RabbitMQ reliability / DLX](https://www.rabbitmq.com/docs/dlx) - [Apache Kafka documentation](https://kafka.apache.org/documentation/) - [Kafka consumer groups](https://kafka.apache.org/documentation/#intro_consumers) ## Job and messaging frameworks - [Celery](https://docs.celeryq.dev/) - [Celery task rate_limit](https://docs.celeryq.dev/en/stable/userguide/tasks.html#Task.rate_limit) - [Celery Beat / periodic tasks](https://docs.celeryq.dev/en/stable/userguide/periodic-tasks.html) - [Taskiq](https://taskiq-python.github.io/) - [Dramatiq](https://dramatiq.io/) - [FastStream](https://faststream.airt.ai/) ## Patterns - [Transactional Outbox (Microservices.io)](https://microservices.io/patterns/data/transactional-outbox.html) - [Idempotent Consumer](https://microservices.io/patterns/communication-style/idempotent-consumer.html) - [Domain events](https://martinfowler.com/eaaDev/DomainEvent.html) - [Bounded Context (DDD)](https://martinfowler.com/bliki/BoundedContext.html) ## Security and ops - [OWASP ASVS](https://owasp.org/www-project-application-security-verification-standard/) - [OpenTelemetry](https://opentelemetry.io/docs/) ## This repository (in-app) | Guide | Path | |-------|------| | Async vs jobs | [/docs/01-async-vs-jobs](/docs/01-async-vs-jobs) | | Brokers | [/docs/02-brokers-celery-redis-rabbitmq-kafka](/docs/02-brokers-celery-redis-rabbitmq-kafka) | | Job lifecycle | [/docs/03-job-lifecycle-and-schema](/docs/03-job-lifecycle-and-schema) | | Outbox / inbox | [/docs/04-outbox-inbox-idempotency](/docs/04-outbox-inbox-idempotency) | | Celery example | [/docs/05-celery-rabbitmq-example](/docs/05-celery-rabbitmq-example) | | Taskiq example | [/docs/06-taskiq-rabbitmq-example](/docs/06-taskiq-rabbitmq-example) | | Dramatiq example | [/docs/07-dramatiq-rabbitmq-example](/docs/07-dramatiq-rabbitmq-example) | | FastStream / Kafka | [/docs/08-faststream-kafka-example](/docs/08-faststream-kafka-example) | | Errors / DLQ | [/docs/09-error-taxonomy-retries-dlq](/docs/09-error-taxonomy-retries-dlq) | | Testing | [/docs/10-testing-workers](/docs/10-testing-workers) | | Event-driven design | [/docs/11-event-driven-system-design](/docs/11-event-driven-system-design) | | DDD with FastAPI | [/docs/12-ddd-with-fastapi](/docs/12-ddd-with-fastapi) | | Laravel queues → FastAPI | [/docs/13-laravel-queues-to-fastapi](/docs/13-laravel-queues-to-fastapi) | | Celery Beat scheduling | [/docs/14-celery-beat-scheduling](/docs/14-celery-beat-scheduling) | | Observability & Flower | [/docs/15-observability-metrics-flower](/docs/15-observability-metrics-flower) | | Locks & rate limits | [/docs/16-locks-and-rate-limits](/docs/16-locks-and-rate-limits) | | Celery RL + lock pipeline | [/docs/17-celery-rate-limit-lock-pipeline](/docs/17-celery-rate-limit-lock-pipeline) | | Monitoring Celery with Flower | [/docs/18-monitoring-celery-flower](/docs/18-monitoring-celery-flower) | | Flower + Prometheus | [/docs/19-flower-prometheus](/docs/19-flower-prometheus) | ## Creator [Amin Sharifi](http://moaminsharifi.com/) ## Implementation recipes ### Celery pipeline (API RL → enqueue → RabbitMQ → job RL + lock → side effect) | Step | Implementation | |------|----------------| | API rate limit | FastAPI dependency + Redis fixed window / token bucket → HTTP 429 | | Enqueue | Insert `jobs` row → `task.apply_async(..., queue="jobs.io")` (Celery → RabbitMQ) | | Job rate limit | Redis token bucket at start of Celery task; `self.retry(countdown=…)` if empty | | Entity lock | Redis `SET key token NX EX ttl`; work; Lua release if token matches | | Side effect | Vendor/DB call inside lock; mark job succeeded/failed in DB | | Product status | `GET /jobs/{id}` from DB — not Flower | **Full walkthrough with code:** [Celery RL + lock pipeline](/docs/17-celery-rate-limit-lock-pipeline) ```text API rate limit → enqueue → RabbitMQ ↓ job rate limit + entity lock → side effect ``` **Concepts:** [Locks and rate limits](/docs/16-locks-and-rate-limits) **Celery wiring:** [Celery + RabbitMQ example](/docs/05-celery-rabbitmq-example) **Retries / DLQ:** [Error taxonomy](/docs/09-error-taxonomy-retries-dlq) ### Official Celery references - [Celery tasks](https://docs.celeryq.dev/en/stable/userguide/tasks.html) - [Retrying](https://docs.celeryq.dev/en/stable/userguide/tasks.html#retrying) - [Task.rate_limit](https://docs.celeryq.dev/en/stable/userguide/tasks.html#Task.rate_limit) (per-worker; prefer Redis for multi-worker fairness) - [Routing](https://docs.celeryq.dev/en/stable/userguide/routing.html) - [Monitoring](https://docs.celeryq.dev/en/stable/userguide/monitoring.html) - [Flower](https://flower.readthedocs.io/) ## Flower and Celery monitoring | Resource | Use | |----------|-----| | [Monitoring Celery with Flower](/docs/18-monitoring-celery-flower) | Run Flower, events (`-E`), security, what to watch | | [Observability & metrics](/docs/15-observability-metrics-flower) | Health, JSON logs, SLOs, alert severities | | [Flower docs](https://flower.readthedocs.io/) | Upstream project | | [Celery monitoring](https://docs.celeryq.dev/en/stable/userguide/monitoring.html) | Events, inspect, dumps | **Reminder:** Flower = ops UI only. User-facing job status = application `jobs` table. ### Prometheus - [Prometheus configuration](https://prometheus.io/docs/prometheus/latest/configuration/configuration/) - [Prometheus basic auth scrape](https://prometheus.io/docs/prometheus/latest/configuration/configuration/#scrape_config) - [RabbitMQ Prometheus](https://www.rabbitmq.com/docs/prometheus) - In-app: [Flower + Prometheus integration](/docs/19-flower-prometheus)