Dead Letter Queue em Python Puro: A Quarentena Que Salva Mensagens Perdidas e Revela Bugs Invisiveis na Sua Pipeline
Você tem uma pipeline de processamento. Tudo funciona — até que uma mensagem malformada trava o worker inteiro. Sem aviso. Sem retry. Sem explicação. Só o silêncio de dados que desapareceram no void.
O Dead Letter Queue (DLQ) é o padrão que resolve isso. Toda mensagem que falha — seja por bug, timeout ou dado corrompido — vai parar numa fila de quarentena. Nada se perde. Tudo pode ser reprocessado. E o melhor: você consegue ver o que está falhando.
Sim, existe RabbitMQ, SQS, Kafka. Mas e se você quer entender o mecanismo por dentro — sem depender de infra externa, sem YAML de configuração, sem 47 microserviços? Python puro, filesystem local, e zero desculpas.
O Problema: Quando a Mensagem Morre, Ninguém Fica Sabendo
Pense num sistema de newsletter. Você dispara 10.000 e-mails por hora. De vez em quando, um e-mail retorna com bounce. A maioria das pipelines simplesmente ignora: o worker pega a mensagem, falha silenciosamente, e a mensagem some. Para sempre.
Agora multiplique isso por meses. Você perdeu 4.000 mensagens. Nenhuma delas foi reprocessada. Nenhuma delas gerou alerta. E quando o cliente pergunta “cadê meu e-mail de confirmação?”, a resposta é “não sei, sumiu”.
É aqui que o Dead Letter Queue entra. Ele não impede a falha — ele captura a falha. E capturar é o primeiro passo para resolver.
O Design: Pipeline + Quarentena + Replay
A arquitetura tem três peças:
- Pipeline principal: processa mensagens normalmente. Se falhar, manda pra DLQ em vez de descartar.
- Dead Letter Queue: armazena as mensagens falhas com metadata (motivo da falha, timestamp, número de tentativas).
- Replay Engine: permite reprocessar mensagens da DLQ individualmente ou em lote, depois que o bug foi corrigido.
Tudo isso com arquivos JSON no filesystem. Simples, auditável, e funciona até no seu notebook.
O Código: Pipeline com DLQ Integrada
Vamos construir tudo num único módulo. Sem dependências externas. Só json, os, time, threading e uuid — tudo da stdlib.
"""
dead_letter_queue.py -- Dead Letter Queue em Python Puro
Pipeline com quarentena, retry automatico e replay manual.
Zero dependencias externas.
"""
import json
import os
import time
import uuid
import threading
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Any, Optional
class DeadLetterQueue:
"""Armazena mensagens que falharam no processamento."""
def __init__(self, storage_dir: str = ".dlq"):
self.storage = Path(storage_dir)
self.storage.mkdir(exist_ok=True)
self._lock = threading.Lock()
def enqueue(self, message: dict, error: str, attempts: int = 1):
"""Coloca uma mensagem na quarentena com metadata."""
entry = {
"id": str(uuid.uuid4()),
"message": message,
"error": error,
"attempts": attempts,
"failed_at": datetime.now(timezone.utc).isoformat(),
"reprocessed": False,
}
filepath = self.storage / f"{entry['id']}.json"
with self._lock:
filepath.write_text(json.dumps(entry, indent=2))
return entry["id"]
def list_pending(self) -> list[dict]:
"""Lista todas as mensagens pendentes de replay."""
pending = []
for f in self.storage.glob("*.json"):
entry = json.loads(f.read_text())
if not entry.get("reprocessed"):
pending.append(entry)
return sorted(pending, key=lambda e: e["failed_at"])
def get_entry(self, entry_id: str) -> Optional[dict]:
"""Recupera uma entrada específica da DLQ."""
filepath = self.storage / f"{entry_id}.json"
if filepath.exists():
return json.loads(filepath.read_text())
return None
def mark_reprocessed(self, entry_id: str, success: bool):
"""Marca uma entrada como reprocessada (sucesso ou falha permanente)."""
filepath = self.storage / f"{entry_id}.json"
if not filepath.exists():
return
entry = json.loads(filepath.read_text())
entry["reprocessed"] = True
entry["reprocessed_at"] = datetime.now(timezone.utc).isoformat()
entry["reprocess_success"] = success
filepath.write_text(json.dumps(entry, indent=2))
def stats(self) -> dict:
"""Estatísticas da DLQ: total, pendentes, reprocessadas, taxa de erro."""
total = 0
pending = 0
reprocessed_ok = 0
reprocessed_fail = 0
errors = {}
for f in self.storage.glob("*.json"):
entry = json.loads(f.read_text())
total += 1
if entry.get("reprocessed"):
if entry.get("reprocess_success"):
reprocessed_ok += 1
else:
reprocessed_fail += 1
else:
pending += 1
err = entry.get("error", "unknown")
errors[err] = errors.get(err, 0) + 1
return {
"total": total,
"pending": pending,
"reprocessed_ok": reprocessed_ok,
"reprocessed_fail": reprocessed_fail,
"error_types": errors,
}
def purge(self, older_than_hours: Optional[float] = None):
"""Remove entradas da DLQ. Se older_than_hours, só as mais antigas."""
cutoff = None
if older_than_hours is not None:
cutoff = time.time() - (older_than_hours * 3600)
for f in self.storage.glob("*.json"):
entry = json.loads(f.read_text())
if not entry.get("reprocessed"):
continue # Nao purga pendentes
if cutoff is not None:
failed_ts = datetime.fromisoformat(entry["failed_at"]).timestamp()
if failed_ts > cutoff:
continue # Mais recente que o cutoff
f.unlink()
class Pipeline:
"""
Pipeline generico com Dead Letter Queue integrada.
Aceita um handler e automaticamente desvia falhas pra DLQ.
"""
def __init__(
self,
handler: Callable[[dict], Any],
dlq: DeadLetterQueue,
max_retries: int = 3,
retry_delay: float = 1.0,
):
self.handler = handler
self.dlq = dlq
self.max_retries = max_retries
self.retry_delay = retry_delay
def process(self, message: dict) -> bool:
"""
Processa uma mensagem. Se falhar após max_retries,
envia pra Dead Letter Queue.
Retorna True se processou com sucesso.
"""
last_error = None
for attempt in range(1, self.max_retries + 1):
try:
self.handler(message)
return True
except Exception as e:
last_error = str(e)
if attempt < self.max_retries:
time.sleep(self.retry_delay * attempt)
continue
# Esgotou retries -- vai pra DLQ
self.dlq.enqueue(message, last_error, attempts=self.max_retries)
return False
def process_batch(self, messages: list[dict]) -> dict:
"""Processa um lote e retorna estatisticas."""
results = {"success": 0, "failed": 0, "dlq": []}
for msg in messages:
if self.process(msg):
results["success"] += 1
else:
results["failed"] += 1
pending = self.dlq.list_pending()
if pending:
results["dlq"].append(pending[-1]["id"])
return resultsUsando: Exemplo Real com Newsletter
Agora vamos simular um sistema de envio de e-mails onde alguns endereços são inválidos e causam falhas:
"""
demo_dlq.py -- Exemplo pratico com pipeline de newsletter
"""
from dead_letter_queue import Pipeline, DeadLetterQueue
import random
import json
# Base de e-mails simulada
SUBSCRIBERS = [
{"email": "alice@email.com", "name": "Alice", "plan": "premium"},
{"email": "bob@email.com", "name": "Bob", "plan": "free"},
{"email": "INVALID_EMAIL", "name": "Charlie", "plan": "free"}, # vai falhar
{"email": "diana@email.com", "name": "Diana", "plan": "premium"},
{"email": "", "name": "Eve", "plan": "free"}, # vai falhar
{"email": "frank@email.com", "name": "Frank", "plan": "business"},
{"email": "bad@.com", "name": "Grace", "plan": "free"}, # vai falhar
]
def send_newsletter(subscriber: dict):
"""Handler que simula envio de e-mail."""
email = subscriber["email"]
# Validacao basica
if not email or "@" not in email or email.endswith(".com") and len(email.split("@")[1].split(".")) < 2:
raise ValueError(f"E-mail invalido: '{email}'")
# Simula timeout aleatorio (10% de chance)
if random.random() < 0.1:
raise TimeoutError(f"Timeout ao enviar para {email}")
print(f"Enviado para {email} ({subscriber['name']})")
# Inicializa DLQ e Pipeline
dlq = DeadLetterQueue(storage_dir=".dlq_newsletter")
pipeline = Pipeline(
handler=send_newsletter,
dlq=dlq,
max_retries=2,
retry_delay=0.5,
)
# Processa o lote
print("=== Processando lote de newsletter ===\n")
results = pipeline.process_batch(SUBSCRIBERS)
print(f"\nSucesso: {results['success']}")
print(f"Falhas (DLQ): {results['failed']}")
# Mostra estatisticas da DLQ
stats = dlq.stats()
print(f"\nEstatisticas da DLQ:")
print(f" Total na quarentena: {stats['total']}")
print(f" Pendentes: {stats['pending']}")
print(f" Tipos de erro: {json.dumps(stats['error_types'], indent=4)}")
# Lista as mensagens presas
print(f"\nMensagens na quarentena:")
for entry in dlq.list_pending():
msg = entry["message"]
print(f" ID: {entry['id'][:8]}... | Email: {msg.get('email', '?')} | Erro: {entry['error']}")A saída mostra exatamente o que entrou na quarentena, por que falhou, e permite investigar sem perder dados. Quando você corrige o bug de validação, basta rodar o replay:
O Replay: Recuperando o Que Ficou Preso
Depois de corrigir a validação de e-mail, você quer reprocessar tudo que ficou na DLQ. Sem reenviar os que já funcionaram:
"""
replay_dlq.py -- Reprocessa mensagens da Dead Letter Queue
"""
from dead_letter_queue import DeadLetterQueue, Pipeline
import json
def send_newsletter_fixed(subscriber: dict):
"""Handler corrigido com validacao robusta."""
email = subscriber.get("email", "")
if not email or "@" not in email:
raise ValueError(f"E-mail invalido: '{email}'")
domain = email.split("@")[1]
if "." not in domain or domain.startswith(".") or domain.endswith("."):
raise ValueError(f"Dominio invalido: '{domain}'")
print(f"REPLAY: Enviado para {email}")
# Inicializa com a mesma storage
dlq = DeadLetterQueue(storage_dir=".dlq_newsletter")
print(f"=== Replay da DLQ ===")
pending = dlq.list_pending()
print(f"Mensagens pendentes: {len(pending)}\n")
# Cria um pipeline temporario so pro replay
pipeline = Pipeline(
handler=send_newsletter_fixed,
dlq=dlq,
max_retries=1, # Replay so tenta uma vez
)
for entry in pending:
entry_id = entry["id"]
message = entry["message"]
try:
pipeline.handler(message) # Processa direto
dlq.mark_reprocessed(entry_id, success=True)
print(f" Replay OK: {message.get('email', '?')}")
except Exception as e:
dlq.mark_reprocessed(entry_id, success=False)
print(f" Replay falhou: {message.get('email', '?')} -- {e}")
print(f"\nEstatisticas finais:")
stats = dlq.stats()
print(json.dumps(stats, indent=2))O replay marca cada entrada como reprocessada, então as estatísticas refletem o histórico completo: o que falhou, o que foi recuperado, e o que é permanentemente perdido.
O Perrengue do Olivetto
Num projeto real de automação de disparos, eu não tinha DLQ. As mensagens falhavam silenciosamente — e eu só percebia quando o cliente ligava perguntando por que não recebeu o relatório. Gastei 3 dias reconstruindo dados manualmente porque não tinha quarentena.
O pior? O bug era uma validação de data que quebrava em fuso horário diferente. Uma linha de código. Mas sem DLQ, eu não tinha nenhuma maneira de saber quais mensagens tinham sido afetadas. Com DLQ, teria sido: rodar replay, corrigir o handler, pronto.
Regra que eu sigo agora: se pode falhar, tem que ter quarentena. Sem exceção.
Quando Usar (e Quando Não Usar)
Use quando:
- Você processa filas de mensagens, eventos ou jobs assíncronos
- Falhas silenciosas são inaceitáveis (financeiro, comunicação, auditoria)
- Você quer poder reprocessar dados depois de corrigir bugs
- Precisa de visibilidade sobre padrões de falha
Não use quando:
- Seu sistema é síncrono e o erro é retornado imediatamente ao caller
- Você já usa RabbitMQ/SQS com DLQ nativa configurada
- O volume é tão alto que o filesystem vira gargalo (nesse caso, vá pra Redis ou Kafka)
Para 90% dos projetos que eu vejo — especialmente automações, scripts de batch, e sistemas pequenos — o filesystem é mais que suficiente. E a simplicidade é uma feature, não um bug.
Extensões Possíveis
Este é o núcleo. A partir daqui você pode adicionar:
- Retry automático com backoff: antes de enviar pra DLQ, tenta novamente com delay exponencial
- Dead Letter Dashboard: endpoint HTTP que mostra estatísticas e permite replay via browser
- Alertas: notificação quando a DLQ ultrapassa um threshold
- TTL: expirar mensagens antigas automaticamente
- Storage em Redis: swap do filesystem por Redis pra sistemas distribuídos
Cada uma dessas extensões é um post separado. Mas o fundamental está aqui: nada se perde, tudo pode ser recuperado, e você sabe o que está falhando.
E você — já perdeu dados por falta de quarentena na sua pipeline? Conta aqui nos comentários qual foi o bug mais bizarro que já sumiu sem deixar rastros.
