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.
Scanned 9/4/2026
Install to Claude Code
Skills Directorynpx -y skills add paulpas/agent-skill-router --skill backpressure-handling --agent claude-codeInstalls into .claude/skills of the current project.
Are you the author of Backpressure Handling?
Add the live security badge to your README — it updates automatically with every re-scan.
[](https://www.skillsdirectory.com/skills/paulpas-backpressure-handling)More formats (shields.io, HTML) on the badges page.
---
name: backpressure-handling
description: 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.
license: MIT
compatibility: opencode
metadata:
version: "1.0.0"
domain: coding
triggers: backpressure, flow control, rate limiting, circuit breaker, consumer lag, processing overload, bounded queue, signal propagation
archetypes:
- tactical
- diagnostic
anti_triggers:
- brainstorming
- vague ideation
- code golf
- over-engineering
response_profile:
verbosity: low
directive_strength: high
abstraction_level: operational
role: implementation
scope: implementation
output-format: code
content-types: [code, guidance, do-dont, examples]
related-skills: event-driven-patterns,message-queue-design,system-reliability-architecture,event-bus,data-pipeline-engineering
---
# 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:
import time
self.timestamp = time.monotonic()
class FlowControlChannel:
"""Async channel for propagating backpressure signals from consumers to producers.
Consumers push flow control signals onto the channel when they detect overload.
Producers monitor the channel and respond by pausing, throttling, or resuming.
This implements the signal propagation pattern — the physical mechanism by which
backpressure flows upstream in event-driven systems.
"""
def __init__(self) -> None:
self._signals: asyncio.Queue[FlowControlMessage] = asyncio.Queue(maxsize=100)
self._paused = False
_lock = asyncio.Lock()
async def send_signal(
self,
signal: FlowControlSignal,
reason: str,
) -> None:
"""Send a flow control signal from consumer to producers.
This is called by consumers when they detect overload conditions.
The signal propagates through the channel and triggers producer-side actions.
"""
await self._signals.put(FlowControlMessage(signal=signal, reason=reason))
logger.info("Flow control: %s — %s", signal.value, reason)
async def receive_signals(self) -> AsyncGenerator[FlowControlMessage, None]:
"""Consume flow control signals. Producers iterate this to react to backpressure."""
while True:
message = await self._signals.get()
if message.signal == FlowControlSignal.PAUSE:
logger.warning("PAUSE signal received — stopping publish loop")
yield message
elif message.signal == FlowControlSignal.THROTTLE:
logger.info("THROTTLE signal received — reducing publish rate")
yield message
elif message.signal == FlowControlSignal.RESUME:
logger.info("RESUME signal received — restoring normal rate")
yield message
class BackpressureAwareProducer:
"""A producer that responds to flow control signals from downstream consumers."""
def __init__(self, flow_channel: FlowControlChannel) -> None:
self._flow = flow_channel
self._paused = False
self._batch_size = 100 # Normal batch size
self._throttled_batch_size = 10 # Reduced batch size during throttle
@property
def current_batch_size(self) -> int:
return self._throttled_batch_size if self._paused else self._batch_size
async def run_publish_loop(
self,
event_source: AsyncGenerator[Any, None],
) -> dict[str, int]:
"""Publish events while respecting flow control signals from consumers.
Args:
event_source: Async generator producing events to publish.
Returns:
Statistics about published events and flow control responses.
"""
stats = {"published": 0, "paused_for": 0.0, "batch_reductions": 0}
batch: list[Any] = []
pause_start: float | None = None
async for event in event_source:
# Check for flow control signals before each event
try:
while not self._flow._signals.empty():
signal_msg = await asyncio.wait_for(
self._flow._signals.get(), timeout=0.01
)
if signal_msg.signal == FlowControlSignal.PAUSE:
if not self._paused:
pause_start = time.monotonic() if "time" in dir() else 0
self._paused = True
stats["batch_reductions"] += 1
elif signal_msg.signal == FlowControlSignal.RESUME:
if self._paused and pause_start is not None:
import time as t
stats["paused_for"] += t.monotonic() - pause_start
self._paused = False
pause_start = None
except asyncio.TimeoutError:
pass # No signals available right now
# Batch events for publishing
batch.append(event)
if len(batch) >= (self._throttled_batch_size if self._paused else self._batch_size):
await self._publish_batch(batch)
stats["published"] += len(batch)
batch = []
# Publish remaining events
if batch:
await self._publish_batch(batch)
stats["published"] += len(batch)
return stats
async def _publish_batch(self, batch: list[Any]) -> None:
"""Publish a batch of events to the message broker."""
logger.debug("Publishing batch of %d events", len(batch))
await asyncio.sleep(0.001 * len(batch)) # Simulate network I/O
async def get_status(self) -> dict[str, Any]:
return {
"paused": self._paused,
"batch_size": self.current_batch_size,
"flow_channel_depth": self._flow._signals.qsize(),
}
# --- Usage Example: Closed-Loop Backpressure System ---
async def demonstrate_flow_control() -> None:
"""Demonstrate the complete flow control loop."""
import time
channel = FlowControlChannel()
producer = BackpressureAwareProducer(channel)
async def event_generator():
"""Simulate a stream of events from an upstream source."""
for i in range(200):
yield {"id": i, "type": "user_action"}
await asyncio.sleep(0.001) # Small delay between events
# Run producer with flow control awareness
task = asyncio.create_task(producer.run_publish_loop(event_generator()))
# Simulate consumer detecting overload after some events
async def simulate_consumer_overload():
await asyncio.sleep(0.05) # Let a few events through first
await channel.send_signal(
FlowControlSignal.PAUSE,
"consumer queue depth exceeds 80% capacity",
)
await asyncio.sleep(0.1) # Pause for 100ms
await channel.send_signal(
FlowControlSignal.RESUME,
"queue depth recovered below threshold",
)
await asyncio.gather(
task,
simulate_consumer_overload(),
)
status = await producer.get_status()
print(f"Producer status: {status}")
if __name__ == "__main__":
asyncio.run(demonstrate_flow_control())
```
### Pattern 7: Consumer Lag Monitor and Alerting Thresholds
Effective backpressure requires observability. This monitor tracks queue depth, processing latency percentiles (p50/p95/p99), consumer lag, and resource utilization — providing the metrics that feed into adaptive rate limiting and circuit breaker decisions.
```python
import asyncio
import logging
import math
from dataclasses import dataclass, field
from typing import Any
from collections import deque
logger = logging.getLogger(__name__)
@dataclass
class PercentileTracker:
"""Efficient moving-window percentile tracker using a sorted deque."""
max_window_seconds: float = 60.0
_values: deque[tuple[float, float]] = field(default_factory=deque) # (timestamp, value)
def record(self, value: float) -> None:
now = asyncio.get_event_loop().time() if "asyncio" in dir() else __import__("time").monotonic()
self._values.append((now, value))
def _prune(self, current_time: float) -> None:
cutoff = current_time - self.max_window_seconds
while self._values and self._values[0][0] < cutoff:
self._values.popleft()
def percentile(self, p: float) -> float | None:
"""Calculate the p-th percentile (0-100) of recent values."""
current_time = asyncio.get_event_loop().time() if "asyncio" in dir() else __import__("time").monotonic()
self._prune(current_time)
if not self._values:
return None
sorted_values = sorted(v for _, v in self._values)
index = int(len(sorted_values) * p / 100.0)
index = min(index, len(sorted_values) - 1)
return sorted_values[index]
@property
def count(self) -> int:
current_time = asyncio.get_event_loop().time() if "asyncio" in dir() else __import__("time").monotonic()
self._prune(current_time)
return len(self._values)
@dataclass
class ConsumerLagMetrics:
"""Comprehensive backpressure observability metrics."""
queue_depth: int = 0
p50_latency_ms: float = 0.0
p95_latency_ms: float = 0.0
p99_latency_ms: float = 0.0
events_processed_per_second: float = 0.0
consumer_lag_seconds: float = 0.0
cpu_usage_percent: float = 0.0
memory_pressure_percent: float = 0.0
db_pool_utilization_percent: float = 0.0
@property
def is_overloaded(self) -> bool:
"""Return True if any metric exceeds critical thresholds."""
return (
self.queue_depth > 10000 or
self.p99_latency_ms > 5000 or
self.cpu_usage_percent > 85 or
self.memory_pressure_percent > 85 or
self.db_pool_utilization_percent > 75
)
@property
def is_degraded(self) -> bool:
"""Return True if system is under moderate pressure — early warning."""
return (
self.queue_depth > 5000 or
self.p95_latency_ms > 2000 or
self.cpu_usage_percent > 70 or
self.memory_pressure_percent > 70
)
class ConsumerLagMonitor:
"""Monitors consumer health metrics and provides backpressure observability.
Tracks queue depth, processing latency percentiles, throughput rates, and
system resource utilization. Exposes overload/degraded state flags that can
feed directly into adaptive rate limiters and circuit breakers.
"""
def __init__(self, consumer_name: str = "default") -> None:
self._name = consumer_name
self._latency_tracker = PercentileTracker(max_window_seconds=60.0)
self._throughput_history: deque[tuple[float, int]] = deque(maxlen=100)
self._last_processed_count: int = 0
self._last_timestamp: float | None = None
self._lock = asyncio.Lock()
def record_event_completion(self, processing_time_ms: float) -> None:
"""Record a completed event's processing time for latency tracking."""
self._latency_tracker.record(processing_time_ms)
now = __import__("time").monotonic()
with asyncio.Runner().call(__import__("asyncio").get_running_loop, __import__("asyncio").Lock().__aenter__()) if False else None:
pass # Placeholder — use simple counter in sync context
self._last_processed_count += 1
self._throughput_history.append((now, self._last_processed_count))
def update_queue_depth(self, depth: int) -> None:
"""Update the current queue depth measurement."""
pass # State is accessed via property in real implementation
@property
def queue_depth(self) -> int:
"""Current queue depth — set externally by the consumer."""
return getattr(self, "_queue_depth", 0)
def _compute_throughput(self) -> float:
"""Calculate events processed per second from throughput history."""
if len(self._throughput_history) < 2:
return 0.0
first_time, first_count = self._throughput_history[0]
last_time, last_count = self._throughput_history[-1]
elapsed = last_time - first_time
if elapsed <= 0:
return 0.0
return (last_count - first_count) / elapsed
async def get_metrics(self, system_overrides: dict[str, float] | None = None) -> ConsumerLagMetrics:
"""Compute comprehensive metrics snapshot for backpressure decisions.
Args:
system_overrides: Optional dict of {metric_name: value} to override
real system readings (useful for testing).
Returns:
Full ConsumerLagMetrics snapshot including overload indicators.
"""
overrides = system_overrides or {}
return ConsumerLagMetrics(
queue_depth=overrides.get("queue_depth", self.queue_depth),
p50_latency_ms=self._latency_tracker.percentile(50) or 0.0,
p95_latency_ms=self._latency_tracker.percentile(95) or 0.0,
p99_latency_ms=self._latency_tracker.percentile(99) or 0.0,
events_processed_per_second=self._compute_throughput(),
consumer_lag_seconds=overrides.get("lag_seconds", 0.0),
cpu_usage_percent=overrides.get("cpu_percent", 0.0),
memory_pressure_percent=overrides.get("memory_percent", 0.0),
db_pool_utilization_percent=overrides.get("db_pool_percent", 0.0),
)
async def monitor_loop(
self,
queue_getter: Any = None,
interval_seconds: float = 5.0,
) -> AsyncGenerator[ConsumerLagMetrics, None]:
"""Continuously emit metrics snapshots for observability and alerting.
This coroutine yields a fresh metrics snapshot every `interval_seconds`,
suitable for feeding into monitoring dashboards or adaptive throttling loops.
"""
while True:
try:
metrics = await self.get_metrics()
yield metrics
# Alert on overload conditions
if metrics.is_overloaded:
logger.critical(
"CONSUMER OVERLOAD [%s]: q=%d, p99=%.0fms, cpu=%.0f%%",
self._name, metrics.queue_depth,
metrics.p99_latency_ms, metrics.cpu_usage_percent,
)
elif metrics.is_degraded:
logger.warning(
"CONSUMER DEGRADED [%s]: q=%d, p95=%.0fms, cpu=%.0f%%",
self._name, metrics.queue_depth,
metrics.p95_latency_ms, metrics.cpu_usage_percent,
)
except Exception as e:
logger.error("Metrics collection error: %s", e)
await asyncio.sleep(interval_seconds)
# --- Usage Example: Metrics-Driven Adaptive Backpressure ---
async def demonstrate_monitoring() -> None:
"""Show how metrics drive backpressure decisions."""
monitor = ConsumerLagMonitor(consumer_name="payment-processor")
# Simulate recording event latencies
import random
for i in range(100):
latency = random.gauss(50, 20) # Normal distribution: mean 50ms, sigma 20ms
monitor.record_event_completion(max(1, latency))
metrics = await monitor.get_metrics({
"queue_depth": 7500,
"cpu_percent": 72.0,
"memory_percent": 68.0,
"db_pool_percent": 45.0,
})
print(f"Consumer: {monitor._name}")
print(f" Queue depth: {metrics.queue_depth}")
print(f" Latency — p50: {metrics.p50_latency_ms:.1f}ms, "
f"p95: {metrics.p95_latency_ms:.1f}ms, p99: {metrics.p99_latency_ms:.1f}ms")
print(f" Throughput: {metrics.events_processed_per_second:.1f} events/s")
print(f" Status: {'OVERLOADED' if metrics.is_overloaded else 'DEGRADED' if metrics.is_degraded else 'HEALTHY'}")
if __name__ == "__main__":
asyncio.run(demonstrate_monitoring())
```
---
## Constraints
### MUST DO
- Always use bounded queues (`asyncio.Queue(maxsize=N)`) — never leave consumer queues unbounded in production; calculate N as `consumer_capacity * safety_multiplier (2x)`
- Implement token bucket or sliding window rate limiters before any downstream external I/O call — the limiter is the first line of defense against queue buildup
- Deploy circuit breakers on every consumer that calls an external dependency (database, third-party API, message broker) — track both error rates AND latency as trip signals
- Design priority-based event processing with at least three tiers (critical, high, medium/low) so system degradation doesn't affect business-critical events
- Measure and expose p50/p95/p99 processing latency percentiles, queue depth utilization, and consumer lag — these are the primary indicators for when backpressure should engage
- Implement signal propagation from consumers to producers via async channels or shared flow-control state — backpressure is ineffective if producers cannot detect downstream saturation
- Add alerting thresholds at both degraded (warning) and overloaded (critical) states — use 70% CPU/memory as degradation thresholds, 85% as critical thresholds
### MUST NOT DO
- Never use unbounded queues (`asyncio.Queue()` with no `maxsize`) in any production consumer — an unbounded queue will grow until the process runs out of memory and OOM kills
- Do not implement circuit breakers that silently drop all messages when open — always route to a dead-letter queue or persistence layer so no data is lost during circuit trips
- Never use fixed-window rate limiting for high-throughput systems — boundary effects allow burst traffic at window boundaries; always prefer sliding window or token bucket algorithms
- Do not let backpressure signals require synchronous acknowledgment from producers — flow control channels should be fire-and-forget async signals, not request-response protocols
- Avoid making throttling decisions based on a single metric alone — always use at least two indicators (e.g., queue depth AND latency percentile) to prevent false positive throttle triggers
- Never implement priority queuing without a clear drop policy — if low-priority events can stay in the queue indefinitely, they starve critical events and defeat the purpose of prioritization
---
## Output Template
When implementing or reviewing backpressure handling in an event-driven system, produce:
1. **Architecture Diagram** — ASCII or text-based flow showing all producers, queues, consumers, and circuit breakers with their capacities (maxsize values)
2. **Queue Capacity Calculations** — For each consumer queue: the maxsize value, how it was derived (capacity × safety multiplier), and the expected overflow behavior
3. **Rate Limiter Configuration** — Token bucket rate (events/s, burst capacity) or sliding window parameters (window size, max events) for each downstream call with rationale
4. **Circuit Breaker Settings** — Per-dependency configuration: failure threshold, recovery timeout, half-open probe limits, and what happens when the circuit opens (DLQ routing, degraded path)
5. **Priority Scheme** — Defined priority tiers with event type mappings, drop policies, and the overload thresholds that trigger tier-based filtering
6. **Flow Control Protocol** — How signals propagate from consumers to producers: channel implementation, signal types, and producer response behavior for each signal
7. **Monitoring Plan** — Metrics to expose (queue depth, p50/p95/p99 latency, consumer lag, throughput), alert thresholds (degraded at 70%, overloaded at 85%), and the dashboards or alerting rules to configure
---
## Related Skills
| Skill | Purpose |
|---|---|
| `event-driven-patterns` | Broader event-driven architecture patterns (event sourcing, CQRS) — backpressure is a cross-cutting concern within this architecture |
| `message-queue-design` | Message broker configuration, queue topology, and durability settings — complements application-level backpressure with broker-level flow control |
| `system-reliability-architecture` | System-wide reliability design including redundancy, failover, and disaster recovery — backpressure is one layer in the reliability defense-in-depth strategy |
| `event-bus` | Event bus implementation patterns (publish/subscribe, topic-based routing) — backpressure mechanisms integrate with event bus consumer groups |
| `data-pipeline-engineering` | High-throughput data pipeline construction (Apache Kafka, Flink, Spark Streaming) — backpressure is critical in streaming pipelines to prevent memory pressure |
---
## Live References
| Resource | URL |
|----------|-----|
| Python asyncio.Queue Documentation | https://docs.python.org/3/library/asyncio-stream.html#asyncio.Queue |
| Token Bucket Algorithm (Wikipedia) | https://en.wikipedia.org/wiki/Token_bucket |
| Sliding Window Counter Algorithm | https://medium.com/@gerg.kis/10-microservice-design-patterns-you-should-know-c628e9ebec05 |
| Circuit Breaker Pattern (Martin Fowler) | https://martinfowler.com/bliki/CircuitBreaker.html |
| Backpressure in Reactive Streams (Reactive Streams spec) | https://www.reactivestreams.org/ |
| Kubernetes Resource Management (Memory Pressure) | https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/ |
---
> 📖 skill(local cache): backpressure-handling
Is this your skill, or is something wrong with this listing? Request removal or report an issue. Author removals are honored within 72 hours.
No comments yet. Be the first to comment!
