Token Bucket Rate Limiter

Implement a per-key token bucket limiter. Each caller owns a refillable balance, so one caller's burst must never spend another caller's capacity.

Requirements

• Define Decision(allowed, retry_after, reason).
• Define TokenBucketLimiter(capacity, refill_rate_per_second, now).
• allow(key, cost=1) returns a Decision.
• Each key has an independent bucket that starts at full capacity.
• Tokens refill continuously based on elapsed time and never exceed capacity.
• A blocked request doesn't consume tokens.
• If now() moves backward, treat elapsed time as zero and keep the previous refill timestamp. Returning to that timestamp later must not create refill credit.
• If a request is blocked, retry_after is the token-deficit delay at the configured refill rate, assuming forward clock progress and no intervening consumption. It doesn't reserve tokens.
• When a request is allowed: reason == "allowed" and retry_after == 0.
• When a request is blocked: reason == "rate_limited" and retry_after == (cost - tokens) / refill_rate_per_second (using the bucket's current tokens after refill).
• Assume finite values with capacity > 0, 0 < cost <= capacity, and refill_rate_per_second > 0. Invalid inputs are out of scope.
• During a clock regression, the estimate is measured from the preserved refill timestamp, not the earlier clock reading. For example, an empty bucket last seen at 10, called at 5 with cost/rate 1, reports 1 but won't refill until the clock passes 10.

Example

The following clock trace starts with capacity 2 and a refill rate of 1 token per second:

The blocked middle request leaves its half-token intact. The runnable assertions below cover the full-bucket and empty-bucket boundaries:

clock = {"t": 0.0}
limiter = TokenBucketLimiter(capacity=2, refill_rate_per_second=1, now=lambda: clock["t"])
assert limiter.allow("org").allowed
assert limiter.allow("org").allowed
blocked = limiter.allow("org")
assert not blocked.allowed and blocked.retry_after == 1
assert blocked.reason == "rate_limited"
full = TokenBucketLimiter(3, 1, now=lambda: 0).allow("org", cost=3)
assert (full.allowed, full.retry_after, full.reason) == (True, 0, "allowed")

Constraints

• Use deterministic injected time.
• Use standard-library Python only.
• Keys are hashable and calls are sequential. Thread safety is a follow-up.
• Preserve the last valid refill timestamp when the injected clock moves backward.