Dead Letter Queue: Handle Failed Jobs. Retry and Error Handling

Problem

In distributed systems and message-driven architectures, jobs fail. Network timeouts, database errors, third-party API failures, and resource constraints cause legitimate work to be lost. Without proper handling:

• Data Loss: Failed messages disappear silently
• Silent Failures: No visibility into what went wrong
• Cascading Issues: Unhandled errors propagate downstream
• Resource Waste: Retries consume resources without strategy
• Operational Blindness: Teams don't know the system is broken

Dead Letter Queues (DLQs) solve this by capturing failed jobs for analysis, retry, and recovery.

---

Solution

A Dead Letter Queue is a secondary queue that captures messages that fail processing after exhausting retry attempts. The strategy involves:

1. Attempt Processing: Try to process the job normally
2. Catch Failures: Detect errors during execution
3. Retry with Backoff: Implement exponential backoff for transient failures
4. Move to DLQ: After max retries, move to Dead Letter Queue
5. Monitor & Alert: Track DLQ depth and alert operators
6. Manual Intervention: Operators investigate and reprocess or discard

This separates transient failures (retry) from permanent failures (investigate).

---

Code

1. Core Job Processor with Retry Logic

import json
import time
import logging
from typing import Callable, Any, Dict
from enum import Enum
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class JobStatus(Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SUCCESS = "success"
    FAILED = "failed"
    DLQ = "dlq"

@dataclass
class Job:
    id: str
    type: str
    payload: Dict[str, Any]
    retry_count: int = 0
    max_retries: int = 3
    status: str = JobStatus.PENDING.value
    error_message: str = None
    created_at: str = None
    last_attempt_at: str = None

    def __post_init__(self):
        if not self.created_at:
            self.created_at = datetime.utcnow().isoformat()

    def to_dict(self):
        return asdict(self)

class RetryStrategy:
    """Exponential backoff retry strategy"""

    @staticmethod
    def calculate_backoff(retry_count: int, base_delay: int = 1) -> int:
        """Calculate exponential backoff: 1s, 2s, 4s, 8s..."""
        return base_delay * (2 ** retry_count)

    @staticmethod
    def should_retry(job: Job) -> bool:
        """Determine if job should be retried"""
        return job.retry_count < job.max_retries

class JobProcessor:
    """Process jobs with retry and DLQ handling"""

    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.dlq = []  # Dead Letter Queue
        self.processed = []  # Successfully processed

    def process(self, job: Job, handler: Callable) -> bool:
        """
        Process a job with retry logic

        Args:
            job: Job to process
            handler: Function that processes the job

        Returns:
            True if successful, False if moved to DLQ
        """
        job.status = JobStatus.PROCESSING.value
        job.last_attempt_at = datetime.utcnow().isoformat()

        try:
            logger.info(f"Processing job {job.id} (attempt {job.retry_count + 1})")
            result = handler(job.payload)

            job.status = JobStatus.SUCCESS.value
            self.processed.append(job)
            logger.info(f"Job {job.id} succeeded")
            return True

        except Exception as e:
            job.error_message = str(e)
            logger.warning(f"Job {job.id} failed: {e}")

            if RetryStrategy.should_retry(job):
                return self._retry_job(job, handler)
            else:
                return self._move_to_dlq(job)

    def _retry_job(self, job: Job, handler: Callable) -> bool:
        """Retry job with exponential backoff"""
        job.retry_count += 1
        backoff = RetryStrategy.calculate_backoff(job.retry_count - 1)

        logger.info(f"Retrying job {job.id} in {backoff}s "
                   f"(attempt {job.retry_count}/{job.max_retries})")
        time.sleep(backoff)

        return self.process(job, handler)

    def _move_to_dlq(self, job: Job) -> bool:
        """Move job to Dead Letter Queue"""
        job.status = JobStatus.DLQ.value
        self.dlq.append(job)
        logger.error(f"Job {job.id} moved to DLQ after {job.retry_count} retries")
        return False

    def get_dlq_stats(self) -> Dict[str, Any]:
        """Get Dead Letter Queue statistics"""
        return {
            "dlq_size": len(self.dlq),
            "processed_count": len(self.processed),
            "dlq_jobs": [job.to_dict() for job in self.dlq]
        }

2. Message Queue Implementation with DLQ

from queue import Queue
from threading import Thread, Lock
import uuid

class MessageQueue:
    """Message queue with Dead Letter Queue support"""

    def __init__(self, max_retries: int = 3):
        self.main_queue = Queue()
        self.dlq = Queue()
        self.processor = JobProcessor(max_retries=max_retries)
        self.lock = Lock()
        self.handlers = {}

    def register_handler(self, job_type: str, handler: Callable):
        """Register handler for job type"""
        self.handlers[job_type] = handler

    def enqueue(self, job_type: str, payload: Dict[str, Any]) -> str:
        """Add job to queue"""
        job_id = str(uuid.uuid4())
        job = Job(
            id=job_id,
            type=job_type,
            payload=payload,
            max_retries=3
        )
        self.main_queue.put(job)
        logger.info(f"Job {job_id} enqueued")
        return job_id

    def process_queue(self, num_workers: int = 2):
        """Start worker threads to process queue"""
        threads = []
        for i in range(num_workers):
            t = Thread(target=self._worker, daemon=True)
            t.start()
            threads.append(t)

        logger.info(f"Started {num_workers} worker threads")
        return threads

    def _worker(self):
        """Worker thread that processes jobs"""
        while True:
            try:
                job = self.main_queue.get(timeout=1)

                if job.type not in self.handlers:
                    logger.error(f"No handler for job type: {job.type}")
                    self.dlq.put(job)
                    continue

                handler = self.handlers[job.type]
                self.processor.process(job, handler)

                self.main_queue.task_done()

            except Exception as e:
                logger.error(f"Worker error: {e}")

    def get_dlq_items(self) -> list:
        """Retrieve all DLQ items"""
        items = []
        while not self.dlq.empty():
            items.append(self.dlq.get())
        return items

    def requeue_dlq_item(self, job: Job):
        """Requeue a DLQ item back to main queue"""
        job.retry_count = 0
        job.status = JobStatus.PENDING.value
        self.main_queue.put(job)
        logger.info(f"Job {job.id} requeued from DLQ")

3. Practical Example: Email Service

class EmailService:
    """Example service that sends emails with DLQ handling"""

    def __init__(self):
        self.queue = MessageQueue(max_retries=3)
        self.queue.register_handler("send_email", self.send_email_handler)

    def send_email_handler(self, payload: Dict[str, Any]) -> bool:
        """Handler that processes email jobs"""
        email = payload.get("email")
        subject = payload.get("subject")
        body = payload.get("body")

        # Simulate email sending with potential failures
        import random
        if random.random() < 0.3:  # 30% failure rate
            raise Exception("SMTP connection timeout")

        logger.info(f"Email sent to {email}: {subject}")
        return True

    def send_email(self, email: str, subject: str, body: str) -> str:
        """Queue email for sending"""
        return self.queue.enqueue("send_email", {
            "email": email,
            "subject": subject,
            "body": body
        })

    def start(self, num_workers: int = 2):
        """Start processing emails"""
        self.queue.process_queue(num_workers)

    def get_failed_emails(self) -> list:
        """Get emails that failed permanently"""
        return self.queue.processor.dlq

# Usage
if __name__ == "__main__":
    service = EmailService()
    service.start(num_workers=2)

    # Queue some emails
    for i in range(10):
        service.send_email(
            f"user{i}@example.com",
            "Welcome",
            "Welcome to our service!"
        )

    time.sleep(15)

    # Check DLQ
    stats = service.queue.processor.get_dlq_stats()
    print(json.dumps(stats, indent=2))

4. Monitoring & Alerting

from dataclasses import dataclass
from typing import List

@dataclass
class DLQMetrics:
    dlq_size: int
    dlq_growth_rate: float
    oldest_job_age_seconds: int
    error_categories: Dict[str, int]

class DLQMonitor:
    """Monitor Dead Letter Queue health"""

    def __init__(self, alert_threshold: int = 10):
        self.alert_threshold = alert_threshold
        self.dlq_history = []

    def check_health(self, processor: JobProcessor) -> DLQMetrics:
        """Check DLQ health and generate metrics"""
        dlq_jobs = processor.dlq
        dlq_size = len(dlq_jobs)

        # Calculate growth rate
        self.dlq_history.append(dlq_size)
        if len(self.dlq_history) > 1:
            growth_rate = (self.dlq_history[-1] - self.dlq_history[-2]) / max(1, self.dlq_history[-2])
        else:
            growth_rate = 0

        # Find oldest job
        oldest_age = 0
        if dlq_jobs:
            oldest_time = min(job.created_at for job in dlq_jobs)
            oldest_age = int((datetime.utcnow() - datetime.fromisoformat(oldest_time)).total_seconds())

        # Categorize errors
        error_categories = {}
        for job in dlq_jobs:
            error_type = job.error_message.split(":")[0] if job.error_message else "Unknown"
            error_categories[error_type] = error_categories.get(error_type, 0) + 1

        metrics = DLQMetrics(
            dlq_size=dlq_size,
            dlq_growth_rate=growth_rate,
            oldest_job_age_seconds=oldest_age,
            error_categories=error_categories
        )

        # Alert if threshold exceeded
        if dlq_size > self.alert_threshold:
            logger.critical(f"ALERT: DLQ size {dlq_size} exceeds threshold {self.alert_threshold}")

        return metrics

---

Tips

1. Choose Appropriate Retry Counts

• Transient failures (network): 3-5 retries
• Database operations: 2-3 retries
• External APIs: 3-5 retries with longer backoff
• Permanent failures (validation): 0 retries

2. Implement Exponential Backoff

Retry 1: 1 second
Retry 2: 2 seconds
Retry 3: 4 seconds
Retry 4: 8 seconds

Prevents overwhelming failing services.

3. Add Jitter to Backoff

backoff = base_delay * (2 ** retry_count)
jitter = random.uniform(0, backoff * 0.1)
sleep_time = backoff + jitter

Prevents thundering herd when multiple jobs retry simultaneously.

4. Classify Errors

class ErrorType(Enum):
    TRANSIENT = "transient"      # Retry
    PERMANENT = "permanent"      # DLQ immediately
    UNKNOWN = "unknown"          # Retry then DLQ

5. Monitor DLQ Depth

• Alert when DLQ grows beyond threshold
• Track error patterns
• Measure time-to-resolution

6. Implement Manual Reprocessing

• Provide UI/API to inspect DLQ items
• Allow selective reprocessing after fixes
• Track reprocessing success rate

7. Set TTL on DLQ Items

def cleanup_old_dlq_items(processor, max_age_days=30):
    cutoff = datetime.utcnow() - timedelta(days=max_age_days)
    processor.dlq = [
        job for job in processor.dlq 
        if datetime.fromisoformat(job.created_at) > cutoff
    ]

8. Use Structured Logging

logger.error("Job failed", extra={
    "job_id": job.id,
    "retry_count": job.retry_count,
    "error": str(e),
    "timestamp": datetime.utcnow().isoformat()
})

9. Implement Circuit Breaker Pattern

Stop retrying if downstream service is down:

if error_rate > 0.5:  # 50% failure rate
    raise CircuitBreakerOpen("Service unavailable")

10. Test DLQ Scenarios

• Simulate transient failures (network timeout)
• Simulate permanent failures (invalid data)
• Verify retry backoff timing
• Test DLQ capacity limits
• Validate reprocessing logic

---

Summary

Dead Letter Queues transform error handling from silent failures to observable, manageable processes. By combining retry strategies, exponential backoff, and DLQ capture, systems become resilient and debuggable. The key is balancing automatic recovery (retries) with manual intervention (DLQ inspection), ensuring no work is lost and operators have visibility into system health.