Module

_rate_limiter

Rate limiting and backpressure for pounce.

Implements token bucket rate limiting per client IP with request queuing and load shedding for production overload protection.

Per-worker semantics (issue #109): each worker holds its own token buckets. In thread mode (3.14t) one limiter is genuinely shared, but in process mode (GIL build, fork) and subinterpreter mode every worker inherits an independent copy of the bucket state with no IPC between them. The real per-IP ceiling is thereforerate x workers (burst burst x workers). The configured number is the aggregate guarantee only whenworkers=1. See docs/deployment/backpressure.md.

2Classes1Function

Classes

TokenBucket 4 ▼
Token bucket rate limiter for a single client. Classic token bucket algorithm: - Tokens refill at …

Token bucket rate limiter for a single client.

Classic token bucket algorithm:

  • Tokens refill at a constant rate (requests per second)
  • Bucket has a maximum capacity (burst size)
  • Each request consumes one token
  • Requests are denied when bucket is empty

Thread-safe for free-threading mode.

Per-worker (issue #109): a bucket lives in a single worker's address space. With multiple workers the aggregate per-IP allowance scales with the worker count -- see the module docstring anddocs/deployment/backpressure.md.

Methods

consume 0 bool ▼
Try to consume one token.
def consume(self) -> bool
Returns
bool True if token was available, False if rate limited
is_full_at 1 bool ▼
Return True if the bucket would be at capacity after refill at ``now``. Used b…
def is_full_at(self, now: float) -> bool

Return True if the bucket would be at capacity after refill atnow.

Used by RateLimiter cleanup to identify inactive buckets without reaching into private fields from outside the lock.

Parameters
Name Type Description
now —
Returns
bool
idle_since 1 float ▼
Return seconds since this bucket last refilled (i.e. last consume). ``_last_re…
def idle_since(self, now: float) -> float

Return seconds since this bucket last refilled (i.e. last consume).

_last_refill advances on every consume(), so it doubles as a last-touched timestamp. Used by RateLimiter cleanup to evict buckets that have seen no traffic, even when they never refilled to full capacity.

Parameters
Name Type Description
now —
Returns
float
Internal Methods 1 ▼
__init__ 2 ▼
Initialize token bucket.
def __init__(self, rate: float, burst: int) -> None
Parameters
Name Type Description
rate —

Tokens per second to refill
burst —

Maximum tokens (burst capacity)
RateLimiter 3 ▼
Per-IP rate limiter with token buckets. Tracks rate limits per client IP address using token bucke…

Per-IP rate limiter with token buckets.

Tracks rate limits per client IP address using token bucket algorithm. Automatically cleans up stale buckets to prevent memory leaks.

Thread-safe for concurrent worker threads.

Methods

check_rate_limit 1 bool ▼
Check if request is rate limited.
def check_rate_limit(self, client_ip: str) -> bool
Parameters
Name Type Description
client_ip —

Client IP address
Returns
bool True if request is allowed, False if rate limited
Internal Methods 2 ▼
__init__ 3 ▼
Initialize rate limiter.
def __init__(self, rate: float, burst: int, max_tracked_ips: int = DEFAULT_MAX_TRACKED_IPS) -> None
Parameters
Name Type Description
rate —

Requests per second allowed per IP
burst —

Maximum burst size per IP
max_tracked_ips —

Hard cap on the number of distinct client IPs tracked at once. When exceeded, the least-recently-seen bucket is evicted (LRU). Bounds memory under a unique-IP flood.

Default:DEFAULT_MAX_TRACKED_IPS
_maybe_cleanup 0 ▼
Clean up stale buckets to prevent unbounded growth. Triggers when the cleanup …
def _maybe_cleanup(self) -> None

Clean up stale buckets to prevent unbounded growth.

Triggers when the cleanup interval has elapsed OR the map has grown past the tracked-IP cap. Evicts buckets that are either at full capacity (no pending traffic) or idle (no request within the idle window) so sustained floods cannot pin entries forever.

Functions

create_rate_limit_wrapper 2 Callable ▼
Wrap an ASGI app with rate limiting. Intercepts requests and applies rate limi…
def create_rate_limit_wrapper(app: Callable, rate_limiter: RateLimiter) -> Callable

Wrap an ASGI app with rate limiting.

Intercepts requests and applies rate limiting before passing to app. Returns 429 Too Many Requests when rate limit is exceeded.

Parameters
Name Type Description
app Callable

Original ASGI app
rate_limiter RateLimiter

RateLimiter instance
Returns
Callable