Dead Letter Queue em Python Puro: O Cemiterio Que Captura Mensagens Que Falharam e Garante Que Nenhuma Tarefa Se Perca Para Sempre (Sem RabbitMQ, Sem SQS)
Voce ja acordou de manha, olhou pro log e viu aquela linha gelada: TaskFailedError: payment webhook #8847 - max retries exceeded. O cliente pagou. O dinheiro caiu. Mas o sistema nunca registrou o pedido porque a terceira tentativa de chamar a API de confirmacao falhou, e o framework de fila simplesmente… descartou a mensagem. Para sempre.
Isso aconteceu comigo numa sexta-feira as 23h. O cliente ligou no sabado de manha furioso. Eu passei o fim de semana reconstruindo pedidos perdidos manualmente a partir de logs espalhados. Nunca mais.
Uma Dead Letter Queue (DLQ) e exatamente o que o nome sugere: um cemiterio organizado para mensagens mortas. Quando uma tarefa falha apos todas as tentativas, em vez de sumir no void, ela vai pra DLQ – com metadados, stack trace, payload original e timestamp. De la, voce pode inspecionar, corrigir a causa raiz e reprocessar com um comando.
Hoje vamos construir uma DLQ completa em Python puro. Zero dependencias externas. Zero RabbitMQ, zero SQS, zero Redis. So stdlib. E no final, voce vai ter um sistema que:
- Aceita tarefas com retry configuravel e backoff exponencial
- Manda tarefas mortas pra DLQ automaticamente apos N falhas
- Persiste tudo em disco (JSON Lines) – sobrevive a restarts
- Permite inspecionar, reprocessar e purgar mensagens mortas
- Expoe metricas: taxa de mortalidade, tempo medio na DLQ, top erros
O Problema Que Ninguem Te Conta Sobre Filas Simples
A maioria dos tutoriais de fila em Python mostra algo assim:
# A fila "feliz" que todo tutorial mostra
import queue, threading
q = queue.Queue()
def worker():
while True:
task = q.get()
process(task) # Se falhar? A mensagem some.
q.task_done()
threading.Thread(target=worker, daemon=True).start()
q.put({"order_id": 8847, "action": "confirm_payment"})Funciona? Funciona. Ate o dia que a API de pagamento retorna 503 tres vezes seguidas. A excecao explode dentro do process(), o worker pega a proxima tarefa, e o pedido #8847? Virou fumaca. Nao tem log estruturado, nao tem payload salvo, nao tem como reprocessar.
A solucao “profissional” e jogar RabbitMQ ou AWS SQS no problema. Mas se voce esta rodando automacoes em uma VPS de 5 dolares, ou um script de ETL local, ou um bot que processa webhooks – subir um broker inteiro pra gerenciar meia duzia de tarefas e como contratar um seguranca pra vigiar sua bicicleta.
A Arquitetura: Tres Camadas, Zero Magica
Nossa DLQ tem tres componentes:
- TaskRunner – executa tarefas com retry, backoff e captura de erros
- DeadLetterQueue – persiste tarefas mortas em disco com metadados completos
- DLQManager – interface para inspecionar, reprocessar e gerar metricas
Vamos construir de baixo pra cima.
Camada 1: O TaskRunner com Retry Inteligente
import json
import time
import uuid
import random
import threading
from pathlib import Path
from datetime import datetime, timezone
from dataclasses import dataclass, field, asdict
from typing import Callable, Any, Optional
@dataclass
class Task:
id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
payload: dict = field(default_factory=dict)
created_at: str = field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
attempts: int = 0
max_retries: int = 3
errors: list = field(default_factory=list)
def is_dead(self) -> bool:
return self.attempts >= self.max_retries
def record_failure(self, error: str):
self.attempts += 1
self.errors.append({
"attempt": self.attempts,
"error": error,
"timestamp": datetime.now(timezone.utc).isoformat()
})
def to_dict(self) -> dict:
return asdict(self)
@classmethod
def from_dict(cls, data: dict) -> "Task":
return cls(**data)
class TaskRunner:
def __init__(self, base_delay: float = 1.0, max_delay: float = 60.0):
self.base_delay = base_delay
self.max_delay = max_delay
def _backoff(self, attempt: int) -> float:
delay = min(self.base_delay * (2 ** (attempt - 1)), self.max_delay)
jitter = random.uniform(0, delay * 0.5)
return delay + jitter
def execute(self, task, handler):
try:
handler(task.payload)
return True, None
except Exception as e:
error_msg = f"{type(e).__name__}: {e}"
task.record_failure(error_msg)
if not task.is_dead():
wait = self._backoff(task.attempts)
time.sleep(wait)
return self.execute(task, handler)
return False, error_msgO detalhe importante aqui e o jitter. Sem ele, se 50 tarefas falharem ao mesmo tempo, todas vao retryar ao mesmo tempo – e vao falhar de novo juntas. O jitter espalha as tentativas no tempo. E a diferenca entre um sistema que se recupera e um que entra em espiral da morte.
Camada 2: A Dead Letter Queue Persistente
A DLQ precisa sobreviver a restarts. Se o processo cair, as mensagens mortas nao podem morrer de verdade. Usamos JSON Lines – um JSON por linha – porque e append-only, facil de ler com grep, e nao precisa carregar o arquivo inteiro na memoria.
class DeadLetterQueue:
def __init__(self, path: str = "dlq.jsonl"):
self.path = Path(path)
self.lock = threading.Lock()
self.path.parent.mkdir(parents=True, exist_ok=True)
def bury(self, task, original_error: str) -> str:
record = {
"dlq_id": uuid.uuid4().hex[:12],
"task": task.to_dict(),
"buried_at": datetime.now(timezone.utc).isoformat(),
"final_error": original_error,
"status": "dead",
}
with self.lock:
with open(self.path, "a") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
return record["dlq_id"]
def list_dead(self, limit: int = 100) -> list:
if not self.path.exists():
return []
records = []
with self.lock:
for line in self.path.read_text().strip().split("\n"):
if line.strip():
record = json.loads(line)
if record.get("status") == "dead":
records.append(record)
return list(reversed(records[-limit:]))
def count(self) -> int:
return len(self.list_dead(limit=999999))
def _rewrite(self, records):
with open(self.path, "w") as f:
for r in records:
f.write(json.dumps(r, ensure_ascii=False) + "\n")
def remove(self, dlq_id: str):
with self.lock:
lines = self.path.read_text().strip().split("\n")
records = [json.loads(l) for l in lines if l.strip()]
records = [r for r in records if r["dlq_id"] != dlq_id]
self._rewrite(records)
def purge_all(self) -> int:
with self.lock:
if not self.path.exists():
return 0
lines = self.path.read_text().strip().split("\n")
records = [json.loads(l) for l in lines if l.strip()]
dead = sum(1 for r in records if r.get("status") == "dead")
self._rewrite([])
return deadPor que JSON Lines e nao SQLite? Para volumes pequenos (ate uns 100k registros), JSON Lines e mais simples de debugar – voce abre o arquivo no editor, da grep "payment" dlq.jsonl, e pronto. Se precisar escalar, trocar por SQLite e um refactor de 20 minutos, nao uma reescrita.
Camada 3: O DLQManager – Inspecionar, Reprocessar, Medir
class DLQManager:
def __init__(self, dlq_path: str = "dlq.jsonl"):
self.dlq = DeadLetterQueue(dlq_path)
self.runner = TaskRunner()
def submit(self, payload, handler, max_retries=3):
task = Task(payload=payload, max_retries=max_retries)
success, error = self.runner.execute(task, handler)
if success:
return {"status": "success", "task_id": task.id}
else:
dlq_id = self.dlq.bury(task, error)
return {
"status": "dead", "task_id": task.id,
"dlq_id": dlq_id, "attempts": task.attempts,
"errors": task.errors,
}
def inspect(self, limit=20):
return self.dlq.list_dead(limit)
def reprocess(self, dlq_id, handler):
dead = self.dlq.list_dead(limit=999999)
target = next((t for t in dead if t["dlq_id"] == dlq_id), None)
if not target:
return {"status": "not_found", "dlq_id": dlq_id}
task = Task.from_dict(target["task"])
task.attempts = 0
task.errors = []
success, error = self.runner.execute(task, handler)
if success:
self.dlq.remove(dlq_id)
return {"status": "recovered", "dlq_id": dlq_id}
else:
self.dlq.remove(dlq_id)
new_id = self.dlq.bury(task, error)
return {"status": "still_dead", "new_dlq_id": new_id}
def reprocess_all(self, handler):
dead = self.dlq.list_dead(limit=999999)
recovered, still_dead = 0, 0
for record in dead:
r = self.reprocess(record["dlq_id"], handler)
if r["status"] == "recovered":
recovered += 1
else:
still_dead += 1
return {"recovered": recovered, "still_dead": still_dead}
def metrics(self):
dead = self.dlq.list_dead(limit=999999)
total = len(dead)
if total == 0:
return {"total_dead": 0, "error_breakdown": {}}
error_counts = {}
total_attempts = 0
for record in dead:
task = record["task"]
total_attempts += task["attempts"]
for err in task["errors"]:
key = err["error"].split(":")[0]
error_counts[key] = error_counts.get(key, 0) + 1
return {
"total_dead": total,
"error_breakdown": dict(sorted(
error_counts.items(), key=lambda x: -x[1]
)),
"avg_attempts_per_task": round(total_attempts / total, 1),
"oldest_buried": dead[-1]["buried_at"] if dead else None,
}Testando Com Cenario Real: Webhook de Pagamento
Vamos simular o cenario que me deu pesadelos: um handler que confirma pagamentos via API externa, e essa API esta instavel.
# Simula uma API instavel - falha 80% das vezes
call_count = {"n": 0}
def unstable_payment_api(payload: dict):
call_count["n"] += 1
if call_count["n"] % 5 != 0:
raise ConnectionError(
f"API timeout after 30s (call #{call_count['n']})"
)
return {"status": "confirmed", "order_id": payload["order_id"]}
# === Uso ===
manager = DLQManager(dlq_path="/tmp/dlq_demo.jsonl")
results = []
for i in range(10):
result = manager.submit(
payload={"order_id": f"ORD-{i+1:04d}", "amount": 99.90 + i},
handler=unstable_payment_api,
max_retries=3,
)
results.append(result)
success = sum(1 for r in results if r["status"] == "success")
dead = sum(1 for r in results if r["status"] == "dead")
print(f"Sucesso: {success} | Mortas: {dead}")
print(f"Metricas: {json.dumps(manager.metrics(), indent=2)}")
for task in manager.inspect():
print(f" {task['dlq_id']} | "
f"{task['task']['payload']['order_id']} | "
f"{task['final_error']}")Saida tipica:
Sucesso: 2 | Mortas: 8
Metricas: {
"total_dead": 8,
"error_breakdown": {"ConnectionError": 24},
"avg_attempts_per_task": 3.0,
"oldest_buried": "2026-07-27T21:00:01+00:00"
}Agora a API voltou ao normal. Hora de reprocessar:
# Simula API estavel agora
call_count["n"] = 4
result = manager.reprocess_all(handler=unstable_payment_api)
print(f"Recovered: {result['recovered']} | "
f"Ainda mortas: {result['still_dead']}")Integracao com Loop de Processamento Continuo
Na vida real, voce nao submete tarefas uma por uma. Voce tem um loop consumindo de alguma fonte (banco, API, arquivo). Aqui esta o pattern que uso em producao:
import signal
class ContinuousProcessor:
def __init__(self, dlq_path="dlq.jsonl"):
self.manager = DLQManager(dlq_path)
self.running = True
signal.signal(signal.SIGINT, self._shutdown)
signal.signal(signal.SIGTERM, self._shutdown)
def _shutdown(self, signum, frame):
print("\nShutdown received. Finishing current task...")
self.running = False
def run(self, source, handler, poll_interval=5.0, max_retries=3):
"""
source: callable que retorna lista de payloads
handler: callable que processa cada payload
"""
while self.running:
try:
payloads = source()
except Exception as e:
print(f"Source error: {e}")
time.sleep(poll_interval)
continue
for payload in payloads:
if not self.running:
break
result = self.manager.submit(
payload, handler, max_retries
)
if result["status"] == "dead":
print(f"Buried: {result['dlq_id']}")
# Auto-reprocess periodico
if self.manager.dlq.count() > 0:
r = self.manager.reprocess_all(handler)
if r["recovered"] > 0:
print(f"Auto-recovered: {r['recovered']}")
time.sleep(poll_interval)
print(f"Final: {self.manager.metrics()}")O ContinuousProcessor faz tres coisas que salvam vidas:
- Graceful shutdown – nao perde a tarefa atual quando voce da Ctrl+C
- Auto-reprocess periodico – tenta ressuscitar tarefas mortas automaticamente
- Metricas no shutdown – voce sabe exatamente o estado quando o processo morre
O Perrengue do Olivetto
A primeira versao deste sistema usava pickle pra serializar tarefas na DLQ. Funcionava perfeitamente. Ate o dia que mudei a classe Task, adicionei um campo, e tentei reprocessar tarefas antigas. AttributeError. Cento e quarenta e duas tarefas mortas que nao podiam ser desserializadas. Perdi 3 horas escrevendo um script de migracao que lia pickle antigo e convertia pra dict.
Licao: nunca use pickle para persistencia de longo prazo. JSON e verboso, chato, e vai funcionar daqui a 10 anos quando voce mudar tudo. Pickle e rapido, conveniente, e vai te trair na primeira refatoracao.
A segunda licao foi mais sutil: eu nao tinha jitter no backoff. Resultado? Quando a API caia, todas as 50 tarefas em fila retryavam exatamente ao mesmo tempo, sobrecarregavam a API de novo, e morriam de novo – num ciclo infinito. O jitter quebra essa sincronizacao maligna. E uma linha de codigo que evita um desastre.
Quando Usar (e Quando Nao Usar)
Use esta DLQ quando:
- Volume e moderado (ate ~10k tarefas/hora)
- Voce precisa de algo que funciona agora, sem infraestrutura
- O custo de perder uma tarefa e alto (pagamentos, notificacoes legais)
- Voce quer debugar falhas inspecionando payloads reais, nao metricas abstratas
Nao use quando:
- Precisa de garantias exactly-once com multiplos consumidores (use Kafka + DLQ nativa)
- Volume e massivo (100k+/hora – use SQS ou RabbitMQ com dead-letter exchange)
- Precisa de TTL por mensagem ou delay queues nativas
Proximos Passos
Tres melhorias que transformam isso num sistema mais robusto:
- Alertas automaticos – se a DLQ passar de N tarefas, manda webhook pra Slack/Discord
- Dashboard web – uma interface Flask/FastAPI de 50 linhas pra inspecionar e reprocessar pelo browser
- DLQ por categoria de erro – erros de rede retryam mais; erros de validacao morrem rapido
O codigo completo deste post esta pronto pra copiar e colar. Adapta pro teu caso, quebra coisas, me conta o que deu errado.
E voce? Ja perdeu dados porque uma fila descartou mensagens silenciosamente? Como resolveu? Conta ai nos comentarios que eu quero saber se alguem ja viu coisa pior que o meu pickle-gate de 142 tarefas mortas.
