Skip to content

Task Retry and Error Handling

DodaTech Updated 2026-06-28 7 min read

In this tutorial, you will learn about Task Retry and Error Handling. We cover key concepts, practical examples, and best practices to help you master this topic.

Handle errors in Celery tasks with automatic retries, exponential backoff, max retries limits, dead letter queues, and proper Exception Handling patterns.

What You Learn

You will learn how to use self.retry() for automatic retries, configure retry policies with exponential backoff, handle different exception types, use dead letter queues for failed tasks, and implement circuit breakers.

Why It Matters

Network failures, database timeouts, and temporary service disruptions are inevitable. Tasks that fail without retry are lost forever. Proper error handling ensures tasks eventually succeed or are safely routed to dead letter queues for manual inspection.

Real-World Use

Durga Antivirus Pro uses Celery retries for signature update tasks. If the update server is temporarily unreachable, the task retries with exponential backoff up to 5 times before routing to a dead letter queue for ops team review.

Basic Retry

from celery import Celery

app = Celery('retry', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def fetch_url(self, url):
    """Fetch a URL with automatic retry on failure."""
    import requests

    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return response.text[:100]
    except requests.RequestException as exc:
        raise self.retry(exc=exc, countdown=60)

from tasks import fetch_url

result = fetch_url.delay('https://httpbin.org/status/500')
try:
    data = result.get(timeout=120)
    print(f"Got: {data[:50]}")
except Exception as e:
    print(f"Task failed after retries: {e}")

Retry Policy

from celery import Celery

app = Celery('retry_policy', broker='redis://localhost:6379/0')

@app.task(
    bind=True,
    max_retries=5,
    default_retry_delay=30,
    acks_late=True,
    reject_on_worker_lost=True,
)
def task_with_retry_policy(self, item_id):
    """Process item with configurable retry policy."""

    try:
        print(f"Processing item {item_id} (attempt {self.request.retries + 1})")
        if item_id % 3 == 0:
            raise ValueError(f"Simulated error for item {item_id}")
        return f"Item {item_id} processed"
    except Exception as exc:
        retry_delay = 10 * (2 ** self.request.retries)
        print(f"Retrying in {retry_delay}s (attempt {self.request.retries + 1}/{self.max_retries})")
        raise self.retry(exc=exc, countdown=retry_delay)

# Test with some items that will fail
for i in range(5):
    result = task_with_retry_policy.delay(i)
    print(f"Task for item {i}: {result.id[:8]}")

Expected output:

Processing item 0 (attempt 1)
Retrying in 10s (attempt 1/5)
Processing item 0 (attempt 2)
Retrying in 20s (attempt 2/5)
...
Task for item 1: 550e8400...
Task for item 2: 6ba7b810...

Exponential Backoff with Jitter

import random
from celery import Celery

app = Celery('jitter', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=5)
def backoff_task(self):
    """Task with exponential backoff and jitter."""
    try:
        print(f"Attempt {self.request.retries + 1}")
        raise ConnectionError("Network timeout")
    except ConnectionError as exc:
        base_delay = 10 * (2 ** self.request.retries)
        jitter = random.uniform(0, base_delay * 0.5)
        delay = base_delay + jitter
        print(f"Retry delay: {delay:.1f}s (base: {base_delay}s, jitter: {jitter:.1f}s)")
        raise self.retry(exc=exc, countdown=delay)

Different Exceptions, Different Handling

from celery import Celery
import requests
import json

app = Celery('exceptions', broker='redis://localhost:6379/0')

class PermanentError(Exception):
    """Error that should not be retried."""
    pass

@app.task(bind=True, max_retries=3)
def robust_task(self, url):
    """Handle different exceptions differently."""

    try:
        response = requests.get(url, timeout=10)

        if response.status_code == 404:
            raise PermanentError(f"Resource not found: {url}")
        elif response.status_code == 429:
            retry_after = int(response.headers.get('Retry-After', 60))
            print(f"Rate limited, retrying in {retry_after}s")
            raise self.retry(countdown=retry_after)
        elif response.status_code >= 500:
            print(f"Server error, will retry with backoff")
            response.raise_for_status()

        return response.json()

    except PermanentError:
        print(f"Permanent error, not retrying: {url}")
        return {'error': str(e), 'url': url}
    except requests.ConnectionError as exc:
        print(f"Connection error, retrying...")
        raise self.retry(exc=exc, countdown=30)
    except requests.Timeout as exc:
        print(f"Timeout, retrying with longer timeout...")
        raise self.retry(exc=exc, countdown=60)

Dead Letter Pattern

Route permanently failed tasks to a dead letter queue:

from celery import Celery
import json

app = Celery('dead_letter', broker='redis://localhost:6379/0')

def send_to_dead_letter(task_name, args, kwargs, error, retries):
    """Log failed task info to dead letter store."""
    record = {
        'task': task_name,
        'args': args,
        'kwargs': kwargs,
        'error': str(error),
        'retries': retries,
        'timestamp': '2026-06-28T10:00:00Z'
    }
    print(f"DEAD LETTER: {json.dumps(record, indent=2)}")
    return record

@app.task(bind=True, max_retries=3)
def critical_task(self, data):
    """Task that uses dead letter on final failure."""
    try:
        if data == 'fail':
            raise ValueError("Processing failed")
        return f"Processed: {data}"
    except Exception as exc:
        if self.request.retries >= self.max_retries - 1:
            send_to_dead_letter(
                self.name,
                self.request.args,
                self.request.kwargs,
                exc,
                self.request.retries + 1
            )
        raise self.retry(exc=exc, countdown=5)

# Test with failing task
result = critical_task.delay('fail')
try:
    result.get(timeout=60)
except Exception as e:
    print(f"Final failure logged")

Expected output:

DEAD LETTER: {
  "task": "tasks.critical_task",
  "args": ["fail"],
  "kwargs": {},
  "error": "Processing failed",
  "retries": 3,
  "timestamp": "2026-06-28T10:00:00Z"
}
Final failure logged

Circuit Breaker Pattern

from celery import Celery
import time

app = Celery('circuit', broker='redis://localhost:6379/0')

class CircuitBreaker:
    """Simple circuit breaker."""

    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.last_failure_time = 0
        self.state = 'CLOSED'

    def call(self, func, *args, **kwargs):
        if self.state == 'OPEN':
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = 'HALF_OPEN'
            else:
                raise Exception("Circuit breaker is OPEN")

        try:
            result = func(*args, **kwargs)
            if self.state == 'HALF_OPEN':
                self.state = 'CLOSED'
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure_time = time.time()
            if self.failures >= self.failure_threshold:
                self.state = 'OPEN'
            raise e

circuit_breaker = CircuitBreaker(failure_threshold=3, reset_timeout=30)

@app.task(bind=True, max_retries=2)
def circuit_aware_task(self, data):
    """Task protected by circuit breaker."""
    try:
        if circuit_breaker.state == 'OPEN':
            raise self.retry(countdown=10)

        if data == 'bad':
            raise ValueError("Simulated error")

        return f"Success: {data}"
    except Exception as exc:
        raise self.retry(exc=exc, countdown=5)

Common Mistakes

1. Not Setting max_retries

Without max_retries, tasks retry indefinitely (default is 3). Set it explicitly. Infinite retries can mask permanent failures.

2. Retrying Permanent Errors

Do not retry errors that will never succeed (404 Not Found, validation errors, permission denied). Check exception type before retrying.

3. Using Constant Retry Delays

Constant delays cause thundering herd problems. Always use exponential backoff so the system has time to recover between retries.

4. Forgetting bind=True

self.retry() requires bind=True. Without it, self is not available and the task crashes with AttributeError.

5. Not Logging Retry Information

When debugging task failures, you need to know how many retries happened and what delays were used. Log retry count, delay, and exception details.

Practice Questions

1. How do you retry a Celery task?

Call self.retry(exc=exc, countdown=N) inside the task. This raises a Retry exception that Celery catches and reschedules the task.

2. What is exponential backoff?

Doubling the retry delay after each attempt: 10s, 20s, 40s, 80s. This prevents overwhelming failing services.

3. What is the default max_retries value?

  1. Tasks stop retrying after 3 failed attempts. Set explicitly to control retry behavior.

4. How do you implement a dead letter pattern for failed tasks?

Track retry count with self.request.retries. When it reaches max_retries, log the failure to a database or file instead of retrying again.

Challenge

Build a retry system for a payment processing task. Handle: insufficient funds (no retry), network timeout (retry 5x with exponential backoff), rate limit (retry after Retry-After header), and server error (retry 3x with circuit breaker). Log all failed payments to a dead letter queue.

FAQ

What happens when max_retries is reached?

The task fails with the original exception. The result backend stores the exception. result.get() re-raises it.

Can I manually retry a failed task?

Yes. Get the task ID from the error log, then call task.retry(args=..., kwargs=...). The task resets its retry count.

What is acks_late and how does it relate to retries?

acks_late=True means the broker sends the ack only after the task completes. If the worker crashes, the task is redelivered. Combined with retries, this provides at-least-once execution.

How do I set different retry delays for different exceptions?

Catch each exception type separately and call self.retry with different countdown values. Use permanent exceptions to skip retrying.

Does retrying affect task idempotency?

Yes. Retries cause the same task to execute multiple times. Make tasks idempotent by checking if the work was already done before processing.

Mini Project: Robust Retry System

# robust_tasks.py
from celery import Celery
import json
import time
import random

app = Celery('robust', broker='redis://localhost:6379/0')

DEAD_LETTER_QUEUE = []

@app.task(bind=True, max_retries=4, acks_late=True)
def process_with_retry(self, item_id, data):
    """Process item with comprehensive retry logic."""
    attempt = self.request.retries + 1

    print(f"[Attempt {attempt}/{self.max_retries}] Processing item {item_id}")

    try:
        if random.random() < 0.4:
            raise ConnectionError("Simulated transient failure")
        if item_id == 'fail_permanent':
            raise ValueError("Invalid data, cannot process")

        result = {"item_id": item_id, "status": "success", "data": data}
        print(f"Item {item_id} processed successfully")
        return result

    except ValueError as exc:
        print(f"Permanent failure for {item_id}, not retrying")
        DEAD_LETTER_QUEUE.append({
            'item_id': item_id,
            'error': str(exc),
            'type': 'permanent',
            'attempts': attempt
        })
        return {"item_id": item_id, "status": "failed", "error": str(exc)}

    except ConnectionError as exc:
        if attempt >= self.max_retries:
            print(f"Exhausted retries for {item_id}")
            DEAD_LETTER_QUEUE.append({
                'item_id': item_id,
                'error': str(exc),
                'type': 'exhausted',
                'attempts': attempt
            })
            return {"item_id": item_id, "status": "dead_letter", "error": str(exc)}

        delay = 5 * (2 ** (attempt - 1)) + random.uniform(0, 5)
        print(f"Retrying in {delay:.1f}s")
        raise self.retry(exc=exc, countdown=delay)

# run_demo.py
from robust_tasks import process_with_retry, DEAD_LETTER_QUEUE
import time

items = [
    {'id': 'item_ok', 'data': 'good'},
    {'id': 'fail_permanent', 'data': 'bad'},
    {'id': 'item_transient', 'data': 'maybe'},
]

results = []
for item in items:
    result = process_with_retry.delay(item['id'], item['data'])
    results.append(result)
    print(f"Submitted {item['id']}: {result.id[:8]}")

time.sleep(30)

print(f"\nDead letter queue: {len(DEAD_LETTER_QUEUE)} items")
for entry in DEAD_LETTER_QUEUE:
    print(f"  {entry}")

What's Next

Now that you understand task retry, learn about task routing to direct tasks to specific workers, then explore task queues for organizing task execution.

← Previous Calling Celery Tasks — Complete Guide Next → Task Routing in Celery — Complete Guide

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Celery