11. Token-Bucket Rate Limiter for the Streaming Endpoint

Problem

Cartesia exposes its real-time text-to-speech and speech-to-text products over a WebSocket API gated by per-customer API keys. A shared streaming endpoint like this needs to protect itself from any single key hammering it with connection or synthesis requests, while still letting a key burst up to some ceiling for legitimate spiky traffic (e.g. a voice agent opening several contexts back-to-back at call start).

Implement a per-key token-bucket limiter: each key's bucket holds up to capacity tokens, refills continuously at rate tokens per second, and each admitted call spends cost tokens (default 1). A call is admitted only if enough tokens are available.

Source: src/11_api_key_rate_limiter.py

class TokenBucketLimiter:
    def __init__(self, rate: float, capacity: int): ...
    def allow(self, key: str, timestamp: float, cost: int = 1) -> bool: ...

>>> limiter = TokenBucketLimiter(rate=1.0, capacity=3)
>>> limiter.allow("key1", 0.0)
True
>>> limiter.allow("key1", 0.0)
True
>>> limiter.allow("key1", 0.0)
True
>>> limiter.allow("key1", 0.0)
False
>>> limiter.allow("key1", 2.0)
True

Step-by-Step Approach

1. Store, per key, a pair (tokens_available, last_refill_timestamp) lazily — a brand-new key starts with a full bucket (capacity tokens) "refilled" as of the current call's timestamp.
2. On each allow(key, timestamp, cost) call, compute elapsed = timestamp - last_refill_timestamp and refill: tokens = min(capacity, tokens + elapsed * rate). The min caps the bucket so idle time doesn't let a key bank unlimited tokens for one giant future burst.
3. If the refilled tokens >= cost, admit the call: subtract cost, store the new (tokens, timestamp), and return True.
4. Otherwise deny the call, but still store the refilled (tokens, timestamp) — the refill that happened this call is real and must not be recomputed (double-counted) on the next call.
5. No explicit background timer or thread is needed: because refill is computed lazily from elapsed time at each call, the bucket is always correct as of "now" without any polling.

The key insight is that a token bucket never needs a background refill loop — each call lazily "catches up" the bucket by computing tokens earned since the last touch from elapsed time alone, which keeps each call O(1) regardless of how long the key has been idle.

Reference solution

class TokenBucketLimiter:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate
        self.capacity = capacity
        # key -> (tokens_available, last_refill_timestamp); lazily created per key
        self._buckets: dict[str, tuple[float, float]] = {}

    def allow(self, key: str, timestamp: float, cost: int = 1) -> bool:
        # unseen keys start with a full bucket, "refilled" as of now
        tokens, last_ts = self._buckets.get(key, (float(self.capacity), timestamp))
        # refill proportional to elapsed time, capped so bursts can't exceed capacity
        elapsed = timestamp - last_ts
        tokens = min(self.capacity, tokens + elapsed * self.rate)
        if tokens >= cost:
            tokens -= cost
            self._buckets[key] = (tokens, timestamp)
            return True
        # denied: still record the refill so the next call isn't double-counted
        self._buckets[key] = (tokens, timestamp)
        return False

Key Functions & Tricks

• dict.get(key, default) — supplies a lazily-computed full bucket for a never-seen key without a separate init pass.
• elapsed * rate — converts idle time directly into earned tokens — the core of lazy (as opposed to timer-driven) refill.
• min(capacity, tokens + earned) — caps the bucket so a long-idle key can't accumulate an unbounded burst allowance.
• cost parameter — lets a single call spend more than one token, useful for weighting expensive operations differently from cheap ones.
• storing tokens on denial too — prevents re-crediting the same elapsed window twice on the next call.

How to Recognize This Pattern