Retry com Backoff Exponencial e Jitter em Python Puro: O Mecanismo Que Faz Sua Automacao Sobreviver a APIs Instaveis Sem Virar Um Ataque DDoS Acidental

PorOlivetto

Voce ja viu isso acontecer: sua automacao faz uma chamada pra uma API externa, recebe um erro 503, e simplesmente morre. Ou pior – entra em loop infinito tentando de novo a cada 100ms e acaba sendo bloqueada pelo rate limiter do servidor. Nos dois casos, voce perdeu.

O problema nao e o erro. APIs caem. Servidores reiniciam. Redes oscilam. O problema e como sua automacao reage quando isso acontece. E a resposta certa quase sempre e: tente de novo, espere mais a cada tentativa, e adicione aleatoriedade pra nao sincronizar com outros clientes.

Isso se chama Retry com Backoff Exponencial e Jitter. E o padrao usado por AWS, Google Cloud, Kubernetes, e basicamente todo sistema distribuido serio. E hoje voce vai implementar do zero, em Python puro, sem bibliotecas externas.

O Problema da Retentativa Ingenua

Antes de mostrar a solucao, vamos ver o que nao funciona. Aqui esta o retry que 90% dos desenvolvedores escrevem na pressa:

import urllib.request
import time

def fetch_ingenuo(url, max_retries=5):
    """Retry sem backoff - vai fritar o servidor."""
    for tentativa in range(max_retries):
        try:
            req = urllib.request.Request(url)
            with urllib.request.urlopen(req, timeout=10) as resp:
                return resp.read()
        except Exception as e:
            print(f"Tentativa {tentativa + 1} falhou: {e}")
            time.sleep(1)  # Espera fixa de 1s - o problema esta aqui
    raise RuntimeError(f"Falhou apos {max_retries} tentativas")

Funciona? Talvez. Mas tem tres problemas graves:

  1. Espera fixa: Se o servidor esta sobrecarregado, 100 clientes retryando a cada 1 segundo criam um pico sincronizado que impede a recuperacao.
  2. Sem jitter: Todos os clientes retryam ao mesmo tempo. Isso se chama “thundering herd” – e derruba servidores.
  3. Sem limite superior: Se voce escala pra 1000 workers, cada um retryando a cada 1s, voce e o ataque DDoS.

A Matematica do Backoff Exponencial

A ideia e simples: a cada tentativa, dobre o tempo de espera. Assim:

Tentativa 1: espera 1s
Tentativa 2: espera 2s
Tentativa 3: espera 4s
Tentativa 4: espera 8s
Tentativa 5: espera 16s

A formula e: delay = base_delay * 2^tentativa. Isso da tempo pro servidor se recuperar e reduz a pressao progressivamente.

Mas tem um problema: se 50 clientes comecam ao mesmo tempo, todos vao esperar 1s, depois 2s, depois 4s… e todos vao retryar nos mesmos instantes. E aqui que entra o jitter.

Jitter: A Aleatoriedade Que Salva Servidores

Jitter e simplesmente adicionar aleatoriedade ao delay. Em vez de esperar exatamente 4 segundos, voce espera algo entre 0 e 4 segundos. Isso dessincroniza os clientes e distribui a carga no tempo.

Existem duas estrategias principais:

  • Full Jitter: delay = random(0, base * 2^tentativa). Totalmente aleatorio dentro do intervalo.
  • Equal Jitter: delay = (base * 2^tentativa / 2) + random(0, base * 2^tentativa / 2). Metade fixa, metade aleatoria.

O Full Jitter funciona melhor na pratica porque maximiza a dessincronizacao. E o que a AWS recomenda no blog deles sobre arquitetura de sistemas distribuidos.

Implementacao Completa em Python Puro

Aqui esta a implementacao que eu uso em producao. Sem dependencias externas, so biblioteca padrao:

import urllib.request
import urllib.error
import time
import random
from typing import Optional, Tuple, Type


class RetryConfig:
    """Configuracao centralizada para retry com backoff exponencial."""

    def __init__(
        self,
        max_retries: int = 5,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        jitter: bool = True,
        exponential_base: float = 2.0,
        retryable_exceptions: Optional[Tuple[Type[Exception], ...]] = None,
        retryable_status_codes: Optional[Tuple[int, ...]] = None,
    ):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.jitter = jitter
        self.exponential_base = exponential_base
        self.retryable_exceptions = retryable_exceptions or (
            urllib.error.URLError,
            ConnectionError,
            TimeoutError,
            OSError,
        )
        self.retryable_status_codes = retryable_status_codes or (
            429,  # Too Many Requests
            500,  # Internal Server Error
            502,  # Bad Gateway
            503,  # Service Unavailable
            504,  # Gateway Timeout
        )

    def get_delay(self, attempt: int) -> float:
        """Calcula o delay com backoff exponencial + jitter."""
        delay = self.base_delay * (self.exponential_base ** attempt)
        delay = min(delay, self.max_delay)
        if self.jitter:
            delay = random.uniform(0, delay)
        return delay


class RetryWithBackoff:
    """
    Cliente HTTP com retry automatico usando backoff exponencial e jitter.
    Usa apenas urllib da biblioteca padrao - zero dependencias externas.
    """

    def __init__(self, config: Optional[RetryConfig] = None):
        self.config = config or RetryConfig()
        self.attempts_log = []

    def _should_retry(self, exception, status_code) -> bool:
        """Decide se deve retryar baseado na excecao ou status code."""
        if exception and isinstance(exception, self.config.retryable_exceptions):
            return True
        if status_code and status_code in self.config.retryable_status_codes:
            return True
        return False

    def _log_attempt(self, attempt, success, delay, error):
        """Registra cada tentativa para observabilidade."""
        entry = {
            "attempt": attempt,
            "success": success,
            "delay_before": delay,
            "error": error,
            "timestamp": time.time(),
        }
        self.attempts_log.append(entry)
        status = "sucesso" if success else error
        print(f"[Retry] Tentativa {attempt}: {status} | delay: {delay:.2f}s")

    def fetch(self, url, method="GET", headers=None,
              data=None, timeout=10.0):
        """
        Faz requisicao HTTP com retry automatico.
        Returns: Tuple com (status_code, response_body, response_headers)
        """
        last_exception = None
        delay = 0.0

        for attempt in range(self.config.max_retries + 1):
            try:
                req = urllib.request.Request(
                    url, method=method,
                    headers=headers or {}, data=data
                )
                with urllib.request.urlopen(req, timeout=timeout) as resp:
                    status = resp.status
                    body = resp.read()
                    resp_headers = dict(resp.headers)

                    if self._should_retry(None, status):
                        if attempt < self.config.max_retries:
                            delay = self.config.get_delay(attempt)
                            self._log_attempt(
                                attempt + 1, False, delay,
                                f"status {status}"
                            )
                            time.sleep(delay)
                            continue

                    self._log_attempt(attempt + 1, True, delay, None)
                    return status, body, resp_headers

            except urllib.error.HTTPError as e:
                last_exception = e
                if not self._should_retry(e, e.code):
                    self._log_attempt(
                        attempt + 1, False, 0,
                        f"HTTP {e.code} (nao retryable)"
                    )
                    raise

                if attempt < self.config.max_retries:
                    delay = self.config.get_delay(attempt)

                    # Respeita Retry-After header se presente
                    retry_after = e.headers.get("Retry-After")
                    if retry_after:
                        try:
                            delay = max(delay, float(retry_after))
                        except ValueError:
                            pass

                    self._log_attempt(
                        attempt + 1, False, delay, f"HTTP {e.code}"
                    )
                    time.sleep(delay)
                else:
                    self._log_attempt(
                        attempt + 1, False, 0,
                        f"HTTP {e.code} (esgotou retries)"
                    )

            except Exception as e:
                last_exception = e
                if not self._should_retry(e, None):
                    self._log_attempt(
                        attempt + 1, False, 0,
                        f"{type(e).__name__} (nao retryable)"
                    )
                    raise

                if attempt < self.config.max_retries:
                    delay = self.config.get_delay(attempt)
                    self._log_attempt(
                        attempt + 1, False, delay,
                        f"{type(e).__name__}: {e}"
                    )
                    time.sleep(delay)
                else:
                    self._log_attempt(
                        attempt + 1, False, 0,
                        f"{type(e).__name__} (esgotou retries)"
                    )

        raise RuntimeError(
            f"Falhou apos {self.config.max_retries + 1} tentativas. "
            f"Ultimo erro: {last_exception}"
        )

    def get_stats(self):
        """Retorna estatisticas das tentativas."""
        total = len(self.attempts_log)
        successes = sum(1 for a in self.attempts_log if a["success"])
        return {
            "total_attempts": total,
            "successes": successes,
            "failures": total - successes,
            "total_delay": sum(a["delay_before"] for a in self.attempts_log),
            "attempts": self.attempts_log,
        }

Uso na Pratica

Agora vamos usar isso num cenario real: consumir uma API que pode estar instavel.

# Exemplo 1: GET simples com retry
client = RetryWithBackoff(RetryConfig(max_retries=3, base_delay=0.5))

try:
    status, body, headers = client.fetch("https://api.github.com/zen")
    print(f"Status: {status}")
    print(f"Resposta: {body.decode()}")
except RuntimeError as e:
    print(f"Erro final: {e}")

# Exemplo 2: Configuracao mais robusta
config = RetryConfig(
    max_retries=5,
    base_delay=1.0,
    max_delay=30.0,  # Nunca espera mais que 30s
    jitter=True,
)
client = RetryWithBackoff(config)

try:
    status, body, headers = client.fetch(
        url="https://httpbin.org/status/503",
        method="GET",
        headers={"User-Agent": "MinhaAutomacao/1.0"},
        timeout=5.0,
    )
except RuntimeError as e:
    print(f"Erro: {e}")
    print(f"Estatisticas: {client.get_stats()}")

Decorator: Reutilizando em Qualquer Funcao

Se voce quer aplicar retry em funcoes que nao sao HTTP (banco de dados, filesystem, etc.), aqui esta um decorator generico:

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


def retry_with_backoff(
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: bool = True,
    retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,),
):
    """
    Decorator que adiciona retry com backoff exponencial a qualquer funcao.

    Uso:
        @retry_with_backoff(max_retries=3, base_delay=0.5)
        def minha_funcao_fragil():
            ...
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None

            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except retryable_exceptions as e:
                    last_exception = e

                    if attempt >= max_retries:
                        break

                    delay = base_delay * (2 ** attempt)
                    delay = min(delay, max_delay)

                    if jitter:
                        delay = random.uniform(0, delay)

                    print(
                        f"[Retry] {func.__name__} tentativa "
                        f"{attempt + 1} falhou: {e} | "
                        f"aguardando {delay:.2f}s"
                    )
                    time.sleep(delay)

            raise RuntimeError(
                f"{func.__name__} falhou apos {max_retries + 1} "
                f"tentativas. Ultimo erro: {last_exception}"
            ) from last_exception

        return wrapper
    return decorator


# Uso pratico
@retry_with_backoff(
    max_retries=3,
    base_delay=0.5,
    retryable_exceptions=(ConnectionError, TimeoutError, OSError),
)
def salva_no_banco(dados):
    """Simula operacao de banco que pode falhar temporariamente."""
    import random
    if random.random() < 0.7:  # 70% de chance de falhar
        raise ConnectionError("Banco temporariamente indisponivel")
    return "Dados salvos com sucesso!"


# Testando
try:
    resultado = salva_no_banco({"user": "alisson", "action": "login"})
    print(resultado)
except RuntimeError as e:
    print(f"Falhou definitivamente: {e}")

O Perrengue do Olivetto

Eu implementei retry sem jitter numa automacao que monitorava 50 endpoints a cada minuto. Tudo funcionava lindo – ate que um dia os 50 endpoints cairam ao mesmo tempo (o provedor reiniciou o load balancer). Meus 50 workers retryaram sincronizados, todos no mesmo instante, e o provedor interpretou como ataque DDoS. Bloqueou meu IP por 24 horas.

A licao: retry sem jitter nao e so “menos eficiente” – pode te transformar no atacante. Depois disso, jitter virou obrigatorio em qualquer retry que eu escrevo. E adicionar e literalmente uma linha: delay = random.uniform(0, delay).

Quando NAO Usar Retry

Nem todo erro merece retry. Aqui esta o guia rapido:

  • NAO retrye: 400 (Bad Request), 401 (Unauthorized), 403 (Forbidden), 404 (Not Found), 422 (Unprocessable Entity). Esses erros nao vao se resolver sozinhos.
  • Retrye: 429 (Rate Limited), 500, 502, 503, 504. Esses sao temporarios por natureza.
  • Retrye: TimeoutError, ConnectionError, URLError. Problemas de rede se resolvem com tempo.
  • NAO retrye: ValueError, TypeError, KeyError. Sao bugs no seu codigo. Corrija, nao retrye.

Comparacao Visual: Com vs Sem Jitter

Pra quem gosta de ver os numeros, aqui esta uma simulacao de 10 clientes retryando simultaneamente:

import random


def simula_sem_jitter(num_clientes=10, max_retries=4, base_delay=1.0):
    """Mostra quando cada cliente retrya sem jitter."""
    print("=== SEM JITTER (todos sincronizados) ===")
    for attempt in range(max_retries):
        delay = base_delay * (2 ** attempt)
        print(f"Tentativa {attempt + 1}: todos retryam em {delay:.0f}s")
    print()


def simula_com_jitter(num_clientes=10, max_retries=4, base_delay=1.0):
    """Mostra quando cada cliente retrya com full jitter."""
    print("=== COM FULL JITTER (dessincronizado) ===")
    for attempt in range(max_retries):
        max_d = base_delay * (2 ** attempt)
        tempos = [
            f"{random.uniform(0, max_d):.1f}s"
            for _ in range(num_clientes)
        ]
        print(f"Tentativa {attempt + 1}: {', '.join(tempos)}")


simula_sem_jitter()
simula_com_jitter()

O resultado fala por si: sem jitter, todos batem no servidor nos mesmos instantes. Com jitter, a carga se distribui no tempo.

Checklist de Producao

Antes de colocar seu retry em producao, verifique:

  • Todas as chamadas externas (HTTP, banco, fila) tem retry?
  • O jitter esta habilitado?
  • Existe um max_delay pra evitar esperas absurdas?
  • Voce respeita o header Retry-After quando o servidor envia?
  • Erros nao-retryable (4xx exceto 429) sao tratados diferente?
  • Existe logging/observabilidade pra saber quantos retries estao acontecendo?
  • O timeout individual de cada tentativa e razoavel?

Proximo Passo

Agora que voce tem retry com backoff exponencial e jitter funcionando, o proximo nivel e combinar isso com um Circuit Breaker (que ja cobrimos aqui no blog). O Circuit Breaker detecta quando um servico esta consistentemente falhando e para de tentar por um tempo, evitando desperdicio de recursos. Retry cuida de falhas temporarias; Circuit Breaker cuida de falhas prolongadas. Juntos, eles formam uma defesa solida pra qualquer automacao que depende de servicos externos.

E voce? Ja teve uma automacao que morreu silenciosamente porque nao tinha retry? Ou pior – que era o problema porque retryava errado demais? Conta aqui nos comentarios qual foi o perrengue. Os melhores bugs viram os melhores posts.

Posts Similares