# Dlq Retry Patterns > Implements dead letter queue architectures with retry strategies (exponential backoff, circuit breaker integration, poison pill detection) for resilient event-driven systems. - Skill: `paulpas/dlq-retry-patterns` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/dlq-retry-patterns` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/dlq-retry-patterns/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Integrations & APIs - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/dlq-retry-patterns --- # Dead Letter Queue and Retry Patterns Implements dead letter queue architectures with retry strategies to ensure resilient event-driven systems. When loaded, this skill makes the model design DLQ routing, configure exponential backoff with jitter, detect and quarantine poison pills, integrate circuit breakers into event handlers, and build replay workflows for recovered messages. ## TL;DR Checklist - [ ] Configure a dedicated dead letter queue for every processing queue — never let failures silently disappear - [ ] Set maximum retry counts with exponential backoff and full jitter before routing to DLQ - [ ] Implement poison pill detection: reject messages that fail consistently (>3 retries on same error type) - [ ] Add idempotency keys to all retry logic so reprocessing never causes duplicate side effects - [ ] Instrument DLQ depth metrics with alerting thresholds for proactive incident response - [ ] Build a replay workflow that moves fixed messages back to original queues after root cause is resolved - [ ] Circuit-break downstream dependencies before allowing retries to hammer an already-failing service --- ## When to Use Use this skill when: - Designing or implementing dead letter queue infrastructure for a message broker (RabbitMQ, Kafka, SQS, NATS) - Building retry logic for event handlers that need exponential backoff with jitter - Debugging poison pill scenarios where one bad message blocks an entire processing pipeline - Integrating circuit breaker patterns into event consumers to prevent cascade failures - Setting up monitoring and alerting on DLQ depth and retry failure rates - Designing replay workflows for reprocessing messages after a known transient issue is resolved - Implementing idempotent retry strategies to safely reprocess events without duplicate side effects --- ## When NOT to Use Avoid this skill for: - **Simple synchronous error handling** — use standard try/except with local retries instead of DLQ machinery - **State machine design** — use `event-sourcing-pattern` for state transitions that need event replay at the domain level - **General message queue topology** — use `event-driven-patterns` for pub/sub, saga coordination, and outbox patterns - **One-off script failures** — logging and manual intervention are sufficient; don't over-engineer DLQ infrastructure for simple tasks --- ## Core Workflow 1. **Map the Failure Surface** — Identify every event handler that can fail, classify failures as transient (network timeout, downstream unavailable) or permanent (schema violation, invalid payload). **Checkpoint:** Every handler has a documented failure classification — transient vs. permanent determines retry policy and DLQ routing logic. 2. **Configure Queue Topology** — Create source queues with `x-dead-letter-exchange` and `x-dead-letter-routing-key` parameters so failed messages automatically route to the DLQ after exhausting retries. Use separate DLQs per handler for diagnostic isolation, or a single shared DLQ for simpler systems. **Checkpoint:** Every queue has explicit retry counts, TTL, and DLQ routing configured — never rely on broker defaults. 3. **Implement Retry with Backoff** — Add exponential backoff with full jitter to each event handler. Configure max retries (typically 3–5), then route to the DLQ. Integrate circuit breaker checks before each retry attempt. **Checkpoint:** The retry loop records every attempt with timestamp, error type, and delay — this data powers poison pill detection and operational dashboards. 4. **Detect Poison Pills** — Implement a poison pill detector that tracks failure patterns per message ID. If the same error repeats beyond a threshold (e.g., 3 failures with identical error types on the same message), quarantine the message immediately instead of continuing futile retries. **Checkpoint:** Quarantined messages are isolated in a separate DLQ sub-queue named after the failure category (schema_error, downstream_timeout, etc.). 5. **Instrument and Alert** — Add metrics collection for DLQ depth per queue, retry count distribution, error type frequency, and average message age in DLQ. Configure alerts when DLQ depth exceeds thresholds (e.g., >100 messages) or when messages age beyond SLA bounds (e.g., >30 minutes). **Checkpoint:** Every metric has a corresponding Grafana dashboard panel and PagerDuty/Slack alerting rule — silent DLQ accumulation is an incident, not a feature. 6. **Build Replay Workflow** — Implement a safe replay mechanism that reads from the DLQ, validates messages for recoverability, applies idempotency keys, and publishes back to the original source queue. Include manual approval gates for high-risk replays. **Checkpoint:** The replay workflow processes messages in bounded batches (max 50 at a time) with per-message retry limits to prevent overwhelming the system on re-entry. --- ## Implementation Patterns ### Pattern 1: Dead Letter Queue Manager with Automatic Routing A production-grade DLQ manager that wraps any message broker, providing automatic routing of failed messages to dead letter queues with metadata preservation and configurable policies. Uses separate DLQs per source queue for diagnostic isolation. ```python from __future__ import annotations import json import logging import time import uuid from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from typing import Any, Callable, Protocol logger = logging.getLogger(__name__) class FailureCategory(Enum): """Classifies why a message failed, enabling targeted DLQ routing.""" TRANSIENT = "transient" # Network timeout, downstream unavailable SCHEMA = "schema" # Payload doesn't match expected schema BUSINESS = "business" # Business rule violation (e.g., negative quantity) INTEGRATION = "integration" # External service call failed permanently @dataclass(frozen=True) class DeliveryMetadata: """Immutable metadata attached to every delivery attempt.""" message_id: str = field(default_factory=lambda: str(uuid.uuid4())) source_queue: str = "" correlation_id: str = field(default_factory=lambda: str(uuid.uuid4())) delivered_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) attempt_number: int = 0 max_retries: int = 3 @dataclass class DeadLetterRecord: """Persistent record of a message that moved to the dead letter queue.""" original_message_id: str source_queue: str payload: dict[str, Any] headers: dict[str, Any] failure_category: FailureCategory error_reason: str attempt_number: int max_retries: int first_failed_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) last_attempted_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) retry_history: list[dict[str, Any]] = field(default_factory=list) @property def age_seconds(self) -> float: """How long this message has been in the DLQ.""" return (datetime.now(timezone.utc) - self.first_failed_at).total_seconds() class MessageBroker(Protocol): """Minimal protocol for broker interactions — any real broker adapter implements this.""" def publish(self, queue: str, payload: dict[str, Any], headers: dict[str, Any] | None = None) -> str: ... def consume(self, queue: str, timeout_ms: int = 5000) -> dict[str, Any] | None: ... def acknowledge(self, delivery_tag: str) -> None: ... def reject(self, delivery_tag: str, requeue: bool = False) -> None: ... class DeadLetterQueueManager: """Manages dead letter queue operations with automatic routing and metadata preservation. When a message fails processing beyond its retry limit, this manager: 1. Classifies the failure type (transient vs permanent) 2. Routes to the appropriate DLQ (per-queue isolation by default) 3. Preserves full message context for forensic analysis 4. Records metrics for monitoring and alerting Separate DLQs per source queue enable targeted investigation — a DLQ depth spike on `order.processing.dlq` immediately signals an issue with order processing, not shipping or payment handlers. """ def __init__( self, broker: MessageBroker, dlq_prefix: str = ".dlq", max_retries: int = 3, ) -> None: self._broker = broker self._max_retries = max_retries self._dlq_prefix = dlq_prefix self._dlq_records: dict[str, DeadLetterRecord] = {} @property def dlq_name(self, source_queue: str) -> str: """Derive the DLQ name from a source queue using convention. Args: source_queue: The original message queue name. Returns: DLQ-qualified name (e.g., 'order.processing' → 'order.processing.dlq'). """ return f"{source_queue}{self._dlq_prefix}" def record_failure( self, source_queue: str, payload: dict[str, Any], headers: dict[str, Any], error: Exception, attempt_number: int, failure_category: FailureCategory = FailureCategory.TRANSIENT, ) -> DeadLetterRecord | None: """Record a message failure and route to DLQ if retries exhausted. Returns the DeadLetterRecord if routed to DLQ, or None if still retriable. Args: source_queue: The queue the message came from. payload: The original message payload. headers: Message metadata headers. error: The exception that caused processing failure. attempt_number: Current retry attempt number (1-based). failure_category: Classification of the failure type. Returns: DeadLetterRecord if DLQ routing occurred, None if still within retry budget. """ if attempt_number < self._max_retries: logger.info( "Message %s failed on attempt %d/%d — will retry", headers.get("message_id", "unknown"), attempt_number, self._max_retries, ) return None # Retry budget exhausted — route to DLQ source_queue_name = source_queue dlq_name = self.dlq_name(source_queue_name) message_id = headers.get("message_id", str(uuid.uuid4())) record = DeadLetterRecord( original_message_id=message_id, source_queue=source_queue_name, payload=payload, headers=headers, failure_category=failure_category, error_reason=str(error), attempt_number=attempt_number, max_retries=self._max_retries, ) # Publish to DLQ with extended metadata dlq_headers = {**headers, "dlq_routed_at": datetime.now(timezone.utc).isoformat(), "failure_category": failure_category.value} self._broker.publish(dlq_name, {"_dlq_record": record.__dict__, "_original_payload": payload}, dlq_headers) self._dlq_records[message_id] = record logger.warning( "Message %s routed to DLQ (%s) after %d attempts: %s", message_id, dlq_name, attempt_number, error, ) return record def get_dlq_records(self, source_queue: str | None = None) -> list[DeadLetterRecord]: """Retrieve DLQ records, optionally filtered by source queue. Args: source_queue: If provided, filter to only records from this queue. Returns: List of DeadLetterRecord instances matching the filter. """ if source_queue is None: return list(self._dlq_records.values()) return [ r for r in self._dlq_records.values() if r.source_queue == source_queue ] class DirectMessageBrokerAdapter: """Concrete broker adapter implementing the MessageBroker protocol. This is a simplified in-memory adapter for demonstration. In production, replace with RabbitMQChannelPublisher, KafkaProducerAdapter, or SqsClientAdapter. """ def __init__(self) -> None: self._queues: dict[str, list[dict[str, Any]]] = {} def publish(self, queue: str, payload: dict[str, Any], headers: dict[str, Any] | None = None) -> str: message_id = str(uuid.uuid4()) message = { "message_id": message_id, "payload": payload, "headers": headers or {}, "published_at": time.monotonic(), } self._queues.setdefault(queue, []).append(message) return message_id def consume(self, queue: str, timeout_ms: int = 5000) -> dict[str, Any] | None: messages = self._queues.get(queue, []) if not messages: return None return messages.pop(0) def acknowledge(self, delivery_tag: str) -> None: logger.debug("Acknowledged delivery %s", delivery_tag) def reject(self, delivery_tag: str, requeue: bool = False) -> None: logger.debug("Rejected delivery %s, requeue=%s", delivery_tag, requeue) ``` --- ### Pattern 2: Exponential Backoff with Full Jitter Retry Decorator A retry mechanism that uses exponential backoff with full random jitter to prevent thundering herd problems when multiple consumers simultaneously retry failed messages. Includes circuit breaker awareness to avoid hammering already-failing downstream services. ```python from __future__ import annotations import asyncio import logging import random import time from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from typing import TypeVar logger = logging.getLogger(__name__) T = TypeVar("T") @dataclass(frozen=True) class RetryPolicy: """Immutable retry policy configuration.""" max_attempts: int = 5 base_delay_ms: float = 100.0 # Base delay for exponential backoff max_delay_ms: float = 30_000.0 # Cap on individual retry delays backoff_multiplier: float = 2.0 # Exponent base jitter_mode: str = "full" # full | random | none retryable_exceptions: tuple[type[Exception], ...] = (ConnectionError, TimeoutError) @property def should_retry(self) -> Callable[[Exception], bool]: """Return a predicate that determines if an exception is worth retrying.""" return lambda exc: isinstance(exc, self.retryable_exceptions) @dataclass class RetryResult: """Outcome of a retry operation.""" success: bool attempts: int = 0 total_duration_ms: float = 0.0 final_error: str | None = None circuit_breaker_opened: bool = False def compute_delay( attempt: int, policy: RetryPolicy, ) -> float: """Calculate delay for a given retry attempt with jitter. Uses full jitter (AWS recommended pattern): random value between 0 and the calculated exponential backoff delay. This spreads retry attempts evenly across time, preventing thundering herd when many clients retry simultaneously. Args: attempt: The 1-based attempt number. policy: Retry policy configuration. Returns: Delay in seconds to wait before the next retry. """ # Exponential backoff: base_delay * (multiplier ^ (attempt - 1)) exponential_delay_ms = min( policy.base_delay_ms * (policy.backoff_multiplier ** (attempt - 1)), policy.max_delay_ms, ) if policy.jitter_mode == "full": # Full jitter: random value in [0, exponential_delay] delay_ms = random.uniform(0, exponential_delay_ms) elif policy.jitter_mode == "random": # Random jitter: exponential_delay * random(0.5, 1.5) delay_ms = exponential_delay_ms * random.uniform(0.5, 1.5) else: # No jitter: exact exponential delay_ms = exponential_delay_ms return delay_ms / 1000.0 class CircuitBreaker: """Simple circuit breaker to prevent retrying when a downstream is clearly failing. States: CLOSED → Normal operation, retries proceed OPEN → Downstream failing, all retries rejected immediately HALF_OPEN → Testing if downstream recovered (one probe attempt allowed) Args: failure_threshold: Number of consecutive failures before opening circuit. recovery_timeout_seconds: Seconds to wait before transitioning to half-open. """ def __init__( self, failure_threshold: int = 5, recovery_timeout_seconds: float = 30.0, ) -> None: self._failure_threshold = failure_threshold self._recovery_timeout = recovery_timeout_seconds self._failure_count: int = 0 self._last_failure_time: float = 0.0 self._state: str = "closed" @property def state(self) -> str: return self._state def allow_request(self) -> bool: """Check if a retry attempt is allowed through the circuit breaker.""" if self._state == "closed": return True if self._state == "open": # Transition to half-open after recovery timeout if time.monotonic() - self._last_failure_time > self._recovery_timeout: self._state = "half-open" logger.info("Circuit breaker transitioning to HALF_OPEN") return True # Allow one probe attempt return False # half-open — allow exactly one probe return True def record_success(self) -> None: """Record a successful call — resets circuit breaker state.""" self._failure_count = 0 self._state = "closed" def record_failure(self) -> None: """Record a failed call — increments failure count and may open circuit.""" self._failure_count += 1 self._last_failure_time = time.monotonic() if self._failure_count >= self._failure_threshold: self._state = "open" logger.warning( "Circuit breaker OPENED after %d consecutive failures", self._failure_count, ) def retry_with_backoff( func: Callable[..., T], policy: RetryPolicy | None = None, circuit_breaker: CircuitBreaker | None = None, ) -> Callable[..., RetryResult]: """Decorator that wraps a function with exponential backoff and jitter retry. Usage: @retry_with_backoff(RetryPolicy(max_attempts=3)) def send_to_downstream(message): ... Args: func: The function to wrap with retry logic. policy: Retry policy configuration (default: 5 attempts, full jitter). circuit_breaker: Optional circuit breaker for downstream protection. Returns: A wrapped function that returns RetryResult with outcome details. """ if policy is None: policy = RetryPolicy() def wrapper(*args: Any, **kwargs: Any) -> RetryResult: cb_opened = False last_error: Exception | None = None for attempt in range(1, policy.max_attempts + 1): # Check circuit breaker before attempting retry if circuit_breaker and not circuit_breaker.allow_request(): logger.warning("Circuit breaker OPEN — skipping retry attempt %d", attempt) cb_opened = True return RetryResult(success=False, attempts=attempt, circuit_breaker_opened=True) try: result = func(*args, **kwargs) if circuit_breaker: circuit_breaker.record_success() duration_ms = 0.0 # Would be tracked with a timer wrapper return RetryResult(success=True, attempts=attempt, total_duration_ms=duration_ms) except Exception as exc: last_error = exc if not policy.should_retry(exc): logger.info("Non-retryable exception %s — giving up", type(exc).__name__) break if attempt < policy.max_attempts: delay = compute_delay(attempt, policy) logger.info( "Attempt %d/%d failed (%s), retrying in %.1fms", attempt, policy.max_attempts, exc, delay * 1000, ) time.sleep(delay) if circuit_breaker: circuit_breaker.record_failure() return RetryResult( success=False, attempts=policy.max_attempts, total_duration_ms=0.0, final_error=str(last_error), circuit_breaker_opened=cb_opened, ) return wrapper ``` --- ### Pattern 3: Poison Pill Detection and Quarantine Detects poison pills by analyzing failure patterns across retry attempts. Messages that fail repeatedly with the same error type are classified as poison pills and quarantined to a separate sub-queue, preventing them from blocking normal processing retries. ```python from __future__ import annotations import logging import time from collections import defaultdict from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any logger = logging.getLogger(__name__) @dataclass(frozen=True) class FailureSignature: """A hashable fingerprint of a failure for pattern detection. Groups failures by error type and message content so that repeated failures from the same cause are detected quickly. """ error_type: str queue_name: str handler_name: str # Optional: hash a snippet of payload to detect malformed content payload_hash: str = "" @classmethod def from_message( cls, message: dict[str, Any], error: Exception, handler_name: str, ) -> FailureSignature: import hashlib return cls( error_type=type(error).__name__, queue_name=message.get("source_queue", "unknown"), handler_name=handler_name, payload_hash=hashlib.md5( str(message.get("payload", "")).encode()[:100] ).hexdigest(), ) @dataclass class PoisonPillAlert: """Alert raised when a poison pill is detected and quarantined.""" message_id: str source_queue: str handler_name: str failure_signature: FailureSignature consecutive_failures: int quarantine_time: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) reason: str = "" class PoisonPillDetector: """Detects and quarantines poison pills — messages that consistently fail processing. A poison pill is a message whose content or state makes it impossible to process, regardless of retries. Examples: corrupted JSON payload, missing required field, permanently invalid business rule violation. Detection strategy: 1. Track consecutive failures per message (grouped by failure signature). 2. If the same error type repeats ≥ threshold times, classify as poison pill. 3. Quarantine to a named quarantine sub-queue (e.g., `order.processing.dlq.quarantine.schema`). 4. Raise an alert for on-call investigation. The key insight: continuing to retry the same error is waste and blocks queue progress. A poison pill needs human intervention or payload correction, not more retries. """ def __init__( self, failure_threshold: int = 3, time_window_seconds: float = 60.0, ) -> None: self._failure_threshold = failure_threshold self._time_window = time_window_seconds # Tracks consecutive failures per message ID grouped by error type self._message_failures: dict[str, list[dict[str, Any]]] = defaultdict(list) # Tracks handler-level failure rates for bulk poison detection self._handler_failures: dict[str, int] = defaultdict(int) def record_failure( self, message_id: str, message: dict[str, Any], error: Exception, handler_name: str, ) -> PoisonPillAlert | None: """Record a failure attempt and check if the message is now a poison pill. Args: message_id: Unique ID of the failing message. message: The original message payload and headers. error: The exception raised during processing. handler_name: Name of the handler that failed. Returns: PoisonPillAlert if quarantined, None if still within normal retry bounds. """ signature = FailureSignature.from_message(message, error, handler_name) now = time.monotonic() # Record this failure attempt self._message_failures[message_id].append({ "timestamp": now, "error_type": signature.error_type, "signature": signature, "handler": handler_name, "queue": message.get("source_queue", "unknown"), }) # Clean old entries outside the time window self._message_failures[message_id] = [ entry for entry in self._message_failures[message_id] if now - entry["timestamp"] < self._time_window ] recent_entries = self._message_failures[message_id] # Check for poison pill: same error type repeated ≥ threshold times error_counts: dict[str, int] = defaultdict(int) for entry in recent_entries: error_counts[entry["error_type"]] += 1 for error_type, count in error_counts.items(): if count >= self._failure_threshold: alert = PoisonPillAlert( message_id=message_id, source_queue=message.get("source_queue", "unknown"), handler_name=handler_name, failure_signature=signature, consecutive_failures=count, reason=( f"Message failed {count} times with {error_type} in " f"{self._time_window:.0f}s window — quarantining" ), ) logger.error( "POISON PILL detected: message %s (queue: %s, handler: %s) — %s", message_id, message.get("source_queue"), handler_name, alert.reason, ) return alert return None def quarantine_subqueue(self, source_queue: str, signature: FailureSignature) -> str: """Derive the quarantine sub-queue name for a given failure signature. Args: source_queue: The original source queue. signature: The failure signature classifying the error type. Returns: Quarantine queue name (e.g., 'order.processing.dlq.quarantine.schema_error'). """ base_dlq = f"{source_queue}.dlq.quarantine" return f"{base_dlq}.{signature.error_type.lower().replace('error', 'err')}" def reset_message(self, message_id: str) -> None: """Reset failure tracking for a message (e.g., after manual fix and replay).""" self._message_failures.pop(message_id, None) logger.info("Reset failure tracking for message %s", message_id) ``` --- ### Pattern 4: Circuit Breaker for Event Handlers A circuit breaker integrated into event handler execution that prevents cascading failures when downstream dependencies are unavailable. Opens after consecutive failures and transitions to half-open to probe recovery. ```python from __future__ import annotations import logging import threading import time from collections import defaultdict from dataclasses import dataclass, field from enum import Enum from typing import Any, Generic, Protocol, TypeVar logger = logging.getLogger(__name__) class CircuitState(Enum): CLOSED = "closed" # Normal — allow requests through OPEN = "open" # Failing — reject all requests immediately HALF_OPEN = "half-open" # Testing — allow limited probe requests T = TypeVar("T") @dataclass class CircuitBreakerMetrics: """Live metrics for a circuit breaker instance.""" total_calls: int = 0 successful_calls: int = 0 failed_calls: int = 0 rejected_calls: int = 0 state_transitions: int = 0 last_failure_time: float | None = None current_state: CircuitState = CircuitState.CLOSED class EventHandler(Protocol, Generic[T]): """Protocol for event handlers that can be wrapped with circuit breaker protection.""" async def handle(self, event: dict[str, Any]) -> T: ... @property def name(self) -> str: ... @dataclass class CircuitBreakerConfig: """Configuration for a circuit breaker protecting a specific downstream dependency.""" failure_threshold: int = 5 # Consecutive failures before opening success_threshold_half_open: int = 2 # Successes needed in half-open to close recovery_timeout_seconds: float = 30.0 # Time to wait before probing maximum_concurrent_probes: int = 1 # Allowed concurrent calls in half-open state class HandlerCircuitBreaker(Generic[T]): """Circuit breaker that protects individual event handlers from cascading failures. Each handler gets its own circuit breaker instance, allowing independent failure tracking per downstream dependency. For example, the `EmailNotificationHandler` might trip while `PaymentProcessingHandler` continues operating normally. State transitions: CLOSED → OPEN: After `failure_threshold` consecutive failures. OPEN → HALF_OPEN: After `recovery_timeout_seconds` elapses. HALF_OPEN → CLOSED: After `success_threshold_half_open` successful probe calls. HALF_OPEN → OPEN: Any single failure in half-open state reopens immediately. Usage: breaker = HandlerCircuitBreaker(name="email_notifications") result = await breaker.execute(lambda: send_email(event)) if result.rejected: route_to_dlq(event, "handler_circuit_open") """ def __init__(self, name: str, config: CircuitBreakerConfig | None = None) -> None: self._name = name self._config = config or CircuitBreakerConfig() self._state = CircuitState.CLOSED self._consecutive_failures = 0 self._consecutive_successes = 0 self._last_failure_time: float = 0.0 self._half_open_probes = 0 self._lock = threading.Lock() self._metrics = CircuitBreakerMetrics() @property def name(self) -> str: return self._name @property def state(self) -> CircuitState: """Current circuit breaker state, with automatic OPEN→HALF_OPEN transition.""" if self._state == CircuitState.OPEN: if (time.monotonic() - self._last_failure_time) >= self._config.recovery_timeout_seconds: self._transition_to(CircuitState.HALF_OPEN) return self._state def execute(self, handler_fn: Callable[[], T]) -> "HandlerExecutionResult[T]": """Execute a handler through the circuit breaker. Args: handler_fn: A callable that performs the event handling logic. Returns: HandlerExecutionResult with outcome details including rejection status. """ current_state = self.state if current_state == CircuitState.OPEN: self._metrics.rejected_calls += 1 return HandlerExecutionResult( success=False, rejected=True, state=self.state, error=f"Circuit breaker OPEN for handler '{self._name}'", ) self._metrics.total_calls += 1 try: result = handler_fn() self._on_success() return HandlerExecutionResult( success=True, rejected=False, state=self.state, result=result, ) except Exception as exc: self._on_failure(exc) return HandlerExecutionResult( success=False, rejected=False, state=self.state, error=str(exc), ) async def execute_async(self, handler_fn: Callable[[], Any]) -> "HandlerExecutionResult[Any]": """Async version of execute for use with async handlers.""" current_state = self.state if current_state == CircuitState.OPEN: self._metrics.rejected_calls += 1 return HandlerExecutionResult( success=False, rejected=True, state=self.state, error=f"Circuit breaker OPEN for handler '{self._name}'", ) self._metrics.total_calls += 1 try: result = await handler_fn() self._on_success() return HandlerExecutionResult(success=True, rejected=False, state=self.state, result=result) except Exception as exc: self._on_failure(exc) return HandlerExecutionResult( success=False, rejected=False, state=self.state, error=str(exc), ) def _on_success(self) -> None: """Handle a successful handler execution.""" if self._state == CircuitState.HALF_OPEN: self._consecutive_successes += 1 if self._consecutive_successes >= self._config.success_threshold_half_open: self._transition_to(CircuitState.CLOSED) else: self._consecutive_failures = 0 self._consecutive_successes += 1 self._metrics.successful_calls += 1 def _on_failure(self, exc: Exception) -> None: """Handle a failed handler execution.""" self._consecutive_failures += 1 self._consecutive_successes = 0 self._last_failure_time = time.monotonic() self._metrics.failed_calls += 1 if self._state == CircuitState.HALF_OPEN: # Any failure in half-open reopens the circuit immediately logger.warning("Circuit breaker for '%s' reopened from HALF_OPEN", self._name) self._transition_to(CircuitState.OPEN) elif self._consecutive_failures >= self._config.failure_threshold: logger.warning( "Circuit breaker for '%s' OPENED after %d consecutive failures: %s", self._name, self._consecutive_failures, exc, ) self._transition_to(CircuitState.OPEN) def _transition_to(self, new_state: CircuitState) -> None: """Transition the circuit breaker to a new state.""" old_state = self._state self._state = new_state self._metrics.state_transitions += 1 self._metrics.current_state = new_state if new_state == CircuitState.HALF_OPEN: self._consecutive_successes = 0 self._half_open_probes = 0 logger.info( "Circuit breaker '%s': %s → %s", self._name, old_state.value, new_state.value, ) @dataclass class HandlerExecutionResult(Generic[T]): """Result of executing a handler through the circuit breaker.""" success: bool rejected: bool state: CircuitState result: T | None = None error: str | None = None # ── BAD vs GOOD Comparison ───────────────────────────────────────────── # ❌ BAD — No circuit breaker: retries hammer an already-downstream service, # consuming resources and potentially causing complete system outage. async def bad_event_handler(event: dict[str, Any]) -> None: """Processes events by retrying until success — no backoff protection.""" for attempt in range(10): # Will retry forever regardless of downstream state try: await call_downstream_service(event) return except ConnectionError: time.sleep(1) # Bare 1-second delay, no exponential backoff continue # Continues hammering the failing service # ✅ GOOD — Circuit breaker prevents cascade: after 5 failures, all retries # are rejected immediately, and downstream gets recovery time. async def good_event_handler(event: dict[str, Any], cb: HandlerCircuitBreaker) -> None: """Processes events with circuit breaker protection for the downstream service.""" result = await cb.execute_async(lambda: call_downstream_service(event)) if result.rejected: # Circuit is open — route to DLQ instead of retrying route_to_dlq(event, f"circuit_open_{cb.name}", "handler_circuit_breaker") return if not result.success: # Retryable failure within closed/half-open state — retry once more await call_downstream_service(event) ``` --- ### Pattern 5: DLQ Monitoring and Metrics Collection Collects operational metrics on dead letter queue depth, message age, error type distribution, and retry patterns. Produces structured telemetry that feeds monitoring dashboards and alerting systems. ```python from __future__ import annotations import json import logging import time from collections import defaultdict from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from typing import Any logger = logging.getLogger(__name__) @dataclass class DlqMetricsSnapshot: """Point-in-time snapshot of DLQ operational metrics. This structure is suitable for structured logging, Prometheus histogram buckets, or export to a metrics backend like Datadog or New Relic. """ timestamp_utc: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) total_dlq_depth: int = 0 queues: dict[str, "QueueMetrics"] = field(default_factory=dict) error_distribution: dict[str, int] = field(default_factory=lambda: defaultdict(int)) oldest_message_age_seconds: float = 0.0 messages_over_threshold: list["QueueMetrics"] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: return { "timestamp": self.timestamp_utc.isoformat(), "total_depth": self.total_dlq_depth, "queues": {k: asdict(v) for k, v in self.queues.items()}, "error_distribution": dict(self.error_distribution), "oldest_age_seconds": self.oldest_message_age_seconds, "messages_over_threshold": len(self.messages_over_threshold), } @dataclass class QueueMetrics: """Per-queue metrics for a single DLQ.""" queue_name: str = "" depth: int = 0 oldest_message_age_seconds: float = 0.0 error_types: dict[str, int] = field(default_factory=lambda: defaultdict(int)) avg_retries_before_dlq: float = 0.0 messages_over_threshold_count: int = 0 class DlqMetricsCollector: """Collects and aggregates metrics from dead letter queues for monitoring. Scans all DLQs, computes depth, age distribution, error type breakdown, and identifies queues that exceed alerting thresholds. Produces snapshots suitable for dashboard rendering and automated alert evaluation. In production, replace the in-memory scanning with real broker API calls (RabbitMQ HTTP API, Kafka JMX metrics, SQS GetQueueAttributes). """ def __init__( self, age_threshold_seconds: float = 1800.0, # 30 minutes depth_threshold_per_queue: int = 100, ) -> None: self._age_threshold = age_threshold_seconds self._depth_threshold = depth_threshold_per_queue self._historical_snapshots: list[DlqMetricsSnapshot] = [] def scan_queues( self, queue_data: dict[str, list[dict[str, Any]]], ) -> DlqMetricsSnapshot: """Scan DLQ queue data and produce a metrics snapshot. Args: queue_data: Mapping of queue name to list of message dicts. Each message dict should contain: 'age_seconds', 'error_type', 'retry_count', and optionally 'source_queue'. Returns: DlqMetricsSnapshot with computed metrics. """ snapshot = DlqMetricsSnapshot() total_depth = 0 for queue_name, messages in queue_data.items(): if not messages: continue depth = len(messages) total_depth += depth …(truncated)