back
loading skill details...
Analyze Celery task configurations for reliability, performance, and debugging. Checks retry policies, task routing, serialization, result backends, worker c...
---
name: cm-celery-task-analyzer
description: Analyze Celery task configurations for reliability, performance, and debugging. Checks retry policies, task routing, serialization, result backends, worker configuration, beat schedules, and task design patterns. Use when asked to review Celery tasks, audit task reliability, check Celery best practices, optimize worker performance, review beat schedules, or troubleshoot task failures. Triggers on "celery", "celery task", "celery worker", "celery beat", "task queue", "celery retry", "celery routing", "celery config", "celery performance", "celery debug", "celery audit".
metadata:
tags: ["celery", "python", "task-queue", "async", "distributed", "worker", "redis", "rabbitmq", "scheduling", "reliability"]
---
# Celery Task Analyzer
Analyze Celery task configurations for reliability, performance, and operational excellence. Reviews task definitions, retry policies, routing rules, serialization settings, result backend configuration, worker tuning, and beat schedules. Acts as a senior distributed systems engineer auditing your Celery deployment.
## Usage
Invoke this skill when you need to review Celery task configurations, optimize worker performance, or ensure reliability best practices.
**Basic invocation:**
> Analyze the Celery tasks in /path/to/app/tasks/
> Review this Celery configuration for production readiness
> Check task reliability across the project
**Focused analysis:**
> Audit retry policies for all Celery tasks
> Review task routing configuration
> Check beat schedule for overlapping tasks
> Analyze worker configuration for memory leaks
The agent reads Celery task files, configuration modules, and beat schedules, then produces a comprehensive quality report.
## How It Works
### Step 1: Discover and Parse Celery Components
The agent locates all Celery-related code:
```bash
# Find Celery app configuration
grep -rl "Celery(\|celery_app\|make_celery" /path/to/app/ --include="*.py"
# Find task definitions
grep -rl "@app.task\|@shared_task\|@celery.task" /path/to/app/ --include="*.py"
# Find beat schedule definitions
grep -rl "beat_schedule\|CELERYBEAT_SCHEDULE\|crontab\|solar" /path/to/app/ --include="*.py"
# Find configuration files
grep -rl "broker_url\|result_backend\|CELERY_" /path/to/app/ --include="*.py"
```
The agent parses each component to extract:
- **App configuration** (broker, backend, serialization, timezone)
- **Task definitions** (name, module, decorators, arguments)
- **Retry policies** (max_retries, retry_backoff, retry_jitter)
- **Routing rules** (queues, exchanges, routing keys)
- **Beat schedules** (periodic tasks, crontab, intervals)
- **Signal handlers** (task_prerun, task_postrun, task_failure)
- **Worker configuration** (concurrency, prefetch, pool type)
### Step 2: Audit Task Definitions
The agent checks each task's configuration:
**Task decorator analysis:**
```python
# GOOD: Well-configured task
@shared_task(
bind=True,
name="orders.process_payment",
max_retries=3,
default_retry_delay=60,
retry_backoff=True,
retry_backoff_max=600,
retry_jitter=True,
acks_late=True,
reject_on_worker_lost=True,
time_limit=300,
soft_time_limit=270,
rate_limit="100/m",
track_started=True,
ignore_result=False,
)
def process_payment(self, order_id: int) -> dict:
...
# PROBLEMS the agent detects:
```
```
FAIL: Task "send_email" — no retry configuration
@shared_task
def send_email(to, subject, body):
RISK: Transient SMTP failures cause permanent message loss
FIX: Add retry policy:
@shared_task(bind=True, max_retries=3, retry_backoff=True)
def send_email(self, to, subject, body):
try: ...
except SMTPException as exc:
raise self.retry(exc=exc)
FAIL: Task "process_payment" — no time_limit
RISK: Stuck payment processing blocks worker indefinitely
FIX: Add time_limit=300, soft_time_limit=270
The soft limit raises SoftTimeLimitExceeded so you can clean up
FAIL: Task "generate_report" — acks_late=False (default)
RISK: If worker crashes mid-task, message is already acknowledged
The report generation is lost with no retry
FIX: acks_late=True, reject_on_worker_lost=True
WARN: Task "sync_inventory" — no rate_limit
This task calls external API with rate limits
RISK: Worker bursts can trigger API throttling for all consumers
FIX: rate_limit="30/m" (match API rate limit)
WARN: Task "cleanup_temp_files" — ignore_result not set
Results are stored but never read for this fire-and-forget task
FIX: ignore_result=True — saves result backend storage and bandwidth
WARN: Task "update_search_index" — no bind=True
Cannot access self.retry() or self.request without bind=True
FIX: @shared_task(bind=True) and add self as first parameter
```
**Task naming audit:**
```
Task Naming Analysis:
FAIL: Task uses auto-generated name
"app.tasks.send_email" — derived from module path
RISK: Renaming the module or moving the file breaks all pending tasks
FIX: Set explicit name: @shared_task(name="notifications.send_email")
FAIL: Inconsistent naming convention
"process_order" (no namespace)
"payments.charge_card" (dotted namespace)
"UserSync" (PascalCase)
FIX: Use consistent dotted namespace: "domain.action_verb"
"orders.process_order", "payments.charge_card", "users.sync_user"
WARN: Task name collision risk
"process" defined in both orders/tasks.py and payments/tasks.py
Auto-generated names differ but explicit names might collide
FIX: Always use namespaced explicit names
```
### Step 3: Analyze Retry Policies
The agent evaluates retry configurations for reliability:
```
Retry Policy Analysis:
Task: "orders.process_payment"
max_retries: 3
retry_backoff: True (exponential: 1s, 2s, 4s)
retry_backoff_max: 600s
retry_jitter: True
PASS: Well-configured — exponential backoff with jitter prevents thundering herd
Task: "integrations.sync_crm"
max_retries: 10
default_retry_delay: 5
retry_backoff: False
FAIL: Fixed 5-second delay with 10 retries = only 50 seconds of retry window
For CRM API issues that may last minutes, this is insufficient
FIX: retry_backoff=True, retry_backoff_max=1800
This gives: 5s, 10s, 20s, 40s, ... up to 30 min — covers longer outages
Task: "reports.generate_pdf"
max_retries: None (infinite retries!)
FAIL: Infinite retries — a poison message retries forever
RISK: Consumes worker slots indefinitely, fills dead letter queue
FIX: Set finite max_retries (3-5 for most tasks)
Task: "notifications.send_push"
Retries on all exceptions (bare except)
FAIL: Retries on ValueError, TypeError — bugs never surface
FIX: Only retry on transient errors:
except (ConnectionError, TimeoutError) as exc:
raise self.retry(exc=exc)
# Let ValueError, TypeError propagate as failures
Task: "etl.import_csv"
autoretry_for=(Exception,)
FAIL: autoretry_for too broad — retries on programming errors
FIX: autoretry_for=(IOError, ConnectionError, OperationalError)
Retry Summary:
Tasks with retry: 8/15
Tasks without retry: 5/15 (REVIEW NEEDED)
Tasks with bad retry: 2/15 (NEEDS FIX)
Infinite retry tasks: 1/15 (CRITICAL)
```
### Step 4: Review Task Routing and Queues
The agent analyzes routing configuration:
```
Routing Analysis:
Queue Configuration:
Queues defined: ["default", "high_priority", "bulk", "notifications"]
FAIL: No task_routes configured
All 15 tasks go to "default" queue
RISK: Bulk import tasks starve payment processing tasks
FIX: Configure routing:
task_routes = {
"orders.process_payment": {"queue": "high_priority"},
"etl.import_*": {"queue": "bulk"},
"notifications.*": {"queue": "notifications"},
}
FAIL: No dedicated workers per queue
Single worker process consumes all queues
RISK: Cannot scale queues independently, cannot set per-queue concurrency
FIX: Run separate workers:
celery -A app worker -Q high_priority -c 4
celery -A app worker -Q bulk -c 2
celery -A app worker -Q notifications -c 8
WARN: No priority_steps configured
Default priority is FIFO only — cannot prioritize within a queue
FIX: For RabbitMQ: task_queue_max_priority=10
Then: process_payment.apply_async(priority=9)
WARN: No dead letter exchange configured
Failed messages after max_retries are silently discarded
FIX: Configure DLX for post-mortem analysis:
task_reject_on_worker_lost = True
task_acks_on_failure_or_timeout = False
```
### Step 5: Audit Serialization and Security
The agent checks serialization settings:
```
Serialization Analysis:
FAIL: task_serializer = "pickle"
RISK: Pickle deserialization allows arbitrary code execution
Any message injected into the broker can execute code on workers
FIX: task_serializer = "json"
AND: accept_content = ["json"] (reject pickle messages)
WARN: result_serializer not explicitly set
Defaults to task_serializer — if pickle, results are also pickled
FIX: result_serializer = "json"
FAIL: accept_content = ["json", "pickle"]
Still accepts pickle — attacker can send pickle-serialized message
FIX: accept_content = ["json"] only
WARN: Task "ml.train_model" passes non-JSON-serializable argument
Argument: numpy array, pandas DataFrame
RISK: Will fail with json serializer
FIX: Serialize data before sending:
- Pass file path or S3 key instead of raw data
- Use msgpack serializer for binary data
- Convert to list/dict before sending
PASS: No task passes ORM model instances as arguments
(ORM objects are not serializable and create stale data issues)
```
### Step 6: Review Result Backend Configuration
The agent evaluates result backend setup:
```
Result Backend Analysis:
Backend: Redis (redis://localhost:6379/1)
FAIL: result_expires not configured
Default: 24 hours — results accumulate in Redis
For 10,000 tasks/day = 10,000 keys consuming memory
FIX: result_expires = 3600 (1 hour, or match your read pattern)
FAIL: Same Redis instance for broker and results
redis://localhost:6379/0 (broker)
redis://localhost:6379/1 (results)
RISK: Result storage OOM can crash the broker
FIX: Use separate Redis instances for broker and results
WARN: result_extended = False (default)
Extended results include task args, worker hostname, timestamps
Valuable for debugging production issues
FIX: result_extended = True
WARN: 8 tasks have ignore_result=False but results are never read
These tasks store results that nobody fetches
FIX: Set ignore_result=True on fire-and-forget tasks:
send_email, update_cache, sync_analytics, log_event,
send_webhook, cleanup_files, update_counter, emit_metric
FAIL: No result_backend_transport_options
No connection pooling or timeout configuration
FIX: result_backend_transport_options = {
"retry_policy": {"max_retries": 3, "interval_start": 0.2},
"socket_timeout": 5,
}
```
### Step 7: Analyze Beat Schedule
The agent reviews periodic task configuration:
```
Beat Schedule Analysis:
Schedule: 12 periodic tasks configured
FAIL: Overlapping schedules detected
"sync_all_users" runs every 5 minutes (crontab(*/5))
"sync_premium_users" runs every 3 minutes (crontab(*/3))
Both query user table — concurrent runs cause lock contention
FIX: Stagger schedules or merge into single task with priority flag
FAIL: Task "generate_daily_report" — no expires setting
Schedule: crontab(hour=2, minute=0) — runs at 2 AM
If worker is down at 2 AM, task queues and runs when worker recovers
By then the report may be stale or a new execution is already scheduled
FIX: Add expires=3600 — discard if not executed within 1 hour
FAIL: No task locking — concurrent executions possible
"sync_inventory" runs every 10 minutes, average runtime 12 minutes
Overlap: new execution starts before previous completes
FIX: Use celery-once or implement distributed lock:
from celery_once import QueueOnce
@task(base=QueueOnce, once={"graceful": True})
WARN: Beat schedule uses intervals instead of crontab for daily tasks
"cleanup" — schedule=timedelta(hours=24)
RISK: Interval drifts over time (runs at 2:00, then 2:01, 2:03...)
FIX: Use crontab for fixed-time tasks:
schedule=crontab(hour=2, minute=0)
WARN: No jitter on high-frequency tasks
5 tasks run every minute — all hit broker simultaneously
FIX: Stagger start times:
crontab(minute="*/5") — :00, :05, :10...
crontab(minute="1-59/5") — :01, :06, :11... (offset by 1)
Beat Health:
Total periodic tasks: 12
Tasks with expires: 3/12 (NEEDS ATTENTION)
Tasks with lock protection: 1/12 (CRITICAL)
Overlapping schedules: 2 pairs detected
```
### Step 8: Review Worker Configuration
The agent audits worker settings:
```
Worker Configuration Analysis:
FAIL: worker_prefetch_multiplier = 4 (default)
With long-running tasks, worker grabs 4 tasks but can only process 1
Other tasks wait in prefetch buffer — poor load distribution
FIX: worker_prefetch_multiplier = 1 for long tasks
OR: worker_prefetch_multiplier = 0 (disable) for strict fair scheduling
FAIL: No worker_max_tasks_per_child configured
RISK: Memory leaks in tasks accumulate until worker OOM
FIX: worker_max_tasks_per_child = 1000
Worker restarts after 1000 tasks — recovers leaked memory
FAIL: No worker_max_memory_per_child configured
RISK: Single memory-intensive task can crash the worker
FIX: worker_max_memory_per_child = 400000 (400 MB in KB)
Worker restarts if memory exceeds limit
WARN: worker_pool = "prefork" with I/O-bound tasks
Most tasks are API calls and database queries (I/O bound)
Prefork creates heavy processes for each worker
CONSIDER: worker_pool = "gevent" or "eventlet" for I/O-bound workloads
CAUTION: Not safe if tasks use non-greenlet-safe libraries
WARN: worker_concurrency not explicitly set
Defaults to CPU count — may be too high for memory-heavy tasks
FIX: Set based on workload:
CPU-bound: worker_concurrency = CPU_COUNT
I/O-bound (prefork): worker_concurrency = CPU_COUNT * 2
I/O-bound (gevent): worker_concurrency = 100-500
FAIL: No task_soft_time_limit default
Individual tasks can run forever if no per-task limit set
FIX: Set global defaults:
task_soft_time_limit = 300 (5 minutes)
task_time_limit = 330 (hard kill 30s after soft limit)
```
### Step 9: Check Task Design Patterns
The agent evaluates task design for correctness:
```
Task Design Analysis:
FAIL: Task "process_batch" accepts queryset as argument
def process_batch(self, queryset):
RISK: Django querysets are not serializable
RISK: Even if pickled, data is stale by the time task runs
FIX: Pass IDs, query in the task:
def process_batch(self, item_ids: list[int]):
items = Item.objects.filter(id__in=item_ids)
FAIL: Task "update_dashboard" has database transaction spanning task
with transaction.atomic():
... (entire task body)
RISK: Long-running transaction holds locks, blocks other queries
FIX: Use smaller transactions within the task, or use bulk operations
FAIL: Task "send_notifications" iterates and sends one by one
for user in users:
send_push(user.id) # Synchronous call inside task
RISK: If task fails at user #500 of 1000, all 500 are lost
FIX: Fan out to individual tasks:
group(send_push.s(user_id) for user_id in user_ids).apply_async()
WARN: Task "import_data" has no idempotency check
Re-running creates duplicate records
FIX: Use unique constraints and upsert logic, or check task ID:
if cache.get(f"task_done:{self.request.id}"):
return # Already processed
WARN: Task "process_order" calls another task synchronously
result = charge_card.delay(order_id).get(timeout=30)
FAIL: .get() blocks the worker — deadlock risk if workers exhausted
FIX: Use chain or callback:
chain(charge_card.s(order_id), fulfill_order.s()).apply_async()
PASS: Task arguments are JSON-serializable primitives
PASS: No global mutable state accessed between tasks
```
### Step 10: Produce the Analysis Report
The agent generates a comprehensive report:
```
# Celery Task Analysis Report
# Project: /path/to/app/ | Date: April 30, 2026
## Overview
Tasks: 15
Periodic tasks: 12
Queues: 4 (default, high_priority, bulk, notifications)
Broker: RabbitMQ
Result backend: Redis
Worker pool: prefork
## Overall Health Score: 51/100
## Category Scores
Task Configuration: 5/10 (missing retries, no time limits)
Retry Policies: 4/10 (infinite retries, broad exception catch)
Routing & Queues: 4/10 (no routing, single worker)
Serialization: 3/10 (pickle in use — security risk)
Result Backend: 5/10 (no expiry, shared instance)
Beat Schedule: 5/10 (overlaps, no expires, no locks)
Worker Config: 4/10 (no memory limits, no prefetch tuning)
Task Design: 6/10 (queryset args, sync calls)
Monitoring: 5/10 (no Flower, no metrics)
## Critical Issues
1. Pickle serializer enabled — remote code execution risk
2. Infinite retry task — poison messages never cleared
3. No task time limits — stuck tasks block workers indefinitely
4. Synchronous .get() call inside task — deadlock risk
5. No worker memory limits — OOM crashes
## Recommendations Summary
Estimated effort: 3-5 days for critical + high priority fixes
Expected improvement: 51 -> 80 health score
Risk reduction: Eliminates security vulnerability and 3 crash scenarios
```
## Output
The agent produces:
- **Health score**: 0-100 overall Celery configuration quality rating
- **Category scores**: granular ratings for each quality dimension
- **Critical issues**: problems that pose reliability or security risk
- **Per-task analysis**: configuration audit for each task definition
- **Retry policy review**: evaluation of each task's failure handling
- **Routing map**: visual representation of task-to-queue assignments
- **Beat schedule timeline**: periodic task execution timeline with overlap detection
- **Worker tuning recommendations**: pool, concurrency, and memory settings
- **Remediation code**: exact Python code to fix each issue
- **Priority matrix**: issues ranked by risk and effort
## Scope Options
| Scope | What It Covers |
|-------|---------------|
| **Full** (default) | All tasks, config, beat, workers |
| **Single task** | Deep analysis of one task definition |
| **Config only** | Broker, backend, serialization, worker settings |
| **Beat only** | Periodic task schedule analysis |
| **Retry audit** | Retry policies across all tasks |
| **Security** | Serialization, auth, and message signing |
## Broker Support
The agent adapts recommendations based on the broker detected:
| Feature | RabbitMQ | Redis | SQS |
|---------|----------|-------|-----|
| Priority queues | Native support | Limited (0-9) | Not supported |
| Dead letter exchange | Native DLX | Manual implementation | Native DLQ |
| Message persistence | Per-message | AOF/RDB | Always persistent |
| Task routing | Exchange/routing key | Queue name only | Queue URL |
| Visibility timeout | ack_late + prefetch | visibility_timeout | Default 30s |
| Max message size | 128 MB default | 512 MB | 256 KB (use S3) |
## Tips for Best Results
- Point the agent at your entire project for full task discovery
- Include celeryconfig.py or Django settings for configuration analysis
- Share Flower metrics or task logs for runtime behavior analysis
- Run before major deployments or scaling events
- Combine with broker monitoring to correlate config issues with runtime failures
- For Django projects, include settings.py and all apps with tasks.py files
don't have the plugin yet? install it then click "run inline in claude" again.