Before you start

Rate limiters are Strategy pattern in the wild. Every production system has one; every design interviewer might ask you to design one.

The Strategy signal, unmistakable

Two algorithms (Token Bucket, Fixed Window) that answer the same question ("allow this request?") through different math. That's the Strategy signal, textbook.

Your design.md Patterns section should name Strategy AND explain why it's clearly right (not just "seems like a good fit").

Where each algorithm wins

• Token Bucket — smooths burst traffic. A user can save up tokens during quiet periods and burst later, up to the capacity. Good for APIs that expect chatty usage.
• Fixed Window — hard cap per time slice. Simple to understand and monitor. Downside: allows 2× the rate right at window boundaries (the last window's leftover plus the new window's full quota, back to back).

Real systems often use Sliding Window (a Fixed Window variant that averages across the boundary) or Leaky Bucket (Token Bucket's cousin). Both fit the same Strategy shape — add them as new strategies without touching RateLimiter.

The state question

Where does per-user state live?

Option A — inside the strategy:

class TokenBucket:
    def __init__(...):
        self._buckets = {}   # user_id -> (tokens, last_time)
    def allow(self, user, now):
        # look up + update self._buckets[user]

Option B — RateLimiter owns the state, strategy is pure:

class RateLimiter:
    def __init__(strategy):
        self._state = {}   # user_id -> whatever the strategy needs
    def allow(user, now):
        self._state[user] = strategy.step(self._state.get(user), now)
        ...

Option A is simpler. Option B is easier to test (strategies are pure functions). Both are defensible. Pin your call.

The rules

• Bucket never exceeds capacity. Enforced by min(capacity, current + elapsed * rate) on every refill.
• Counter never exceeds max within a window. Enforced by the count < max check before incrementing.

Both are one-line enforcements in one place. Name them.

The extension question interviewers love

"How would you add a Sliding Window Log?" (Store timestamps of recent requests; count how many are in the last N seconds. Memory O(requests-in-window), most-accurate.)

If your Strategy interface is clean, this is a new class with zero changes to RateLimiter. That's the OCP payoff.

Real-world callout

• Nginx uses leaky-bucket by default.
• AWS API Gateway uses token-bucket per stage.
• GitHub API uses fixed-window (5000 requests/hour).
• Every SaaS's "requests per minute" tier lives in code like this.

Implement a rate limiter with pluggable algorithms. Strategy signal is central.

The problem

A RateLimiter(strategy) wraps a policy that decides whether to allow(user_id, now).

Two algorithms to implement:

1. Token Bucket

• Each user has a bucket of tokens.
• Bucket has a capacity and a refill_per_sec rate.
• Every allow call: refill tokens based on elapsed time since last observation (capped at capacity), then try to take 1 token. If a token is available, decrement and return True. Else return False.

2. Fixed Window

• Each user has a counter.
• Each allow(user, now):
  • Compute the window this call belongs to: window = now // window_seconds.
  • If it's a new window (differs from stored), reset counter to 0.
  • If counter < max_per_window: increment, return True.
  • Else return False.

Time is caller-provided (seconds, int). No wall clocks.

Requirement-gathering questions

1. Multiple limiters per user? — No, one policy per RateLimiter. Users are keyed by string.
2. What if now goes BACKWARDS? — Treat as no-op (don't add negative time). Not tested aggressively.
3. Per-user vs global? — Per-user for MVP.
4. Thread-safe? — Single-threaded MVP.
5. When does a bucket become "known"? — Lazy: on first allow call for a user. Starts full (Token Bucket) or zeroed (Fixed Window).

Design goals

• Adding a new algorithm (e.g., Sliding Window Log) should be a manageable change.
• Invariants: bucket never exceeds capacity; counter never exceeds max within a window.

Your task

Public API:

from main import RateLimiter, TokenBucket, FixedWindow

tb = TokenBucket(capacity=5, refill_per_sec=1)
rl = RateLimiter(strategy=tb)

for _ in range(5):
    assert rl.allow("alice", now=0) is True
assert rl.allow("alice", now=0) is False   # bucket empty

# After 3 seconds, 3 tokens refilled
assert rl.allow("alice", now=3) is True
assert rl.allow("alice", now=3) is True
assert rl.allow("alice", now=3) is True
assert rl.allow("alice", now=3) is False

fw = FixedWindow(max_per_window=3, window_seconds=10)
rl2 = RateLimiter(strategy=fw)
assert rl2.allow("bob", now=0) is True
assert rl2.allow("bob", now=5) is True
assert rl2.allow("bob", now=9) is True
assert rl2.allow("bob", now=9) is False    # 3 in window [0..9]
assert rl2.allow("bob", now=10) is True    # new window [10..19]

Both TokenBucket and FixedWindow implement a common strategy interface. RateLimiter just delegates.

Click Submit to run the tests.