PythonhardPython

Retry Decorator with Jitter

Retry failed operations with exponential backoff and random jitter to prevent thundering herd.

01

The problem

You need to retry distributed operations while preventing thundering herd problem.

02

The solution

Python
import functools
import random
import time
from typing import Tuple, Type

def retry_with_jitter(
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    backoff: float = 2.0,
    jitter: float = 0.1,
    exceptions: Tuple[Type[Exception], ...] = (Exception,)
):
    """Retry decorator with exponential backoff and jitter."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    
                    if attempt == max_attempts - 1:
                        break
                    
                    # Calculate backoff with jitter
                    delay = min(
                        base_delay * (backoff ** attempt),
                        max_delay
                    )
                    
                    # Add jitter (±jitter%)
                    jitter_amount = delay * jitter
                    delay_with_jitter = delay + random.uniform(-jitter_amount, jitter_amount)
                    delay_with_jitter = max(0, delay_with_jitter)
                    
                    time.sleep(delay_with_jitter)
            
            raise last_exception
        return wrapper
    return decorator

03

Parameters

max_attemptsint

Maximum number of retry attempts

base_delayfloat

Initial delay in seconds

max_delayfloat

Maximum delay in seconds

backofffloat

Exponential backoff multiplier

jitterfloat

Jitter factor (e.g., 0.1 for ±10%)

exceptionstuple

Exception types to retry

04

Put it to work

Example
import requests

@retry_with_jitter(
    max_attempts=5,
    base_delay=1.0,
    max_delay=30.0,
    jitter=0.2,
    exceptions=(requests.RequestException,)
)
def fetch_with_jitter(url):
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

# Will retry with random delays to prevent thundering herd
data = fetch_with_jitter('https://api.example.com/data')