Dead Letter Queues for Background Jobs
In this tutorial, you will learn about Dead Letter Queues for Background Jobs. We cover key concepts, practical examples, and best practices to help you master this topic.
Implement dead letter queues in job processing systems to isolate failed jobs, inspect failure causes, replay jobs after fixes, and prevent queue pollution.
What You Learn
You will learn how to implement dead letter queues, configure automatic routing of failed jobs, build inspection and replay tools, and set up alerts for dead letter activity.
Why It Matters
Failed jobs pile up in the main queue, blocking healthy jobs. Dead letter queues isolate failures, preserve the original job data for debugging, and provide a mechanism for recovery after fixing the root cause.
Real-World Use
DodaTech's payment worker routes failed jobs to a dead letter queue after 3 retries. Operations reviews the DLQ daily, fixes underlying issues, and replays corrected jobs. DLQ depth is monitored and alerts on spike.
Dead Letter Queue Architecture
flowchart LR
Q[Main Queue] --> W[Worker]
W -->|Success| D[Done]
W -->|Fail < 3| R[Retry Queue]
W -->|Fail = 3| DL[Dead Letter Queue]
DL --> I[Inspect]
I -->|Fix issue| RP[Replay]
I -->|Discard| TR[Trash]
RP --> Q
Basic Dead Letter Queue
import redis
import json
import time
r = redis.Redis()
class DeadLetterQueue:
def __init__(self, dlq_key='dead_letter'):
self.dlq_key = dlq_key
def add(self, job, error, queue_name, retry_count):
entry = {
'job': job,
'error': str(error),
'queue': queue_name,
'retry_count': retry_count,
'failed_at': time.time(),
'dlq_id': f"dlq-{time.time_ns()}",
}
r.lpush(self.dlq_key, json.dumps(entry))
r.hincrby('dlq_count', queue_name, 1)
return entry['dlq_id']
def replay(self, dlq_id, target_queue=None):
entries = []
while True:
data = r.rpop(self.dlq_key)
if not data:
break
entries.append(json.loads(data))
for entry in entries:
if entry['dlq_id'] == dlq_id:
queue = target_queue or entry['queue']
r.lpush(queue, json.dumps(entry['job']))
print(f"Replayed {dlq_id}")
else:
r.lpush(self.dlq_key, json.dumps(entry))
def replay_all(self, target_queue=None):
count = 0
while True:
data = r.rpop(self.dlq_key)
if not data:
break
entry = json.loads(data)
queue = target_queue or entry['queue']
r.lpush(queue, json.dumps(entry['job']))
count += 1
print(f"Replayed {count} jobs")
return count
def list_entries(self, limit=20):
entries = []
for i in range(limit):
data = r.lindex(self.dlq_key, i)
if data:
entries.append(json.loads(data))
return entries
def count(self):
return r.llen(self.dlq_key)
def clear(self):
r.delete(self.dlq_key)
dlq = DeadLetterQueue()
dlq.add({'task': 'payment', 'amount': 50}, 'Insufficient funds', 'payments', 3)
dlq.add({'task': 'email', 'to': 'bad-email'}, 'Invalid email', 'email', 3)
print(f"DLQ entries: {dlq.count()}")
entries = dlq.list_entries()
print(f"First entry error: {entries[0]['error']}")
dlq.replay_all(target_queue='retry_queue')
print(f"After replay: {dlq.count()}")
Expected output:
DLQ entries: 2
First entry error: Insufficient funds
Replayed 2 jobs
After replay: 0
Automatic Dead Letter Routing
import json
import time
class AutoDeadLetterRouter:
def __init__(self, max_retries=3):
self.max_retries = max_retries
self.dlq = []
self.retry_counts = {}
def process(self, job, handler):
job_id = job.get('id', 'unknown')
retry_key = f"{job_id}"
retry_count = self.retry_counts.get(retry_key, 0)
job['_retry_count'] = retry_count
try:
result = handler(job)
self.retry_counts.pop(retry_key, None)
print(f"Completed: {job_id}")
return {'status': 'success', 'result': result}
except Exception as e:
retry_count += 1
self.retry_counts[retry_key] = retry_count
if retry_count >= self.max_retries:
dlq_entry = {
'job': job,
'error': str(e),
'retry_count': retry_count,
'failed_at': time.time(),
}
self.dlq.append(dlq_entry)
self.retry_counts.pop(retry_key, None)
print(f"Dead letter: {job_id} after {retry_count} retries")
return {'status': 'dead_letter', 'error': str(e)}
else:
print(f"Retry {retry_count}/{self.max_retries}: {job_id}")
return {'status': 'retry', 'error': str(e)}
def get_dlq_entries(self):
return list(self.dlq)
def get_dlq_count(self):
return len(self.dlq)
def failing_handler(job):
if job.get('type') == 'fail':
raise ValueError("Processing failed")
return 'ok'
router = AutoDeadLetterRouter(max_retries=2)
result1 = router.process({'id': 'job-1', 'type': 'ok'}, failing_handler)
result2 = router.process({'id': 'job-2', 'type': 'fail'}, failing_handler)
result3 = router.process({'id': 'job-2', 'type': 'fail'}, failing_handler)
result4 = router.process({'id': 'job-2', 'type': 'fail'}, failing_handler)
print(f"DLQ count: {router.get_dlq_count()}")
Expected output:
Completed: job-1
Retry 1/2: job-2
Retry 2/2: job-2
Dead letter: job-2 after 2 retries
DLQ count: 1
DLQ with Classification
import time
import json
class ClassifiedDLQ:
def __init__(self):
self.queues = {
'transient': [],
'permanent': [],
}
def route_failure(self, job, error, retry_count):
error_name = type(error).__name__
entry = {
'job': job,
'error': str(error),
'error_type': error_name,
'retry_count': retry_count,
'failed_at': time.time(),
}
if retry_count >= 3:
self.queues['permanent'].append(entry)
print(f"Permanent DLQ: {error}")
elif error_name in ('ValueError', 'TypeError', 'KeyError'):
self.queues['permanent'].append(entry)
print(f"Permanent DLQ (invalid data): {error}")
else:
self.queues['transient'].append(entry)
print(f"Transient DLQ: {error}")
return entry
def get_permanent(self):
return list(self.queues['permanent'])
def get_transient(self):
return list(self.queues['transient'])
def clear_transient(self):
count = len(self.queues['transient'])
self.queues['transient'] = []
return count
def clear_permanent(self):
count = len(self.queues['permanent'])
self.queues['permanent'] = []
return count
def total(self):
return len(self.queues['transient']) + len(self.queues['permanent'])
dlq = ClassifiedDLQ()
dlq.route_failure({'task': 'payment'}, TimeoutError("timed out"), 3)
dlq.route_failure({'task': 'email'}, ValueError("bad email"), 1)
print(f"Total: {dlq.total()}")
print(f"Permanent: {len(dlq.get_permanent())}")
print(f"Transient: {len(dlq.get_transient())}")
Expected output:
Permanent DLQ: timed out
Permanent DLQ (invalid data): bad email
Total: 2
Permanent: 2
Transient: 0
DLQ Monitoring
import time
import json
class DLQMonitor:
def __init__(self, alert_threshold=10):
self.alert_threshold = alert_threshold
self.history = []
self.alerts = []
def record_check(self, dlq_count):
entry = {
'timestamp': time.time(),
'count': dlq_count,
'exceeded': dlq_count > self.alert_threshold,
}
self.history.append(entry)
if entry['exceeded']:
alert = {
'timestamp': time.time(),
'count': dlq_count,
'threshold': self.alert_threshold,
'message': f"DLQ count {dlq_count} exceeds threshold {self.alert_threshold}",
}
self.alerts.append(alert)
print(f"ALERT: {alert['message']}")
def get_recent(self, minutes=60):
cutoff = time.time() - (minutes * 60)
return [h for h in self.history if h['timestamp'] > cutoff]
def get_alert_rate(self, minutes=60):
recent = self.get_recent(minutes)
if not recent:
return 0
alerts = sum(1 for h in recent if h['exceeded'])
return (alerts / len(recent)) * 100
def summary(self):
total_checks = len(self.history)
total_alerts = len(self.alerts)
return {
'total_checks': total_checks,
'total_alerts': total_alerts,
'alert_rate_pct': (total_alerts / total_checks * 100) if total_checks > 0 else 0,
'current_threshold': self.alert_threshold,
}
monitor = DLQMonitor(alert_threshold=5)
monitor.record_check(3)
monitor.record_check(8)
monitor.record_check(12)
print(json.dumps(monitor.summary(), indent=2))
Expected output:
ALERT: DLQ count 12 exceeds threshold 5
{
"total_checks": 3,
"total_alerts": 1,
"alert_rate_pct": 33.33,
"current_threshold": 5
}
Common Mistakes
1. No Dead Letter Queue at All
Failed jobs remain in the main queue, blocking other jobs. Always route permanently failed jobs to a separate queue.
2. Infinite DLQ Growth
DLQ without limits grows forever, consuming memory. Set TTL on DLQ entries or implement a cleanup policy.
3. Ignoring DLQ
A silent DLQ accumulates failures without anyone noticing. Monitor DLQ depth and alert on unusual growth.
4. Replaying Without Inspection
Replaying DLQ jobs without understanding why they failed causes repeated failures. Inspect and fix before replay.
5. Mixing Transient and Permanent Failures
Transient failures might succeed on replay. Permanent failures will never succeed. Separate them for different handling.
Practice Questions
1. What is the purpose of a dead letter queue?
Isolate jobs that failed permanently, preserve them for inspection, and allow replay after fixing the root cause.
2. When should a job be sent to DLQ?
After exhausting all retry attempts (typically 3-5) or when the failure is permanent (invalid data, validation error).
3. How do you replay DLQ jobs?
Remove the job from DLQ, fix the underlying issue (data, code, or configuration), and re-enqueue to the original queue.
4. Why separate transient and permanent DLQs?
Transient failures might succeed on replay. Permanent failures need code or data fixes. Separate handling per type.
Challenge
Build a dead letter queue system with: auto-routing after max retries, separation by failure type, inspection API, replay with optional target queue, TTL-based cleanup, and alerting on DLQ thresholds.
FAQ
Mini Project: DLQ System
import time
import json
class DLQSystem:
def __init__(self):
self.dlq = []
self.max_retries = 3
self.stats = {'added': 0, 'replayed': 0, 'discarded': 0}
def add(self, job, error, retry_count):
entry = {
'id': f"dlq-{time.time_ns()}",
'job': job,
'error': str(error),
'retry_count': retry_count,
'failed_at': time.time(),
}
self.dlq.append(entry)
self.stats['added'] += 1
return entry['id']
def replay(self, dlq_id, handler):
for i, entry in enumerate(self.dlq):
if entry['id'] == dlq_id:
try:
result = handler(entry['job'])
self.dlq.pop(i)
self.stats['replayed'] += 1
return {'status': 'success', 'result': result}
except Exception as e:
return {'status': 'failed', 'error': str(e)}
return {'status': 'not_found'}
def discard(self, dlq_id):
for i, entry in enumerate(self.dlq):
if entry['id'] == dlq_id:
self.dlq.pop(i)
self.stats['discarded'] += 1
return True
return False
def count(self):
return len(self.dlq)
def get_stats(self):
return dict(self.stats)
dlq_sys = DLQSystem()
dlq_sys.add({'task': 'pay', 'amount': -5}, "Negative amount", 3)
dlq_sys.add({'task': 'email', 'to': ''}, "Empty recipient", 3)
dlq_sys.discard(dlq_sys.dlq[0]['id'])
print(f"DLQ count: {dlq_sys.count()}")
print(f"Stats: {dlq_sys.get_stats()}")
Expected output:
DLQ count: 1
Stats: {'added': 2, 'replayed': 0, 'discarded': 1}
What's Next
Now that you understand dead letter queues, explore retry with backoff and jitter for optimal retry strategies, then learn about idempotency keys for safe retries.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro