--- title: "Async Jobs" description: "Async job processing patterns for background tasks, Celery workflows, task scheduling, retry strategies, and distributed task execution. Use when implementing background job processing, task queues, or scheduled task systems." canonical: "https://orchestkit.yonyon.ai/docs/reference/skills/async-jobs" --- # Async Jobs Async job processing patterns for background tasks, Celery workflows, task scheduling, retry strategies, and distributed task execution. Use when implementing background job processing, task queues, or scheduled task systems. Reference medium > **Not directly invocable** — no slash command and no model auto-selection. An agent loads it explicitly via `Read()`. > **Async Jobs** Async job processing patterns for background tasks, Celery workflows, task scheduling, retry strategies, and distributed task execution. Use when implementing background job processing, task queues, or scheduled task systems. # Async Jobs Background task processing with Celery, ARQ, Redis and Temporal. This skill is a wrapper, not a manual: Celery and ARQ document their own product well, so what lives here is our delta, the thresholds, working config, ordering constraints and tool-choice rules we picked. Product mechanics are linked, not restated. Start with `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/references/ork-delta.md")`. ## Quick Reference | Topic | Where our part lives | |-------|----------------------| | [Configuration](#configuration) | `references/celery-config.md`, `rules/jobs-task-queue.md` | | [Task Routing](#task-routing) | `references/ork-delta.md` (queue taxonomy, prefetch tiers, Redis priority) | | [Canvas Workflows](#canvas-workflows) | `rules/celery-canvas.md` | | [Retry Strategies](#retry-strategies) | `references/ork-delta.md` (backoff cap, idempotency layers, lock TTLs) | | [Scheduling](#scheduling) | `rules/jobs-scheduling.md`, `references/ork-delta.md` (beat process model) | | [Monitoring](#monitoring) | `references/ork-delta.md` (alert thresholds, histogram buckets) | | [Result Backends](#result-backends) | `rules/jobs-monitoring.md`, `references/ork-delta.md` (return contract) | | [ARQ Patterns](#arq-patterns) | `rules/jobs-task-queue.md`, `references/ork-delta.md` (budgets, pool ownership) | | [Temporal Workflows](#temporal-workflows) | `rules/temporal-workflows.md` | | [Temporal Activities](#temporal-activities) | `rules/temporal-activities.md` | 10 topic areas, 6 rule files in `rules/`, house delta in `references/ork-delta.md`. ## Quick Start ```python @app.task(bind=True, max_retries=3, default_retry_delay=60) def process_payment(self, order_id: str): try: return gateway.charge(order_id) except TransientError as exc: raise self.retry(exc=exc, countdown=2 ** self.request.retries * 60) ``` Load more examples: `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/references/quick-start-examples.md")` for Celery retry task and ARQ/FastAPI integration patterns. ## Upstream coverage (do not restate) Fetch these when you need product mechanics. The right-hand column is the part we keep, because it is a house threshold, a working config or an ordering constraint that upstream cannot know. | Topic | First-party source | House subset stays in | |-------|--------------------|-----------------------| | Celery settings, serializers, time limits, worker flags | https://docs.celeryq.dev/en/stable/userguide/configuration.html and .../optimizing.html | `references/celery-config.md`, `rules/jobs-task-queue.md` | | Queue declarations, router classes, Redis priority mechanics | https://docs.celeryq.dev/en/stable/userguide/routing.html | `references/ork-delta.md` | | chain / group / chord / signature semantics | https://docs.celeryq.dev/en/stable/userguide/canvas.html | `rules/celery-canvas.md` keeps the house canvas subset. Its `si()`-in-chords guidance is UNVERIFIED and contested: confirm the argument-passing behaviour against the upstream canvas page before relying on it | | `autoretry_for`, `retry_backoff`, `Reject`, task base classes | https://docs.celeryq.dev/en/stable/userguide/tasks.html | `references/ork-delta.md`, `rules/jobs-task-queue.md` | | Beat schedules, crontab syntax, DatabaseScheduler | https://docs.celeryq.dev/en/stable/userguide/periodic-tasks.html and https://django-celery-beat.readthedocs.io/en/latest/ | `rules/jobs-scheduling.md` keeps our `beat_schedule` shapes; `references/ork-delta.md` keeps the process model | | Flower flags, `inspect`, signal names | https://docs.celeryq.dev/en/stable/userguide/monitoring.html, https://docs.celeryq.dev/en/stable/userguide/signals.html, https://flower.readthedocs.io/en/latest/config.html | `references/ork-delta.md` | | Result backend, `AsyncResult`, custom states | https://docs.celeryq.dev/en/stable/userguide/configuration.html | `rules/jobs-monitoring.md` keeps our status endpoints and `update_state()` usage | | Per-task `rate_limit`, `control.rate_limit`, Redis Lua | https://docs.celeryq.dev/en/stable/userguide/workers.html, https://redis.io/docs/latest/develop/programmability/eval-intro/ | `references/ork-delta.md` | | ARQ `WorkerSettings`, `enqueue_job`, `_defer_by` / `_defer_until`, `Job` status | https://arq-docs.helpmanual.io/ | `rules/jobs-task-queue.md` keeps the worker skeleton; `references/ork-delta.md` keeps the budgets | | FastAPI lifespan and dependency wiring | https://fastapi.tiangolo.com/advanced/events/ | `references/ork-delta.md` | | Distributed locks with `SET NX EX` | https://redis.io/docs/latest/commands/set/ | `references/ork-delta.md` | ## Configuration Load: `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/references/celery-config.md")`. | Decision | Recommendation | |----------|----------------| | Serializer | JSON (never pickle) | | Ack mode | Late ack (`task_acks_late=True`) | | Prefetch | 1 for fair, 4-8 for throughput | | Time limit | soft < hard (540 / 600) | | Timezone | UTC always | ## Task Routing | Decision | Recommendation | |----------|----------------| | Queue count | 5: critical / high / default / low / bulk | | Priority levels | 0-9, with all four Redis priority switches set together | | Worker assignment | Dedicated worker per queue | | Prefetch | 1 critical, 2 high, 4 default, 8 low/bulk | | Routing | Router class once past 5 routing rules | ## Canvas Workflows Load: `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/rules/celery-canvas.md")`. | Decision | Recommendation | |----------|----------------| | Sequential | Chain with `s()` | | Parallel | Group for independent tasks | | Fan-in | Chord (all header tasks must succeed for the body to run) | | Ignore input | Use `si()` immutable signature | | Error in chain | `Reject` stops the chain, `retry` continues it | | Partial failures | Return an error dict from chord header tasks | ## Retry Strategies | Decision | Recommendation | |----------|----------------| | Retry delay | Exponential backoff, jitter on, capped at 600s | | Max retries | 3-5 for transient, 0 for permanent | | Idempotency | Redis marker (86400s TTL) plus the vendor idempotency key | | Failed tasks | DLQ for manual review | | Singleton | Redis lock with a TTL longer than the hard time limit | ## Scheduling Load: `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/rules/jobs-scheduling.md")`. | Decision | Recommendation | |----------|----------------| | Schedule type | Crontab for time-based, float interval for frequency | | Dynamic | DatabaseScheduler (`django-celery-beat`) | | Overlap | Redis lock, 3600s default and 7200s for long jobs | | Beat process | Separate process; embedded `--beat` is development only | | Timezone | UTC always | ## Monitoring | Decision | Recommendation | |----------|----------------| | Dashboard | Flower with persistent storage | | Metrics | Prometheus wired to `task_prerun` / `task_postrun` / `task_failure` | | Health | Broker reachable, at least one worker, queue depths | | Alerting | critical > 100, default > 5000, workers < 1 | | Autoscale | Queue depth > 500 | ## Result Backends Load: `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/rules/jobs-monitoring.md")`. | Decision | Recommendation | |----------|----------------| | Status storage | Redis result backend, status and small JSON only | | Large results | S3 or database, task returns a reference dict | | Progress | Custom states with `update_state()` | | Result query | `AsyncResult` with state checks | ## ARQ Patterns Load: `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/rules/jobs-task-queue.md")`. | Decision | Recommendation | |----------|----------------| | Simple async | ARQ (native async), `max_jobs=10`, `job_timeout=300` | | Pool ownership | FastAPI lifespan, never a per-request `create_pool` | | Complex workflows | Celery (chains, chords, DLQ, per-task rate limits) | | In-process quick | FastAPI BackgroundTasks, under 30s, non-critical only | | LLM workflows | LangGraph, not Celery | ## Tool Selection Load: `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/references/quick-start-examples.md")` for the full tool comparison table (ARQ, Celery, RQ, Dramatiq, FastAPI BackgroundTasks). ## Anti-Patterns (FORBIDDEN) Load details: `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/references/anti-patterns.md")` for the full list. Key rules: never run long tasks in request handlers, never block on results inside tasks, never store large results in Redis, always use idempotency for retried tasks. ## Temporal Workflows Load: `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/rules/temporal-workflows.md")`. | Decision | Recommendation | |----------|----------------| | Workflow ID | Business-meaningful, idempotent | | Determinism | Use `workflow.random()`, `workflow.now()` | | I/O | Always via activities, never directly | ## Temporal Activities Load: `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/rules/temporal-activities.md")`. | Decision | Recommendation | |----------|----------------| | Activity timeout | `start_to_close` for most cases | | Error handling | Non-retryable for business errors | | Testing | `WorkflowEnvironment.start_local()` for integration tests | ## Related Skills - `ork:python-backend` - FastAPI, asyncio, SQLAlchemy patterns - `ork:langgraph` - LangGraph workflow patterns (use for LLM workflows, not Celery) - `ork:distributed-systems` - Resilience patterns, circuit breakers - `ork:monitoring-observability` - Metrics and alerting ## Capability Details Load details: `Read("$\{CLAUDE_PLUGIN_ROOT\}/skills/async-jobs/references/capability-details.md")` for the keyword index and problem-to-capability mapping. --- ## Rules (6) ### Compose multi-step task workflows using Celery canvas primitives and chains — HIGH # Canvas Workflows ## Chains (Sequential) ```python from celery import chain workflow = chain( extract_data.s(source_id), # Returns raw_data transform_data.s(), # Receives raw_data load_data.s(destination_id), # Receives clean_data ) result = workflow.apply_async() ``` ## Groups (Parallel) ```python from celery import group parallel = group(process_chunk.s(chunk) for chunk in chunks) group_result = parallel.apply_async() results = group_result.get() # List of results ``` ## Chords (Parallel + Callback) ```python from celery import chord workflow = chord( [process_chunk.s(chunk) for chunk in chunks], aggregate_results.s() # Receives list of all results ) result = workflow.apply_async() ``` ## Signatures ```python # Reusable task signature sig = signature("tasks.process_order", args=[order_id], kwargs={"priority": "high"}) # Immutable signature (won't receive results from previous task) sig = process_order.si(order_id) # Partial signature (curry arguments) partial_sig = send_email.s(subject="Order Update") ``` ## Map and Starmap ```python workflow = process_item.map([item1, item2, item3]) workflow = send_email.starmap([("user1@ex.com", "S1"), ("user2@ex.com", "S2")]) workflow = process_item.chunks(items, batch_size=100) ``` **Incorrect — Mutable signature in chord:** ```python # Body receives polluted args from each parallel task chord( [process_chunk.s(chunk) for chunk in chunks], aggregate_results.s() # Receives all chunk results concatenated! ) ``` **Correct — Immutable signature in chord:** ```python # Body receives clean list of results chord( [process_chunk.si(chunk) for chunk in chunks], # .si() = immutable aggregate_results.s() # Receives [result1, result2, result3] ) ``` ## Error Handling - Chain stops when any task fails; subsequent tasks don't run - If any chord header task fails, the body won't execute - Always use `si()` (immutable signatures) in chords to prevent arg pollution ### Track background job status and execution metrics for operational visibility — HIGH # Job Status Tracking ## Job Status Enum ```python from enum import Enum class JobStatus(Enum): PENDING = "pending" STARTED = "started" PROGRESS = "progress" SUCCESS = "success" FAILURE = "failure" REVOKED = "revoked" ``` ## ARQ Status Endpoint ```python @router.get("/api/v1/jobs/{job_id}") async def get_job_status(job_id: str, arq: ArqRedis = Depends(get_arq_pool)): job = Job(job_id, arq) status = await job.status() result = await job.result() if status == JobStatus.complete else None return {"job_id": job_id, "status": status, "result": result} ``` ## Celery Progress Updates ```python @shared_task(bind=True) def generate_report(self, report_id: str) -> dict: self.update_state(state="PROGRESS", meta={"step": "fetching"}) data = fetch_report_data(report_id) self.update_state(state="PROGRESS", meta={"step": "rendering"}) pdf = render_pdf(data) return {"report_id": report_id, "size": len(pdf)} ``` **Incorrect — No progress updates:** ```python @shared_task def generate_report(report_id: str): # Long-running task with no feedback data = fetch_report_data(report_id) pdf = render_pdf(data) return {"report_id": report_id} ``` **Correct — Progress updates:** ```python @shared_task(bind=True) def generate_report(self, report_id: str): self.update_state(state="PROGRESS", meta={"step": "fetching", "percent": 25}) data = fetch_report_data(report_id) self.update_state(state="PROGRESS", meta={"step": "rendering", "percent": 75}) pdf = render_pdf(data) return {"report_id": report_id} ``` ## Celery Status Endpoint ```python @router.get("/api/v1/jobs/{job_id}") async def get_job(job_id: str): result = AsyncResult(job_id, app=celery_app) return { "job_id": job_id, "status": result.status, "result": result.result if result.ready() else None, "progress": result.info if result.status == "PROGRESS" else None, } ``` ### Schedule reliable periodic background tasks without overlap or timing drift — HIGH # Scheduling & Background Tasks ## Celery Beat (Periodic Tasks) ```python from celery.schedules import crontab celery_app.conf.beat_schedule = { "cleanup-expired-sessions": { "task": "app.workers.tasks.cleanup_sessions", "schedule": crontab(minute=0, hour="*/6"), # Every 6 hours }, "generate-daily-report": { "task": "app.workers.tasks.daily_report", "schedule": crontab(minute=0, hour=2), # 2 AM daily }, "sync-external-data": { "task": "app.workers.tasks.sync_data", "schedule": 300.0, # Every 5 minutes }, } ``` ## FastAPI BackgroundTasks (In-Process) ```python from fastapi import BackgroundTasks @router.post("/api/v1/users") async def create_user(data: UserCreate, background_tasks: BackgroundTasks): user = await service.create_user(data) background_tasks.add_task(send_welcome_email, user.email) return user ``` ## FastAPI + Distributed Queue ```python @router.post("/api/v1/exports") async def create_export(data: ExportRequest, arq: ArqRedis = Depends(get_arq_pool)): job = await arq.enqueue_job("export_data", data.dict()) return {"job_id": job.job_id} ``` **Incorrect — Using BackgroundTasks for long jobs:** ```python # In-process task blocks other requests @router.post("/export") async def create_export(background_tasks: BackgroundTasks): background_tasks.add_task(generate_large_export) # 5+ minutes! return {"status": "started"} ``` **Correct — Use distributed queue:** ```python # Offload to worker, instant response @router.post("/export") async def create_export(arq: ArqRedis = Depends(get_arq_pool)): job = await arq.enqueue_job("generate_large_export") return {"job_id": job.job_id} ``` ## Key Decisions | Scenario | Use | |----------|-----| | Quick, non-critical | FastAPI BackgroundTasks | | Periodic/scheduled | Celery Beat | | Distributed, durable | ARQ or Celery | | LLM workflows | LangGraph (not Celery) | ### Set up task queues as the foundation for reliable background job processing — HIGH # Task Queue Setup ## ARQ (Async Redis Queue) ```python from arq import create_pool from arq.connections import RedisSettings async def startup(ctx: dict): ctx["db"] = await create_db_pool() ctx["http"] = httpx.AsyncClient() async def shutdown(ctx: dict): await ctx["db"].close() await ctx["http"].aclose() class WorkerSettings: redis_settings = RedisSettings(host="redis", port=6379) functions = [send_email, generate_report, process_webhook] on_startup = startup on_shutdown = shutdown max_jobs = 10 job_timeout = 300 ``` ## ARQ Task Definition ```python async def send_email(ctx: dict, to: str, subject: str, body: str) -> dict: http = ctx["http"] response = await http.post("https://api.sendgrid.com/v3/mail/send", json={"to": to, "subject": subject, "html": body}, headers={"Authorization": f"Bearer {SENDGRID_KEY}"}) return {"status": response.status_code, "to": to} ``` ## Celery Setup ```python from celery import Celery celery_app = Celery("orchestkit", broker="redis://redis:6379/0", backend="redis://redis:6379/1") celery_app.conf.update( task_serializer="json", task_track_started=True, task_time_limit=600, task_soft_time_limit=540, worker_prefetch_multiplier=1, task_acks_late=True, task_reject_on_worker_lost=True, ) ``` ## Celery Task with Retry ```python @shared_task(bind=True, max_retries=3, default_retry_delay=60, autoretry_for=(ConnectionError, TimeoutError)) def send_email(self, to: str, subject: str, body: str) -> dict: try: response = requests.post(url, json=data, timeout=30) response.raise_for_status() return {"status": "sent", "to": to} except Exception as exc: raise self.retry(exc=exc) ``` **Incorrect — No retry strategy:** ```python # Fails permanently on first error @shared_task def send_email(to: str, subject: str): response = requests.post(url, json=data) # Network error = lost job return response.json() ``` **Correct — Retry with exponential backoff:** ```python # Retries with backoff @shared_task(bind=True, max_retries=3, default_retry_delay=60, autoretry_for=(ConnectionError, TimeoutError)) def send_email(self, to: str, subject: str): try: response = requests.post(url, json=data, timeout=30) response.raise_for_status() return response.json() except Exception as exc: raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries)) ``` ## Tool Selection | Tool | Best For | Complexity | |------|----------|------------| | ARQ | FastAPI, async jobs | Low | | Celery | Complex workflows | High | | RQ | Simple Redis queues | Low | | FastAPI BackgroundTasks | Quick in-process | None | ### Configure Temporal activity timeouts, heartbeats, and retry policies to prevent data loss — HIGH ## Temporal Activity and Worker Patterns **Incorrect — missing heartbeat and error classification:** ```python @activity.defn async def process_payment(input: PaymentInput) -> PaymentResult: # WRONG: No heartbeat for long operation # WRONG: No error classification (all errors retry) response = await httpx.post("https://payments.example.com/charge", json={"order_id": input.order_id, "amount": input.amount}) return PaymentResult(**response.json()) ``` **Correct — heartbeat, error classification, and proper worker setup:** ```python from temporalio import activity from temporalio.exceptions import ApplicationError @activity.defn async def process_payment(input: PaymentInput) -> PaymentResult: activity.logger.info(f"Processing payment for order {input.order_id}") try: async with httpx.AsyncClient() as client: response = await client.post( "https://payments.example.com/charge", json={"order_id": input.order_id, "amount": input.amount}) response.raise_for_status() return PaymentResult(**response.json()) except httpx.HTTPStatusError as e: if e.response.status_code == 402: # Non-retryable: payment declined is a business error raise ApplicationError("Payment declined", non_retryable=True, type="PaymentDeclined") raise # Retryable: transient HTTP errors @activity.defn async def send_notification(input: NotificationInput) -> None: for i, recipient in enumerate(input.recipients): # Heartbeat for long operations (required for activities > 60s) activity.heartbeat(f"Sending {i+1}/{len(input.recipients)}") await send_email(recipient, input.subject, input.body) ``` ### Worker Configuration ```python from temporalio.client import Client from temporalio.worker import Worker async def main(): client = await Client.connect("localhost:7233") worker = Worker( client, task_queue="order-processing", workflows=[OrderWorkflow], activities=[create_order, process_payment, reserve_inventory, cancel_order_activity], ) await worker.run() async def start_order_workflow(order_data: OrderInput) -> str: client = await Client.connect("localhost:7233") handle = await client.start_workflow( OrderWorkflow.run, order_data, id=f"order-{order_data.order_id}", task_queue="order-processing", ) return handle.id ``` ### Testing with WorkflowEnvironment ```python import pytest from temporalio.testing import WorkflowEnvironment @pytest.fixture async def workflow_env(): async with await WorkflowEnvironment.start_local() as env: yield env @pytest.mark.asyncio async def test_order_workflow(workflow_env): async with Worker(workflow_env.client, task_queue="test", workflows=[OrderWorkflow], activities=[create_order, process_payment]): result = await workflow_env.client.execute_workflow( OrderWorkflow.run, OrderInput(id="test-1", total=100), id="test-order-1", task_queue="test", ) assert result.order_id == "test-1" ``` ### Key Decisions | Decision | Recommendation | |----------|----------------| | Activity timeout | `start_to_close` for most cases | | Retry policy | 3 attempts default, exponential backoff | | Heartbeating | Required for activities > 60s | | Error handling | `ApplicationError(non_retryable=True)` for business errors | | Worker deployment | Separate workers per task queue in production | | Testing | `WorkflowEnvironment.start_local()` for integration tests | ### Define deterministic Temporal workflows with correct signal and query patterns — HIGH ## Temporal Workflow Definitions **Incorrect — non-deterministic workflow operations:** ```python @workflow.defn class OrderWorkflow: @workflow.run async def run(self, order_data: OrderInput) -> OrderResult: # WRONG: Non-deterministic in workflow code if random.random() > 0.5: await do_something() if datetime.now() > deadline: await cancel() # WRONG: Direct I/O in workflow response = await httpx.get("https://api.example.com") ``` **Correct — deterministic workflow with proper APIs:** ```python from temporalio import workflow from temporalio.common import RetryPolicy from datetime import timedelta @workflow.defn class OrderWorkflow: def __init__(self): self._status = "pending" self._order_id: str | None = None @workflow.run async def run(self, order_data: OrderInput) -> OrderResult: self._order_id = await workflow.execute_activity( create_order, order_data, start_to_close_timeout=timedelta(seconds=30), retry_policy=RetryPolicy(maximum_attempts=3, initial_interval=timedelta(seconds=1)), ) self._status = "processing" # Parallel activities via asyncio.gather payment, inventory = await asyncio.gather( workflow.execute_activity(process_payment, PaymentInput(order_id=self._order_id), start_to_close_timeout=timedelta(minutes=5)), workflow.execute_activity(reserve_inventory, InventoryInput(order_id=self._order_id), start_to_close_timeout=timedelta(minutes=2)), ) self._status = "completed" return OrderResult(order_id=self._order_id, payment_id=payment.id) @workflow.query def get_status(self) -> str: return self._status @workflow.signal async def cancel_order(self, reason: str): self._status = "cancelling" await workflow.execute_activity(cancel_order_activity, CancelInput(order_id=self._order_id), start_to_close_timeout=timedelta(seconds=30)) self._status = "cancelled" ``` ### Saga Pattern with Compensation ```python @workflow.defn class OrderSagaWorkflow: @workflow.run async def run(self, order: OrderInput) -> OrderResult: compensations: list[tuple[Callable, Any]] = [] try: reservation = await workflow.execute_activity( reserve_inventory, order.items, start_to_close_timeout=timedelta(minutes=2)) compensations.append((release_inventory, reservation.id)) payment = await workflow.execute_activity( charge_payment, PaymentInput(order_id=order.id), start_to_close_timeout=timedelta(minutes=5)) compensations.append((refund_payment, payment.id)) shipment = await workflow.execute_activity( create_shipment, ShipmentInput(order_id=order.id), start_to_close_timeout=timedelta(minutes=3)) return OrderResult(order_id=order.id, payment_id=payment.id, shipment_id=shipment.id) except Exception: workflow.logger.warning(f"Saga failed, running {len(compensations)} compensations") for compensate_fn, compensate_arg in reversed(compensations): try: await workflow.execute_activity(compensate_fn, compensate_arg, start_to_close_timeout=timedelta(minutes=2)) except Exception as e: workflow.logger.error(f"Compensation failed: {e}") raise ``` ### Key Decisions | Decision | Recommendation | |----------|----------------| | Workflow ID | Business-meaningful, idempotent (e.g., `order-\{order_id\}`) | | Task queue | Per-service or per-workflow-type | | Determinism | Use `workflow.random()`, `workflow.now()` — never stdlib | | I/O | Always via activities, never directly in workflows | | Timers | `workflow.wait_condition()` with timeout for human-in-the-loop | --- ## References (5) ### Anti Patterns # Anti-Patterns (FORBIDDEN) ```python # NEVER run long tasks synchronously in request handlers @router.post("/api/v1/reports") async def create_report(data: ReportRequest): pdf = await generate_pdf(data) # Blocks for minutes! # NEVER block on results inside tasks (causes deadlock) @celery_app.task def bad_task(): result = other_task.delay() return result.get() # Blocks worker! # NEVER store large results in Redis @shared_task def process_file(file_id: str) -> bytes: return large_file_bytes # Store in S3/DB instead! # NEVER skip idempotency for retried tasks @celery_app.task(max_retries=3) def create_order(order): Order.create(order) # Creates duplicates on retry! # NEVER use BackgroundTasks for distributed work background_tasks.add_task(long_running_job) # Lost if server restarts # NEVER ignore task acknowledgment settings celery_app.conf.task_acks_late = False # Default loses tasks on crash # ALWAYS use immutable signatures in chords chord([task.s(x) for x in items], callback.si()) # si() prevents arg pollution ``` ### Capability Details # Capability Details ### celery-config **Keywords:** celery, configuration, broker, worker, setup **Solves:** - Production Celery app configuration - Broker and backend setup - Worker tuning and time limits ### task-routing **Keywords:** priority, queue, routing, high priority, worker **Solves:** - Premium user task prioritization - Multi-queue worker deployment - Dynamic task routing ### canvas-workflows **Keywords:** chain, group, chord, signature, canvas, workflow, pipeline **Solves:** - Complex multi-step task pipelines - Parallel task execution with aggregation - Sequential task dependencies ### retry-strategies **Keywords:** retry, backoff, idempotency, dead letter, resilience **Solves:** - Exponential backoff with jitter - Duplicate prevention for retried tasks - Failed task handling with DLQ ### scheduled-tasks **Keywords:** periodic, scheduled, cron, celery beat, interval **Solves:** - Run tasks on schedule (crontab) - Dynamic schedule management - Overlap prevention for long tasks ### monitoring-health **Keywords:** flower, monitoring, health check, metrics, alerting **Solves:** - Production task monitoring dashboard - Worker health checks - Queue depth autoscaling ### result-backends **Keywords:** result, state, progress, AsyncResult, status **Solves:** - Task progress tracking with custom states - Result storage strategies - Job status API endpoints ### arq-patterns **Keywords:** arq, async queue, redis queue, fastapi background **Solves:** - Lightweight async background tasks for FastAPI - Simple Redis job queue with async/await - Job status tracking ### Celery Config # Celery Configuration Production Celery app setup with secure defaults, broker configuration, and worker tuning. ## Application Setup ```python # backend/app/workers/celery_app.py from celery import Celery celery_app = Celery( "myapp", broker="redis://redis:6379/0", backend="redis://redis:6379/1", ) celery_app.conf.update( # Serialization — JSON only (never pickle) task_serializer="json", accept_content=["json"], result_serializer="json", # Timezone timezone="UTC", enable_utc=True, # Task tracking task_track_started=True, # Time limits (seconds) task_time_limit=600, # 10 min hard kill task_soft_time_limit=540, # 9 min soft limit (raises SoftTimeLimitExceeded) # Worker behavior worker_prefetch_multiplier=1, # Fair task distribution worker_max_tasks_per_child=1000, # Restart worker after N tasks (leak protection) # Reliability task_acks_late=True, # Ack after completion (not before) task_reject_on_worker_lost=True, # Re-queue if worker dies mid-task # Result backend result_expires=86400, # 24 hours result_backend_transport_options={ "global_keyprefix": "celery_result:", }, ) ``` ## Broker Configuration ```python # Redis broker options celery_app.conf.broker_transport_options = { "visibility_timeout": 43200, # 12 hours (must exceed longest task) "retry_policy": { "timeout": 5.0, }, "max_retries": 3, } # Connection pool celery_app.conf.broker_pool_limit = 10 celery_app.conf.broker_connection_retry_on_startup = True ``` ## Worker Tuning ```bash # Production worker startup celery -A app worker \ --loglevel=INFO \ --concurrency=4 \ --prefetch-multiplier=1 \ --max-tasks-per-child=1000 \ --without-heartbeat \ --without-gossip \ --without-mingle \ -Ofair ``` ### Concurrency Guidelines | Workload Type | Concurrency | Prefetch | Notes | |---------------|-------------|----------|-------| | CPU-bound | N cores | 1 | Process pool | | I/O-bound | 2-4x cores | 1-4 | Gevent/eventlet or prefork | | Mixed | N cores | 1 | Process pool, fair scheduling | | Bulk/batch | N cores | 4-8 | Higher prefetch for throughput | ## Docker Compose ```yaml # docker-compose.yml services: redis: image: redis:7-alpine ports: - "6379:6379" volumes: - redis-data:/data celery-worker: build: . command: celery -A app worker --loglevel=INFO --concurrency=4 depends_on: - redis environment: - CELERY_BROKER_URL=redis://redis:6379/0 - CELERY_RESULT_BACKEND=redis://redis:6379/1 celery-beat: build: . command: celery -A app beat --loglevel=INFO depends_on: - redis environment: - CELERY_BROKER_URL=redis://redis:6379/0 volumes: redis-data: ``` ## Configuration Anti-Patterns ```python # NEVER use pickle serializer (security risk) celery_app.conf.task_serializer = "pickle" # FORBIDDEN # NEVER disable late ack in production celery_app.conf.task_acks_late = False # Tasks lost on crash # NEVER set visibility_timeout shorter than longest task celery_app.conf.broker_transport_options = { "visibility_timeout": 60, # Task re-dispatched if still running! } # NEVER skip time limits # Without limits, a hung task blocks the worker slot forever ``` ### Ork Delta # Async Jobs: our delta House thresholds, working config and ordering constraints for background jobs in OrchestKit projects. Everything here is the part vendor docs cannot supply: the numbers we picked and the order we do things in. Product mechanics live upstream, linked per entry. Neighbouring skills: `ork:python-backend` (FastAPI, asyncio, SQLAlchemy runtime), `ork:distributed-systems` (circuit breakers around the queue), `ork:monitoring-observability` (the alerting layer these thresholds feed), `ork:langgraph` (LLM workflows, which never belong in Celery). Provenance: the retired reference files carried no issue, PR or commit reference, so no entry below claims one. Each names the retired file it was distilled from. ## Run exactly five named queues: critical, high, default, low, bulk Why: house queue taxonomy. Three queues under-separate payment work from reporting; past five nobody can keep the worker fleet straight. `task_default_queue` stays `default` and `task_default_priority` stays 5, so an unrouted task lands mid-pack instead of starving. Distilled from the retired task-routing.md; no traced incident. Upstream: https://docs.celeryq.dev/en/stable/userguide/routing.html ## Set prefetch per queue tier, never globally Why: house numeric budget, one worker process per queue: `--prefetch-multiplier=1` on critical, 2 on high, 4 on default, 8 on low and bulk. A single global prefetch either starves the latency-sensitive queues or wastes round trips on bulk. Distilled from the retired task-routing.md; no traced incident. Upstream: https://docs.celeryq.dev/en/stable/userguide/optimizing.html ## Turn on Redis priority explicitly before using priority= Why: ordering constraint. On a Redis broker `priority=` is silently ignored unless `broker_transport_options` sets `priority_steps: list(range(10))`, `sep: ":"` and `queue_order_strategy: "priority"`, with `x-max-priority: 10` in `queue_arguments`. House default is all four together, configured before the first `apply_async`; setting one of them is the failure mode that looks like priority working. Distilled from the retired task-routing.md; no traced incident. Upstream: https://docs.celeryq.dev/en/stable/userguide/routing.html ## Alert and autoscale on queue depth with tier-specific thresholds Why: house numbers. Autoscale trigger at depth > 500, generic queue alert at depth > 1000, then tier overrides: critical queue > 100 is a critical alert because that queue should never back up, default queue > 5000 is a warning, and fewer than 1 active worker is a critical alert. Distilled from the retired task-routing.md and monitoring-health.md; no traced incident. Upstream: https://docs.celeryq.dev/en/stable/userguide/monitoring.html ## Wire metrics off task signals with our latency buckets Why: house observability budget. Metrics hang off `task_prerun`, `task_postrun`, `task_failure` and `task_retry`, with a duration histogram bucketed at `[0.1, 0.5, 1, 5, 10, 30, 60, 300, 600]` so the top bucket lines up with our 600s hard time limit, and a queue-depth gauge refreshed every 30s by a beat task. Ordering constraint: `task_prerun` stores the start time and `task_postrun` reads it, so registering one without the other yields a histogram that never observes. Distilled from the retired monitoring-health.md; no traced incident. Upstream: https://docs.celeryq.dev/en/stable/userguide/signals.html ## Cap retry backoff at 600s and always jitter Why: house retry budget. `retry_backoff=True`, `retry_backoff_max=600`, `retry_jitter=True`, with `max_retries` 3 to 5 for transient failures and 0 for anything a retry cannot fix. Uncapped exponential backoff pushes the last attempt hours out, and unjittered backoff reconverges the whole fleet on the recovering dependency. Distilled from the retired retry-strategies.md; no traced incident. Upstream: https://docs.celeryq.dev/en/stable/userguide/tasks.html ## Make a retried task idempotent at two layers Why: house rule for money-moving and record-creating tasks. Layer one is a Redis marker keyed `processed:<domain>:<id>` with an 86400s TTL, checked before the side effect; layer two is the vendor's own idempotency key on the outbound call. Layer one alone loses the race between a worker crash and the marker write, so both are required, in that order. Distilled from the retired retry-strategies.md; no traced incident. Upstream: https://docs.celeryq.dev/en/stable/userguide/tasks.html ## Give every lock a TTL longer than the task hard time limit Why: ordering constraint. Singleton task locks use `SET key 1 NX EX 300`; scheduled-task locks use 3600 by default and 7200 for known-long jobs. The TTL must exceed `task_time_limit` (600 in our house config) or the lock expires while the first run is still executing and a second run starts on top of it. Always release in a `finally`. Distilled from the retired retry-strategies.md and scheduled-tasks.md; no traced incident. Upstream: https://redis.io/docs/latest/commands/set/ ## Return a reference from a task, never a payload Why: house decision, and the reason the result backend stays small. Working `result_expires` and `global_keyprefix` values live in `references/celery-config.md` and the prohibition lives in `references/anti-patterns.md`; what this entry adds is the contract: bytes go to S3 or the database and the task returns a reference dict (key, size, generated_at), so a task result always fits a Redis value. Distilled from the retired result-backends.md; no traced incident. Upstream: https://docs.celeryq.dev/en/stable/userguide/configuration.html ## Own the ARQ pool in the FastAPI lifespan, never in a per-request dependency Why: ordering constraint. A dependency shaped `async def get_arq_pool(): return await create_pool(...)` opens a fresh Redis pool on every request and never closes it. Create the pool once in the lifespan handler, stash it on `app.state`, and have the dependency read it back. The per-request shape appears in `rules/jobs-task-queue.md` and `rules/jobs-scheduling.md` snippets for brevity; production wiring uses the lifespan. Distilled from the retired arq-patterns.md; no traced incident. Upstream: https://fastapi.tiangolo.com/advanced/events/ ## Budget ARQ workers at max_jobs 10 and job_timeout 300 Why: house numbers for the lightweight lane. ARQ is chosen for FastAPI-native async work that finishes inside five minutes; anything needing chains, chords, RabbitMQ or SQS, per-task rate limits or a dead letter queue goes to Celery, and LLM workflows go to LangGraph. Shared resources (DB pool, HTTP client) are built in `on_startup` and torn down in `on_shutdown` on the worker `ctx`, not per job, and deferral uses `_defer_by` or `_defer_until` (not `_delay`). Distilled from the retired arq-patterns.md; no traced incident. Upstream: https://arq-docs.helpmanual.io/ ## Run Celery Beat as its own process, in UTC Why: house decision. `celery -A app beat` runs standalone with a pidfile; the embedded `worker --beat` form is development only, because a worker restart silently stops the schedule. `celery_app.conf.timezone = "UTC"` everywhere so DST never shifts a schedule, and dynamic per-tenant schedules use the `django-celery-beat` DatabaseScheduler rather than hand-rolled schedule tables. Distilled from the retired scheduled-tasks.md; no traced incident. Upstream: https://django-celery-beat.readthedocs.io/en/latest/ ## Rate limit at the tier that owns the quota Why: house decision. `@task(rate_limit="100/m")` is per worker process, so it only bounds a real external quota when a single worker owns the queue; fleet-wide quotas need a Redis token bucket in a Lua script (one round trip, atomic) or `app.control.rate_limit()` for a temporary load shed. Pick the tier deliberately, because the static decorator quietly multiplies by worker count. Distilled from the retired result-backends.md; no traced incident. Upstream: https://docs.celeryq.dev/en/stable/userguide/workers.html ## Token-bucket rate limiter: bucket 100, 10 tokens per second, 25 for the Stripe lane Why: distilled from the retired `references/result-backends.md`; these are the house numbers behind the Redis Lua bucket. Defaults are `bucket_size=100` with `tokens_per_second=10.0`, and the Stripe lane runs at `tokens_per_second=25` because that provider tolerates it. The Lua script sets `EXPIRE 3600` on the bucket key so an idle tenant's key does not persist forever, and a rejected task re-queues with an exponential countdown capped by `2 ** min(retries, 6)`. Without the numbers the prose "a Redis token bucket in a Lua script" is not implementable. Upstream: https://docs.celeryq.dev/en/stable/userguide/tasks.html#Task.rate_limit ## Adaptive polling: re-poll at 30s on activity, double when idle, cap at 300s Why: distilled from the retired `references/scheduled-tasks.md`; the house shape is `self.apply_async(args=(resource_id,), countdown=30)` when the poll found activity, and `next_delay = min(current_delay * 2, 300)` when it did not. A fixed-interval poller either burns the worker pool on quiet resources or lags behind a busy one; the 30s floor and the 5 minute ceiling are the bounds we hold. Upstream: https://docs.celeryq.dev/en/stable/userguide/calling.html#eta-and-countdown ## Count beat_sent, and route alerts by task-name prefix Why: distilled from the retired `references/monitoring-health.md`. Two things no vendor doc supplies. First, a `celery_beat_tasks_sent_total` counter wired to the `beat_sent` signal is the only way to detect a schedule that silently stopped firing, because a task that is never sent produces no failure to alert on. Second, alert severity is routed by task-name prefix: `sender.name.startswith("tasks.payment")` raises a critical alert on `task_failure`, while other prefixes stay at warning. Upstream: https://docs.celeryq.dev/en/stable/userguide/signals.html ## /health/celery returns 503 when unhealthy, and probes the broker with max_retries=3 Why: distilled from the retired `references/monitoring-health.md`; the endpoint returns `200` only when healthy and `503` otherwise, so a load balancer or Kubernetes probe can act on it without parsing the body. The broker check is `conn.ensure_connection(max_retries=3)`, bounded so the health endpoint cannot itself hang on a dead broker and turn a readiness probe into a timeout. Upstream: https://docs.celeryq.dev/en/stable/userguide/monitoring.html ## Use the named house task states VALIDATING, PROCESSING, UPLOADING Why: distilled from the retired `references/result-backends.md`; the house lifecycle is a named vocabulary, not a single generic PROGRESS state, with meta shaped `\{step, total, description\}`. `rules/jobs-monitoring.md` retains only PROGRESS, so the named stages live here. A polling client that has to infer progress from a percentage cannot tell "still validating" from "stuck uploading", which is the distinction the vocabulary exists to expose. Custom states also require `result_expires=86400` to stay readable for the 24 hours the house keeps results. Upstream: https://docs.celeryq.dev/en/stable/userguide/tasks.html#custom-states ### Quick Start Examples # Quick Start Examples ## Celery Task with Retry ```python from celery import shared_task @shared_task( bind=True, max_retries=3, autoretry_for=(ConnectionError, TimeoutError), retry_backoff=True, ) def process_order(self, order_id: str) -> dict: result = do_processing(order_id) return {"order_id": order_id, "status": "completed"} ``` ## ARQ Task with FastAPI ```python from arq import create_pool from arq.connections import RedisSettings async def generate_report(ctx: dict, report_id: str) -> dict: data = await ctx["db"].fetch_report_data(report_id) pdf = await render_pdf(data) return {"report_id": report_id, "size": len(pdf)} @router.post("/api/v1/reports") async def create_report(data: ReportRequest, arq: ArqRedis = Depends(get_arq_pool)): job = await arq.enqueue_job("generate_report", data.report_id) return {"job_id": job.job_id} ``` ## Tool Selection Guide | Tool | Best For | Complexity | |------|----------|------------| | ARQ | FastAPI, simple async jobs | Low | | Celery | Complex workflows, enterprise | High | | RQ | Simple Redis queues | Low | | Dramatiq | Reliable messaging | Medium | | FastAPI BackgroundTasks | In-process quick tasks | Minimal | ### Decision Criteria - **ARQ**: Native async/await, lightweight, ideal for FastAPI apps with simple background tasks - **Celery**: Full-featured canvas workflows (chains, chords, groups), production monitoring with Flower - **RQ**: Simple Redis-based queue, minimal setup, no async support - **Dramatiq**: Reliable messaging with automatic retries, simpler than Celery - **FastAPI BackgroundTasks**: In-process only, no persistence, use for fire-and-forget tasks under 30s