paulpas 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-patternsinstead - 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
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.
Implement Consumer-Side Bounded Queues — Replace every
Queue()withQueue(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; everyasyncio.Queuehas an explicitmaxsize.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.
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.
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.
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.
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.
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.
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.
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.
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.
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)