# Backpressure Handling

> Implements production-grade backpressure mechanisms (bounded queues, adaptive rate limiting, circuit breakers, priority queuing) to prevent resource exhaustion in high-throughput event-driven message systems.

- Skill: `paulpas/backpressure-handling` (Agent Skill)
- Install (CLI): `npx skillmds@latest add paulpas/backpressure-handling`
- Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/backpressure-handling/raw
- Safety review: PASS (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- License: MIT
- Author: paulpas (https://skillmd.com/u/paulpas)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/paulpas/backpressure-handling

---






# Backpressure Handling for Event-Driven Systems

Implements backpressure mechanisms that flow from slow consumers upstream to producers — bounded queues, adaptive rate limiting, circuit breakers, priority-based processing, and signal propagation patterns. When loaded, this skill makes the model design, implement, and validate flow-control layers that prevent resource exhaustion (OOM, thread starvation, connection pool depletion) in high-throughput asynchronous message systems.

## TL;DR Checklist

- [ ] All consumer queues are bounded with explicit `maxsize` — never use unbounded queues in production
- [ ] Implement a token bucket or sliding window rate limiter before any external I/O call
- [ ] Add a circuit breaker to every downstream dependency that can become slow or unresponsive
- [ ] Design priority-based routing so low-priority events are dropped before system resources degrade
- [ ] Measure queue depth, processing latency percentiles (p50/p95/p99), and consumer lag continuously
- [ ] Propagate backpressure signals upstream via pause publishing, batch size reduction, or async flow-control channels

---

## When to Use

Use this skill when:

- Building high-throughput event-driven systems where producers can outpace consumers during traffic spikes
- Designing message pipelines that connect microservices with differing processing capabilities (e.g., a fast HTTP ingestion layer feeding slower batch processors)
- Experiencing OOM crashes, thread pool starvation, or connection pool exhaustion under load
- Implementing streaming data pipelines where late arrivals or backlogged events must be handled gracefully
- Designing event bus architectures where fan-out to multiple consumers creates asymmetric processing loads
- Adding observability for queue depth, consumer lag, and processing latency as early-warning signals

---

## When NOT to Use

Avoid this skill for:

- Simple synchronous request-response APIs with no message queuing — backpressure is an async/event-driven concern
- Implementing general circuit breaker patterns for HTTP calls without message flow context — use `microservice-resilience-patterns` instead
- Designing retry strategies or bulkhead isolation — those are distinct resilience concerns; backpressure complements but does not replace them
- Low-throughput systems processing fewer than a few hundred events per second — the overhead of flow-control layers may outweigh benefits

---

## Core Workflow

1. **Map the Event Flow and Identify Pressure Points** — Trace every producer-to-consumer path. Identify which components process synchronously (database writes, external API calls) versus asynchronously (in-memory transforms). Mark each step as a potential pressure point where backpressure could be injected. **Checkpoint:** You have a diagram listing every queue, its current capacity (bounded or unbounded), and the expected vs actual processing rate at each stage.

2. **Implement Consumer-Side Bounded Queues** — Replace every `Queue()` with `Queue(maxsize=N)` where N is calculated from consumer capacity times a safety multiplier (typically 2x). Configure producers to block when the queue fills, creating natural backpressure that propagates upstream. **Checkpoint:** No unbounded queues exist in the production codebase; every `asyncio.Queue` has an explicit `maxsize`.

3. **Add Adaptive Rate Limiting** — Deploy a rate limiter (token bucket or sliding window) before each downstream external call. The limiter should adapt its throughput based on real-time system metrics (CPU utilization, memory pressure, database connection pool availability). Start with static thresholds and transition to adaptive mode once observability is in place. **Checkpoint:** Rate limiters expose current throughput metrics; alerts fire when effective rate drops below 50% of configured maximum.

4. **Deploy Circuit Breakers on Slow Dependencies** — For every consumer that calls an external service or database, add a circuit breaker. When the downstream becomes slow (high latency) or failing (errors), open the circuit to prevent queue buildup. Route messages to a dead-letter queue or degraded processing path when the circuit is open. **Checkpoint:** Circuit breaker state transitions are logged and exposed as metrics; every consumer has at least one circuit breaker configured.

5. **Implement Priority-Based Processing During Overload** — Design event classes with priority levels. During normal operation, all events process in priority order. When system load exceeds a threshold, lower-priority events are either delayed or dropped entirely while critical events continue processing. **Checkpoint:** Priority thresholds and drop policies are documented; monitoring shows distinct latency curves for each priority tier.

6. **Establish Signal Propagation Mechanisms** — Design how backpressure information flows from the most constrained consumer back to the original producers. This can be done via async channels, shared state with lock-free reads, or explicit pause/resume protocols on message brokers. **Checkpoint:** Producers can detect when downstream consumers are saturated and respond within 1–2 seconds by throttling their publish rate.

---

## Implementation Patterns

### Pattern 1: Bounded Async Queue with Natural Backpressure

The simplest and most effective backpressure mechanism — a bounded `asyncio.Queue` that blocks producers when full, creating natural upstream pressure without any explicit flow-control protocol.

```python
import asyncio
import logging
from dataclasses import dataclass, field
from enum import IntEnum
from typing import Any

logger = logging.getLogger(__name__)


class EventPriority(IntEnum):
    """Event priority levels for scheduling during overload conditions."""
    CRITICAL = 0       # Must never be dropped (e.g., payment events)
    HIGH = 1           # Dropped only under extreme load (e.g., user actions)
    MEDIUM = 2         # Delayed or dropped when queue is filling (e.g., analytics)
    LOW = 3            # First to drop during overload (e.g., background tasks)


@dataclass(order=True)
class PrioritizedEvent:
    """An event wrapper that enables priority-ordered queue consumption."""
    priority: EventPriority
    sequence_number: int = field(compare=True)
    payload: Any = field(compare=False, repr=False)

    @property
    def timestamp(self) -> float:
        import time
        return time.monotonic()


class BoundedEventQueue:
    """A bounded async queue with backpressure and priority support.

    When the queue reaches maxsize, producers are blocked (natural backpressure).
    During overload, lower-priority events can be rejected before blocking producers.
    """

    def __init__(
        self,
        maxsize: int = 1000,
        drop_on_overload: bool = True,
        drop_below_priority: EventPriority = EventPriority.MEDIUM,
    ) -> None:
        """Initialize the bounded event queue.

        Args:
            maxsize: Maximum number of events in the queue before producers block.
            drop_on_overload: If True, drop low-priority events before blocking producers.
            drop_below_priority: Events at or below this priority are dropped during overload.
        """
        self._queue: asyncio.PriorityQueue[PrioritizedEvent] = asyncio.PriorityQueue(
            maxsize=maxsize
        )
        self._maxsize = maxsize
        self._drop_on_overload = drop_on_overload
        self._drop_below_priority = drop_below_priority
        self._sequence_counter: int = 0
        self._stats = {
            "enqueued": 0,
            "dequeued": 0,
            "dropped": 0,
            "blocked_producers": 0,
        }

    @property
    def current_size(self) -> int:
        return self._queue.qsize()

    @property
    def utilization(self) -> float:
        """Return queue utilization as a fraction (0.0 to 1.0)."""
        return self.current_size / self._maxsize if self._maxsize > 0 else 0.0

    @property
    def is_near_capacity(self) -> bool:
        """Return True when queue exceeds 80% capacity — overload warning threshold."""
        return self.utilization >= 0.8

    async def put(
        self,
        event: Any,
        priority: EventPriority = EventPriority.MEDIUM,
        *,
        timeout: float | None = None,
    ) -> bool:
        """Put an event into the queue with backpressure handling.

        Args:
            event: The payload to enqueue.
            priority: Event priority level for scheduling.
            timeout: Optional timeout in seconds for blocking put. None means block indefinitely.

        Returns:
            True if event was enqueued, False if it was dropped due to overload.
        """
        # During overload, drop low-priority events instead of blocking producers
        if self._drop_on_overload and self.is_near_capacity:
            if priority <= self._drop_below_priority:
                self._stats["dropped"] += 1
                logger.debug(
                    "Dropped %s event (priority=%d) — queue near capacity (%.0f%%)",
                    type(event).__name__, priority, self.utilization * 100,
                )
                return False

        # Block producer when queue is full — this IS the backpressure signal
        wrapped = PrioritizedEvent(
            priority=priority,
            sequence_number=self._sequence_counter,
            payload=event,
        )
        self._sequence_counter += 1

        try:
            if timeout is not None:
                await asyncio.wait_for(self._queue.put(wrapped), timeout=timeout)
            else:
                await self._queue.put(wrapped)

            self._stats["enqueued"] += 1
            return True

        except (asyncio.QueueFull, asyncio.TimeoutError):
            self._stats["blocked_producers"] += 1
            logger.warning(
                "Queue full — producer blocked for %s (utilization=%.0f%%)",
                type(event).__name__, self.utilization * 100,
            )
            return False

    async def get(self) -> PrioritizedEvent:
        """Get the highest-priority event from the queue. Blocks until available."""
        event = await self._queue.get()
        self._stats["dequeued"] += 1
        return event

    def task_done(self) -> None:
        """Mark a task as done (required for join())."""
        self._queue.task_done()

    @property
    def stats(self) -> dict[str, int]:
        return self._stats.copy()

    async def wait_empty(self) -> None:
        """Wait until all queued events have been processed."""
        await self._queue.join()


# --- Usage Example ---
async def main():
    queue = BoundedEventQueue(maxsize=100, drop_on_overload=True)

    # Producer side — automatically blocked when queue is full
    async def producer(events: list[str], priority: EventPriority) -> int:
        sent = 0
        for event in events:
            if await queue.put(event, priority):
                sent += 1
        return sent

    # Consumer side — processes in priority order
    async def consumer(consumer_id: int) -> None:
        while True:
            try:
                event = await asyncio.wait_for(queue.get(), timeout=5.0)
                logger.info(
                    "Consumer %d processing %s (priority=%d, seq=%d)",
                    consumer_id, type(event.payload).__name__,
                    event.priority, event.sequence_number,
                )
                await asyncio.sleep(0.01)  # Simulate work
                queue.task_done()
            except asyncio.TimeoutError:
                break

    # Run the pipeline
    tasks = [
        asyncio.create_task(consumer(i)) for i in range(3)
    ]
    await producer([f"event-{i}" for i in range(50)], EventPriority.HIGH)

    await queue.wait_empty()
    for t in tasks:
        t.cancel()

    print(f"Queue stats: {queue.stats}")
    print(f"Final utilization: {queue.utilization:.1%}")


if __name__ == "__main__":
    asyncio.run(main())
```

### Pattern 2: Token Bucket Rate Limiter

The token bucket algorithm controls throughput by regulating how many operations can execute per time unit. Tokens are added at a fixed rate; each operation consumes one token. If no tokens are available, the caller blocks or fails — implementing smooth rate limiting that handles burst traffic gracefully.

```python
import asyncio
import time
import logging
from dataclasses import dataclass
from typing import AsyncGenerator, Any

logger = logging.getLogger(__name__)


@dataclass
class RateLimiterMetrics:
    """Runtime metrics for a token bucket rate limiter."""
    tokens_per_second: float
    burst_capacity: int
    current_tokens: float
    total_allowed: int = 0
    total_rejected: int = 0
    total_wait_time_seconds: float = 0.0

    @property
    def rejection_rate(self) -> float:
        total = self.total_allowed + self.total_rejected
        return self.total_rejected / total if total > 0 else 0.0


class TokenBucketRateLimiter:
    """Token bucket rate limiter with async support and metrics collection.

    Tokens refill at a constant rate up to a maximum burst capacity.
    Each acquire() call consumes one token. When no tokens are available,
    the caller waits (async) until a token is refilled — implementing
    smooth backpressure on downstream services.
    """

    def __init__(
        self,
        rate: float,              # Tokens added per second
        burst_capacity: int = 10, # Maximum tokens (burst limit)
        initial_tokens: float | None = None,
    ) -> None:
        """Initialize the token bucket.

        Args:
            rate: Tokens added per second (sustained throughput).
            burst_capacity: Maximum tokens allowed — controls burst size.
            initial_tokens: Starting token count. Defaults to burst_capacity.
        """
        self._rate = rate
        self._burst_capacity = burst_capacity
        self._tokens = initial_tokens if initial_tokens is not None else float(burst_capacity)
        self._last_refill_time = time.monotonic()
        self._lock = asyncio.Lock()
        self._metrics = RateLimiterMetrics(
            tokens_per_second=rate,
            burst_capacity=burst_capacity,
            current_tokens=self._tokens,
        )

    @property
    def metrics(self) -> RateLimiterMetrics:
        return self._metrics

    def _refill(self) -> None:
        """Refill tokens based on elapsed time. Must be called under lock."""
        now = time.monotonic()
        elapsed = now - self._last_refill_time
        self._tokens = min(
            self._burst_capacity,
            self._tokens + elapsed * self._rate,
        )
        self._last_refill_time = now

    async def acquire(self, tokens: int = 1, *, timeout: float | None = None) -> bool:
        """Acquire one or more tokens, waiting (backpressure) if none available.

        Args:
            tokens: Number of tokens to consume.
            timeout: Maximum seconds to wait for tokens. None waits indefinitely.

        Returns:
            True if tokens were acquired, False if timeout expired.
        """
        deadline = None if timeout is None else time.monotonic() + timeout

        while True:
            async with self._lock:
                self._refill()

                if self._tokens >= tokens:
                    self._tokens -= tokens
                    self._metrics.total_allowed += 1
                    self._metrics.current_tokens = self._tokens
                    return True

                # Calculate wait time until enough tokens are available
                deficit = tokens - self._tokens
                wait_time = deficit / self._rate

            # Check timeout before waiting
            if deadline is not None and time.monotonic() >= deadline:
                self._metrics.total_rejected += 1
                return False

            # Wait for token refill — this IS the backpressure mechanism
            await asyncio.sleep(min(wait_time, deadline - time.monotonic()) if deadline else wait_time)

    async def try_acquire(self, tokens: int = 1) -> bool:
        """Non-blocking attempt to acquire tokens. Returns immediately."""
        async with self._lock:
            self._refill()
            if self._tokens >= tokens:
                self._tokens -= tokens
                self._metrics.total_allowed += 1
                self._metrics.current_tokens = self._tokens
                return True
            self._metrics.total_rejected += 1
            return False

    @property
    def available_rate(self) -> float:
        """Current effective throughput rate after refill calculations."""
        return min(self._rate, self._tokens / max(time.monotonic() - self._last_refill_time, 0.001))


# --- Usage Example: Rate-Limited Downstream Call ---
class RateLimitedServiceClient:
    """Example of a service client protected by token bucket rate limiting."""

    def __init__(self, events_per_second: float = 50.0, burst_size: int = 100) -> None:
        self._limiter = TokenBucketRateLimiter(
            rate=events_per_second,
            burst_capacity=burst_size,
        )

    async def process_event(self, event_data: dict[str, Any]) -> bool:
        """Process a single event with rate limiting backpressure."""
        acquired = await self._limiter.acquire(timeout=10.0)
        if not acquired:
            logger.warning("Rate limit timeout — event dropped: %s", event_data.get("type"))
            return False

        # Simulate downstream processing
        await asyncio.sleep(0.01)
        return True

    @property
    def metrics(self) -> RateLimiterMetrics:
        return self._limiter.metrics
```

### Pattern 3: Adaptive Rate Limiter Driven by System Load Metrics

Static rate limits become ineffective when system conditions change. An adaptive rate limiter reads real-time CPU, memory, and database connection pool utilization to dynamically adjust throughput — reducing rates before the system degrades rather than crashing after.

```python
import asyncio
import os
import logging
from dataclasses import dataclass, field
from typing import Protocol
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)


class SystemMetricsProvider(Protocol):
    """Interface for providing real-time system load metrics."""

    async def get_cpu_usage(self) -> float: ...       # 0.0 to 1.0
    async def get_memory_pressure(self) -> float: ...  # 0.0 to 1.0
    async def get_db_pool_utilization(self) -> float: ...  # 0.0 to 1.0


class DummyMetricsProvider:
    """Default metrics provider using real system data for demonstration."""

    async def get_cpu_usage(self) -> float:
        try:
            import psutil
            return psutil.cpu_percent(interval=0.1) / 100.0
        except ImportError:
            return 0.3  # Default assumption if psutil unavailable

    async def get_memory_pressure(self) -> float:
        try:
            import psutil
            mem = psutil.virtual_memory()
            return mem.used / mem.total
        except ImportError:
            return 0.5

    async def get_db_pool_utilization(self) -> float:
        # In production, this queries your DB driver's pool status
        return 0.4


@dataclass
class AdaptiveRateConfig:
    """Configuration for adaptive rate limiting behavior."""
    # Throughput bounds
    max_events_per_second: float = 100.0
    min_events_per_second: float = 5.0

    # System load thresholds (0.0 to 1.0) — trigger throttling at these levels
    cpu_threshold_high: float = 0.80
    memory_threshold_high: float = 0.85
    db_pool_threshold_high: float = 0.75

    # Throttling aggressiveness (0.0 = gentle, 1.0 = aggressive)
    throttle_aggressiveness: float = 0.6

    # Smoothing to prevent rate oscillation
    smoothing_factor: float = 0.3  # New rate = old * (1-factor) + new * factor


class AdaptiveRateLimiter:
    """Dynamically adjusts throughput based on real-time system load metrics.

    When CPU, memory, or DB pool utilization exceeds thresholds, the effective
    rate is reduced proportionally — providing automatic backpressure that adapts
    to system conditions without manual tuning.
    """

    def __init__(
        self,
        config: AdaptiveRateConfig | None = None,
        metrics_provider: SystemMetricsProvider | None = None,
    ) -> None:
        self._config = config or AdaptiveRateConfig()
        self._provider = metrics_provider or DummyMetricsProvider()
        self._current_rate: float = self._config.max_events_per_second
        self._target_rate: float = self._config.max_events_per_second
        self._lock = asyncio.Lock()

    @property
    def effective_rate(self) -> float:
        return self._current_rate

    async def compute_target_rate(self) -> float:
        """Compute the ideal throughput based on current system metrics.

        Uses a weighted penalty approach: each overloaded subsystem reduces
        the target rate by its excess load multiplied by throttle aggressiveness.
        """
        cpu = await self._provider.get_cpu_usage()
        memory = await self._provider.get_memory_pressure()
        db_pool = await self._provider.get_db_pool_utilization()

        penalties: list[float] = []

        # CPU penalty — linear increase in excess
        if cpu > self._config.cpu_threshold_high:
            excess = (cpu - self._config.cpu_threshold_high) / (
                1.0 - self._config.cpu_threshold_high
            )
            penalties.append(excess * self._config.throttle_aggressiveness)

        # Memory penalty — more aggressive due to OOM risk
        if memory > self._config.memory_threshold_high:
            excess = (memory - self._config.memory_threshold_high) / (
                1.0 - self._config.memory_threshold_high
            )
            penalties.append(excess * min(1.0, self._config.throttle_aggressiveness * 1.2))

        # DB pool penalty — connection exhaustion is catastrophic
        if db_pool > self._config.db_pool_threshold_high:
            excess = (db_pool - self._config.db_pool_threshold_high) / (
                1.0 - self._config.db_pool_threshold_high
            )
            penalties.append(excess * self._config.throttle_aggressiveness)

        # Calculate target rate reduction
        if penalties:
            total_penalty = sum(penalties) / len(penalties)
            reduction_factor = max(0.0, 1.0 - total_penalty)
        else:
            reduction_factor = 1.0

        target = self._config.max_events_per_second * reduction_factor
        return max(self._config.min_events_per_second, min(
            self._config.max_events_per_second, target
        ))

    async def update_rate(self) -> float:
        """Compute new target rate and apply smoothing to avoid oscillation."""
        new_target = await self.compute_target_rate()

        async with self._lock:
            # Apply exponential smoothing to prevent rate oscillation
            old = self._current_rate
            self._current_rate = (
                old * (1.0 - self._config.smoothing_factor) +
                new_target * self._config.smoothing_factor
            )

        logger.debug(
            "Adaptive rate updated: %.1f -> %.1f events/s "
            "(smoothed from %.1f)",
            old, self._current_rate, new_target,
        )
        return self._current_rate

    async def acquire(self) -> bool:
        """Non-blocking check — always returns True. The limiter controls rate externally."""
        # This method exists so the limiter can be used uniformly with static limiters
        return True


# --- Usage Example: Adaptive Rate-Limited Consumer ---
async def adaptive_consumer_pipeline(
    events: asyncio.Queue[dict],
    limiter: AdaptiveRateLimiter,
) -> None:
    """Consumer that adapts its processing rate to system load."""
    while True:
        # Periodically update the effective rate based on system health
        await limiter.update_rate()

        try:
            event = await asyncio.wait_for(events.get(), timeout=1.0)
        except asyncio.TimeoutError:
            continue

        # The limiter's effective_rate dictates how fast we should process.
        # If it drops to 10 events/s due to high CPU, we naturally slow down
        # by increasing our wait between events.
        min_delay = 1.0 / limiter.effective_rate if limiter.effective_rate > 0 else 0.1

        # Process the event
        logger.info("Processing event (rate=%.1f evt/s, delay=%.3fs)",
                     limiter.effective_rate, min_delay)

        # Simulate variable processing time
        await asyncio.sleep(min_delay)
```

### Pattern 4: Circuit Breaker for Message Consumers

When a downstream service becomes slow or unresponsive, messages pile up in the consumer's queue. A circuit breaker detects this degradation and opens — stopping new work from being dispatched to the failing dependency while allowing existing queued work to drain. Messages are rerouted to a dead-letter queue or degraded processing path.

```python
import asyncio
import logging
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, TypeVar

T = TypeVar("T")

logger = logging.getLogger(__name__)


class CircuitState(Enum):
    """Three-state circuit breaker lifecycle."""
    CLOSED = "closed"         # Normal: pass-through, monitoring errors
    OPEN = "open"             # Tripped: reject new work, queue for later
    HALF_OPEN = "half_open"   # Testing: allow limited probe requests


@dataclass
class ConsumerCircuitBreakerConfig:
    """Configuration for a consumer-side circuit breaker."""
    failure_threshold: int = 5            # Successive failures before opening
    recovery_timeout_seconds: float = 30.0  # Time in Open before Half-Open
    half_open_max_calls: int = 3          # Probe calls allowed in Half-Open
    success_threshold: int = 2            # Successes needed to close from Half-Open
    latency_threshold_ms: float = 5000.0  # Slow call threshold (ms)


class CircuitTripError(Exception):
    """Raised when work is rejected by an open circuit breaker."""

    def __init__(self, dependency: str, state: CircuitState) -> None:
        self.dependency = dependency
        self.state = state
        super().__init__(
            f"Circuit breaker {state.value} for '{dependency}' — "
            f"rejecting work to prevent queue buildup"
        )


@dataclass
class CircuitBreakerMetrics:
    """Observable metrics for circuit breaker monitoring."""
    total_calls: int = 0
    successful_calls: int = 0
    failed_calls: int = 0
    rejected_calls: int = 0
    state_transitions: list[dict] = field(default_factory=list)

    def record_transition(self, from_state: CircuitState, to_state: CircuitState, reason: str) -> None:
        self.state_transitions.append({
            "from": from_state.value,
            "to": to_state.value,
            "reason": reason,
            "timestamp": time.time(),
        })


class ConsumerCircuitBreaker:
    """Circuit breaker that protects downstream dependencies in message consumers.

    When a consumer's downstream call becomes slow or fails repeatedly, the circuit
    opens and rejects new work — preventing queue overflow and OOM. Existing queued
    work continues processing (for retries of previously accepted calls), but no NEW
    work is dispatched until recovery is detected.
    """

    def __init__(
        self,
        dependency_name: str,
        config: ConsumerCircuitBreakerConfig | None = None,
    ) -> None:
        self._name = dependency_name
        self._config = config or ConsumerCircuitBreakerConfig()
        self._state = CircuitState.CLOSED
        self._failure_count: int = 0
        self._success_count: int = 0
        self._half_open_calls: int = 0
        self._opened_at: float | None = None
        self._metrics = CircuitBreakerMetrics()

    @property
    def state(self) -> CircuitState:
        """Get current state, with auto-transition from Open to Half-Open."""
        if self._state == CircuitState.OPEN and self._opened_at:
            elapsed = time.monotonic() - self._opened_at
            if elapsed >= self._config.recovery_timeout_seconds:
                old_state = self._state
                self._transition(CircuitState.HALF_OPEN, "recovery timeout expired")
        return self._state

    @property
    def metrics(self) -> CircuitBreakerMetrics:
        return self._metrics

    def _transition(self, new_state: CircuitState, reason: str) -> None:
        old = self._state
        self._state = new_state
        self._metrics.record_transition(old, new_state, reason)
        logger.info(
            "Circuit '%s': %s -> %s (%s)",
            self._name, old.value, new_state.value, reason,
        )

    def allow_work(self) -> bool:
        """Check if new work can be dispatched to the downstream dependency.

        Returns True if the circuit allows work (Closed or Half-Open with capacity).
        Raises CircuitTripError if the circuit is Open — this is the backpressure signal.
        """
        current = self.state  # Triggers Open -> Half-Open transition check

        if current == CircuitState.CLOSED:
            return True
        elif current == CircuitState.OPEN:
            self._metrics.rejected_calls += 1
            raise CircuitTripError(self._name, CircuitState.OPEN)
        else:  # HALF_OPEN
            if self._half_open_calls < self._config.half_open_max_calls:
                self._half_open_calls += 1
                return True
            self._metrics.rejected_calls += 1
            raise CircuitTripError(self._name, CircuitState.HALF_OPEN)

    def record_success(self, latency_ms: float = 0.0) -> None:
        """Record a successful call to the downstream dependency."""
        self._metrics.total_calls += 1
        self._metrics.successful_calls += 1

        if self._state == CircuitState.CLOSED:
            self._failure_count = 0
        elif self._state == CircuitState.HALF_OPEN:
            self._success_count += 1
            if self._success_count >= self._config.success_threshold:
                self._half_open_calls = 0
                self._success_count = 0
                self._failure_count = 0
                self._transition(CircuitState.CLOSED, "recovery confirmed")

    def record_failure(self) -> None:
        """Record a failed or slow call to the downstream dependency."""
        self._metrics.total_calls += 1
        self._metrics.failed_calls += 1

        if self._state == CircuitState.CLOSED:
            self._failure_count += 1
            if self._failure_count >= self._config.failure_threshold:
                self._opened_at = time.monotonic()
                self._transition(CircuitState.OPEN, f"{self._failure_count} failures exceeded threshold")
        elif self._state == CircuitState.HALF_OPEN:
            # Any failure in Half-Open immediately re-trips the circuit
            self._half_open_calls = 0
            self._opened_at = time.monotonic()
            self._transition(CircuitState.OPEN, "probe call failed")


# --- Usage Example: Circuit-Guarded Message Processing ---
class MessageProcessorWithCircuitBreaker:
    """Example showing how a circuit breaker protects message consumers."""

    def __init__(self) -> None:
        self._circuit = ConsumerCircuitBreaker("payment-gateway")

    async def process_payment(self, order_id: str, amount: float) -> dict[str, Any]:
        """Process a payment through the circuit breaker protected gateway."""
        try:
            # This call blocks if circuit is open — backpressure signal to caller
            self._circuit.allow_work()

            # Simulate downstream call (would be an actual HTTP/API call in production)
            await asyncio.sleep(0.05)
            result = {"order_id": order_id, "status": "processed", "amount": amount}

            self._circuit.record_success(latency_ms=50)
            return result

        except CircuitTripError:
            # Circuit is open — route to dead-letter queue instead of queuing forever
            logger.warning("Circuit open for payment-gateway — routing %s to DLQ", order_id)
            return {"order_id": order_id, "status": "dead_lettered", "reason": "circuit_open"}

    def get_status_report(self) -> dict[str, Any]:
        """Return circuit breaker status for monitoring dashboards."""
        m = self._circuit.metrics
        return {
            "dependency": self._circuit._name,
            "state": self._circuit.state.value,
            "total_calls": m.total_calls,
            "success_rate": m.successful_calls / max(m.total_calls, 1),
            "rejected_count": m.rejected_calls,
            "recent_transitions": m.state_transitions[-5:],
        }
```

### Pattern 5: Sliding Window Throttling Algorithm

The sliding window algorithm divides time into discrete buckets and counts events per bucket. When the sum of events in recent buckets exceeds a threshold, new events are rejected. This provides smoother rate limiting than fixed-window approaches and eliminates boundary burst issues.

```python
import asyncio
import logging
import time
from collections import deque
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class SlidingWindowMetrics:
    """Runtime metrics for a sliding window throttle."""
    window_size_seconds: float
    max_events_per_window: int
    current_count: int
    total_allowed: int = 0
    total_rejected: int = 0

    @property
    def utilization(self) -> float:
        return self.current_count / self.max_events_per_window if self.max_events_per_window > 0 else 0.0


class SlidingWindowThrottle:
    """Sliding window rate throttler with time-bucketed event counting.

    Events are tracked in fixed-size time buckets. The throttle counts events
    across all non-expired buckets to determine current throughput. When the
    total exceeds the threshold, new events are rejected — providing smooth,
    boundary-free rate limiting.
    """

    def __init__(
        self,
        max_events: int,                # Max events allowed per window
        window_size_seconds: float = 1.0,
        bucket_duration_ms: float = 100.0,  # Sub-bucket granularity (ms)
    ) -> None:
        """Initialize the sliding window throttle.

        Args:
            max_events: Maximum events allowed per window period.
            window_size_seconds: Duration of the sliding window.
            bucket_duration_ms: Granularity of sub-buckets for finer resolution.
        """
        self._max_events = max_events
        self._window_size = window_size_seconds
        self._bucket_duration = bucket_duration_ms / 1000.0  # Convert to seconds
        self._buckets: deque[tuple[float, int]] = deque()  # (bucket_start_time, count)
        self._lock = asyncio.Lock()
        self._metrics = SlidingWindowMetrics(
            window_size_seconds=window_size_seconds,
            max_events_per_window=max_events,
            current_count=0,
        )

    def _cleanup_expired_buckets(self, now: float) -> None:
        """Remove buckets that have fallen outside the sliding window."""
        window_start = now - self._window_size
        while self._buckets and self._buckets[0][0] < window_start:
            self._buckets.popleft()

    def _current_count(self, now: float) -> int:
        """Calculate total events across all active buckets."""
        return sum(count for _, count in self._buckets)

    async def allow_event(self) -> bool:
        """Check if an event is allowed under the current rate limit.

        Returns True if the event can proceed, False if it would exceed
        the sliding window threshold (backpressure signal to caller).
        """
        now = time.monotonic()

        async with self._lock:
            self._cleanup_expired_buckets(now)

            current = self._current_count(now)
            self._metrics.current_count = current

            if current < self._max_events:
                # Calculate which bucket this event falls into
                bucket_start = now - (now % self._bucket_duration)
                self._metrics.total_allowed += 1

                # Add to existing bucket or create new one
                if self._buckets and abs(self._buckets[-1][0] - bucket_start) < 0.0001:
                    last_time, last_count = self._buckets.pop()
                    self._buckets.append((last_time, last_count + 1))
                else:
                    self._buckets.append((bucket_start, 1))

                return True
            else:
                self._metrics.current_count = current
                self._metrics.total_rejected += 1
                logger.debug(
                    "Sliding window throttle: REJECTED (count=%d/%d)",
                    current, self._max_events,
                )
                return False

    @property
    def metrics(self) -> SlidingWindowMetrics:
        return self._metrics

    async def wait_for_slot(self, timeout: float | None = None) -> bool:
        """Wait until a slot opens in the window (blocking backpressure).

        Args:
            timeout: Maximum seconds to wait. None waits indefinitely.

        Returns:
            True if a slot became available, False if timed out.
        """
        deadline = time.monotonic() + timeout if timeout else None

        while True:
            now = time.monotonic()

            if deadline is not None and now >= deadline:
                return False

            if await self.allow_event():
                return True

            # Sleep for the smallest bucket duration before re-checking
            await asyncio.sleep(self._bucket_duration)


# --- Usage Example: Sliding Window on High-Throughput Ingestion ---
class HighThroughputIngestor:
    """Example of a high-throughput event ingester using sliding window throttling."""

    def __init__(self, max_rps: int = 1000) -> None:
        self._throttle = SlidingWindowThrottle(
            max_events=max_rps,
            window_size_seconds=1.0,
            bucket_duration_ms=50.0,
        )

    async def ingest(self, event: dict[str, Any]) -> bool:
        """Ingest a single event with sliding window rate control."""
        if await self._throttle.allow_event():
            # Process event — write to storage, dispatch downstream, etc.
            logger.debug("Ingested %s", event.get("type"))
            return True
        else:
            # Backpressure applied — reject or buffer
            logger.warning("Throttled event ingestion (rate limit reached)")
            return False

    @property
    def metrics(self) -> SlidingWindowMetrics:
        m = self._throttle.metrics
        m.total_allowed += 0  # Just pass through for consistency
        return m
```

### Pattern 6: Producer Pause with Flow-Control Signal Propagation

Backpressure must flow upstream from the slowest consumer to producers. This pattern implements an async signal channel that consumers use to tell producers to pause, reduce batch size, or resume publishing — closing the feedback loop.

```python
import asyncio
import logging
from dataclasses import dataclass
from enum import Enum
from typing import Any, AsyncGenerator

logger = logging.getLogger(__name__)


class FlowControlSignal(Enum):
    """Signals that flow from consumers back to producers."""
    PAUSE = "pause"          # Stop publishing immediately
    THROTTLE = "throttle"    # Continue but reduce publish rate
    RESUME = "resume"        # Resume normal publish rate


@dataclass
class FlowControlMessage:
    """Flow-control message carrying backpressure signals from consumers to producers."""
    signal: FlowControlSignal
    reason: str
    timestamp: float = 0.0

    def __post_init__(self) -> None:
        if self.timestamp == 0.0:
       

…(truncated)