Token Bucket and Leaky Bucket

Overview

Let's get to the first rate limiting you'll build. Token bucket and leaky bucket are the two classic algorithms most APIs use. Both solve the same problem (limiting requests per unit of time) but with subtly different behaviors that affect the user's experience.

Token bucket allows "bursts" — the user can make 10 quick requests if they didn't use up their quota earlier. It's the model that feels fair: if you didn't use your limit in the last minute, you have a bit "saved up" to use now. APIs like AWS, Stripe, and most cloud gateways use token bucket.

Leaky bucket smooths traffic — it turns irregular bursts into a steady flow. It's like a hose with controlled flow: even if you force water in under pressure, it comes out at a fixed rate. Useful when the downstream system CAN'T handle bursts (e.g., sending notifications to an SMS provider with a fixed per-second quota).

You'll implement both with INCR + EXPIRE — atomic operations Redis guarantees are thread-safe no matter how many concurrent workers you have. And by the end of the capsule you'll understand the famous fixed window bug (the "double window" attack), which is the problem capsule 03's sliding window is going to solve. Without understanding that bug first, you don't appreciate why a sliding window is worth the extra complexity.

Token Bucket: the most-used algorithm

The analogy

Imagine a bucket with a capacity of 10 tokens. A "machine" refills the bucket at a fixed rate — 1 token per second. Each request consumes 1 token from the bucket.

Maximum capacity: 10 tokens
Refill rate:      1 token/second
Cost per request: 1 token

If 5 quick requests arrive → 5 tokens consumed → fine, 5 are left
If 15 quick requests arrive → only 10 go through → the rest are rate limited (429)
If you wait 10 seconds with no requests → the bucket refills to 10 → you can burst 10 again

The key behavior: bursts are allowed up to the bucket's capacity. If the user doesn't use the API, they "save up" tokens to the maximum. When they need them, they can spend them all at once.

Visualizing it over time

Time    Requests  Tokens   Allowed?
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
t=0     -         10        -
t=1s    +5 reqs   5         ✓ (5 went through)
t=2s    -         6         (refill +1)
t=5s    -         9         (refill +3)
t=6s    +12 reqs  9 → 0     ✓ (9 went through, 3 blocked)
t=7s    -         1         (refill +1)

A simple implementation with Redis (INCR + EXPIRE)

We're going to start with the simplest version: counting requests per minute with an atomic counter.

"""
A basic rate limiter with INCR + EXPIRE.
This is the "fixed window" version — useful for understanding the concept,
but it has a bug we'll see at the end.
"""
import time
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)


def is_allowed_basic(user_id: str, limit: int = 10, window: int = 60) -> tuple[bool, int]:
    """
    Rate limiting with an atomic counter.
    Returns: (allowed, remaining)
    """
    key = f"rate:{user_id}"

    current = r.incr(key)

    if current == 1:
        # The first request in this window: set the TTL
        r.expire(key, window)

    remaining = max(0, limit - current)
    allowed = current <= limit

    return (allowed, remaining)


# Test
user_id = "user:42"

# Clean up
r.delete(f"rate:{user_id}")

# The first 10 requests: all allowed
for i in range(10):
    allowed, remaining = is_allowed_basic(user_id)
    print(f"Request {i+1}: allowed={allowed}, remaining={remaining}")

# Request 11: blocked
allowed, remaining = is_allowed_basic(user_id)
print(f"Request 11: allowed={allowed}, remaining={remaining}")

Output:

Request 1: allowed=True, remaining=9
Request 2: allowed=True, remaining=8
...
Request 10: allowed=True, remaining=0
Request 11: allowed=False, remaining=0

It works. But this is NOT a token bucket yet — it's a fixed window counter. The difference matters.

A "real" token bucket with Redis

A true token bucket keeps state about how many tokens there are and when it was last refilled. The implementation:

"""
A token bucket with Redis.
It keeps state: the available tokens + the timestamp of the last refill.
"""
import time
import json
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)


class TokenBucket:
    def __init__(self, capacity=10, refill_rate=1):
        """
        capacity: the maximum number of tokens in the bucket
        refill_rate: tokens per second
        """
        self.capacity = capacity
        self.refill_rate = refill_rate

    def allow(self, key: str) -> tuple[bool, dict]:
        """
        Returns: (allowed, info)
        info includes: tokens_remaining, retry_after_seconds
        """
        bucket_key = f"bucket:{key}"
        now = time.time()

        # Read the current state
        state = r.hgetall(bucket_key)

        if not state:
            # A new bucket: start it full
            tokens = self.capacity - 1  # we consume 1 right now
            r.hset(bucket_key, mapping={
                "tokens": tokens,
                "last_refill": now
            })
            r.expire(bucket_key, int(self.capacity / self.refill_rate) + 60)
            return (True, {"tokens_remaining": tokens, "retry_after_seconds": 0})

        # Calculate the tokens available now (with the refill since last_refill)
        last_refill = float(state["last_refill"])
        elapsed = now - last_refill
        tokens_to_add = elapsed * self.refill_rate
        current_tokens = min(self.capacity, float(state["tokens"]) + tokens_to_add)

        if current_tokens >= 1:
            # Consume 1 token
            new_tokens = current_tokens - 1
            r.hset(bucket_key, mapping={
                "tokens": new_tokens,
                "last_refill": now
            })
            return (True, {
                "tokens_remaining": int(new_tokens),
                "retry_after_seconds": 0
            })
        else:
            # No tokens, calculate when there will be one
            seconds_to_next_token = (1 - current_tokens) / self.refill_rate
            return (False, {
                "tokens_remaining": 0,
                "retry_after_seconds": round(seconds_to_next_token, 2)
            })


if __name__ == "__main__":
    # Demo: a capacity of 10 tokens, a 1 token/sec refill
    bucket = TokenBucket(capacity=10, refill_rate=1)
    user_id = "user:42"

    # Clean up
    r.delete(f"bucket:{user_id}")

    print("=== The initial burst: 12 quick requests ===")
    for i in range(12):
        allowed, info = bucket.allow(user_id)
        status = "✓" if allowed else "✗"
        print(f"Request {i+1}: {status} - tokens={info['tokens_remaining']}, retry_after={info.get('retry_after_seconds')}s")

    print("\n=== Wait 5 seconds (refill) ===")
    time.sleep(5)

    print("\n=== 6 more requests ===")
    for i in range(6):
        allowed, info = bucket.allow(user_id)
        status = "✓" if allowed else "✗"
        print(f"Request {i+13}: {status} - tokens={info['tokens_remaining']}, retry_after={info.get('retry_after_seconds')}s")

Output:

=== The initial burst: 12 quick requests ===
Request 1:  ✓ - tokens=9, retry_after=0s
Request 2:  ✓ - tokens=8, retry_after=0s
Request 3:  ✓ - tokens=7, retry_after=0s
...
Request 10: ✓ - tokens=0, retry_after=0s
Request 11: ✗ - tokens=0, retry_after=1.0s
Request 12: ✗ - tokens=0, retry_after=1.0s

=== Wait 5 seconds (refill) ===

=== 6 more requests ===
Request 13: ✓ - tokens=4, retry_after=0s   (the refill added 5 tokens)
Request 14: ✓ - tokens=3, retry_after=0s
Request 15: ✓ - tokens=2, retry_after=0s
Request 16: ✓ - tokens=1, retry_after=0s
Request 17: ✓ - tokens=0, retry_after=0s
Request 18: ✗ - tokens=0, retry_after=1.0s

The critical part: after 5 seconds with no requests, the bucket refilled with 5 tokens. The next 5 requests went through before being blocked again. This allows bursts — the behavior that separates a token bucket from a simple counter.

When to use token bucket

✅ Public APIs: users expect to be able to make occasional bursts without being blocked ✅ APIs with pricing tiers: a capacity and refill rate per plan (free: 10 tokens, 1/sec; pro: 100 tokens, 10/sec) ✅ Calls to expensive external services: controlling your own usage of APIs like OpenAI or Stripe

❌ Do NOT use it for: downstream systems that CAN'T handle bursts (e.g., an SMS provider with a strict per-second quota)

Leaky Bucket: traffic smoothing

The analogy

A bucket with a hole in the bottom. Requests fall into the bucket. The bucket drains (it releases requests to the downstream system) at a constant rate. If the bucket fills up, new requests are dropped.

Bucket capacity:  10 requests
Drain rate:       1 req/second
If 100 reqs arrive in 1 sec → 10 enter the bucket, 90 are rejected
                            → the bucket drains 1 req/sec to the system
                            → it takes 10 seconds to empty

The difference from a token bucket: the downstream system never receives bursts. It always receives a steady flow.

Implementation

"""
A leaky bucket with Redis.
The implementation: a queue with a fixed drain rate.
"""
import time
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)


class LeakyBucket:
    def __init__(self, capacity=10, leak_rate=1):
        """
        capacity: the maximum number of requests in the bucket
        leak_rate: requests processed per second
        """
        self.capacity = capacity
        self.leak_rate = leak_rate

    def allow(self, key: str) -> tuple[bool, dict]:
        bucket_key = f"leaky:{key}"
        now = time.time()

        state = r.hgetall(bucket_key)

        if not state:
            # A new bucket: 1 request enters
            r.hset(bucket_key, mapping={
                "level": 1,
                "last_leak": now
            })
            r.expire(bucket_key, int(self.capacity / self.leak_rate) + 60)
            return (True, {"queue_size": 1, "retry_after_seconds": 0})

        # Calculate how much "drained" since the last operation
        last_leak = float(state["last_leak"])
        elapsed = now - last_leak
        leaked = elapsed * self.leak_rate
        current_level = max(0, float(state["level"]) - leaked)

        if current_level + 1 <= self.capacity:
            new_level = current_level + 1
            r.hset(bucket_key, mapping={
                "level": new_level,
                "last_leak": now
            })
            return (True, {"queue_size": int(new_level), "retry_after_seconds": 0})
        else:
            # The bucket is full: reject
            seconds_until_space = (1 - (self.capacity - current_level)) / self.leak_rate
            return (False, {
                "queue_size": int(current_level),
                "retry_after_seconds": round(seconds_until_space, 2)
            })


if __name__ == "__main__":
    bucket = LeakyBucket(capacity=10, leak_rate=1)
    user_id = "user:42"
    r.delete(f"leaky:{user_id}")

    print("=== A burst of 15 quick requests ===")
    for i in range(15):
        allowed, info = bucket.allow(user_id)
        status = "✓" if allowed else "✗"
        print(f"Request {i+1}: {status} - queue={info['queue_size']}, retry_after={info.get('retry_after_seconds')}s")

    print("\n=== Wait 3 seconds (a drain of ~3 requests) ===")
    time.sleep(3)

    print("\n=== 5 more requests ===")
    for i in range(5):
        allowed, info = bucket.allow(user_id)
        status = "✓" if allowed else "✗"
        print(f"Request {i+16}: {status} - queue={info['queue_size']}")

Output:

=== A burst of 15 quick requests ===
Request 1:  ✓ - queue=1
Request 2:  ✓ - queue=2
...
Request 10: ✓ - queue=10
Request 11: ✗ - queue=10, retry_after=1.0s
Request 12: ✗ - queue=10
Request 13: ✗ - queue=10
Request 14: ✗ - queue=10
Request 15: ✗ - queue=10

=== Wait 3 seconds (a drain of ~3 requests) ===

=== 5 more requests ===
Request 16: ✓ - queue=8
Request 17: ✓ - queue=9
Request 18: ✓ - queue=10
Request 19: ✗ - queue=10
Request 20: ✗ - queue=10

The critical difference from a token bucket:

• Token bucket: 12 quick requests → 10 go through, 2 fail, then it recovers (5 sec later you can make 5 more together because the bucket refilled)
• Leaky bucket: 15 quick requests → 10 enter the bucket, 5 are rejected immediately. The bucket drains slowly, it doesn't recover in a burst

Token bucket vs Leaky bucket: the decision

Which one to choose?

Question 1: Can the downstream system handle bursts?

• Yes → Token bucket
• No → Leaky bucket

Question 2: Does UX matter more than protection?

• Yes (a public API, developers as users) → Token bucket
• No (an internal job processor) → Leaky bucket

Question 3: Do you have user tiers with quotas?

• Yes → Token bucket (a capacity + refill_rate per tier is elegant)
• No → Either one

The practical recommendation: For 90% of public APIs, use token bucket. Only switch to leaky bucket if you have a specific reason (downstream limitations).

The Fixed Window Counter bug

Let's go back to the first implementation you saw — is_allowed_basic with INCR + EXPIRE. Even though it apparently works, it has a serious bug that shows up under concurrency. Let's look at the attack.

The problem

Assume a rate limit of 100 requests per minute. The counter resets every natural minute (00:00, 00:01, 00:02...).

Time              The user's requests
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
10:00:50 - 10:00:59  100 requests (they all go through, the counter reaches 100)
10:01:00              the counter resets to 0 (a new window)
10:01:00 - 10:01:09  100 more requests (they all go through)

In a span of 20 seconds (10:00:50 to 10:01:10), the user made 200 requests, not 100.

The "100/min" rate limit broke because the attacker exploited the exact moment of the reset. That's the fixed window bug.

Visualizing the attack

Window 1 (10:00:00-10:00:59)   Window 2 (10:01:00-10:01:59)
[                              ][                              ]
                          100 reqs                100 reqs
                          ████████                ████████
                          10:00:50                10:01:00
                                  ↑↑↑
                          200 reqs in 10 seconds!

Why it matters in production

• Adversarial bots can time their requests with the reset
• Load spikes at the start of a window overload the downstream system
• Pricing fraud: a free user could make 2x their limit with the right timing

The solution: a Sliding Window (a preview)

Instead of "a fixed 1-minute window that resets every natural minute," we use "a moving 60-second window from RIGHT NOW":

Now = 10:01:10
Window = the last 60 seconds (10:00:10 to 10:01:10)

How many requests in that window? → ZRANGEBYSCORE rate:user 10:00:10 10:01:10
If > 100 → block

The window "moves" with time. The fixed window bug disappears.

This is what you're going to implement in capsule 03 with sorted sets.

A real case: rate limiting by IP in FastAPI

Let's integrate a token bucket into a FastAPI API with middleware. Each IP has its own rate limit.

"""
A FastAPI app with rate limiting by IP using a token bucket.
"""
from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import time
import redis


r = redis.Redis(host='localhost', port=6379, decode_responses=True)


class TokenBucketRateLimiter:
    def __init__(self, capacity=60, refill_rate=1):
        self.capacity = capacity
        self.refill_rate = refill_rate

    def check(self, key: str) -> tuple[bool, dict]:
        bucket_key = f"bucket:{key}"
        now = time.time()
        state = r.hgetall(bucket_key)

        if not state:
            r.hset(bucket_key, mapping={"tokens": self.capacity - 1, "last_refill": now})
            r.expire(bucket_key, 600)
            return (True, {
                "tokens_remaining": self.capacity - 1,
                "limit": self.capacity,
                "retry_after": 0,
                "reset_in": int(self.capacity / self.refill_rate),
            })

        last_refill = float(state["last_refill"])
        elapsed = now - last_refill
        tokens_to_add = elapsed * self.refill_rate
        current_tokens = min(self.capacity, float(state["tokens"]) + tokens_to_add)

        if current_tokens >= 1:
            new_tokens = current_tokens - 1
            r.hset(bucket_key, mapping={"tokens": new_tokens, "last_refill": now})
            return (True, {
                "tokens_remaining": int(new_tokens),
                "limit": self.capacity,
                "retry_after": 0,
                "reset_in": int((self.capacity - new_tokens) / self.refill_rate),
            })
        else:
            return (False, {
                "tokens_remaining": 0,
                "limit": self.capacity,
                "retry_after": round((1 - current_tokens) / self.refill_rate, 2),
                "reset_in": int(self.capacity / self.refill_rate),
            })


limiter = TokenBucketRateLimiter(capacity=60, refill_rate=1)


class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Get the client's IP
        client_ip = request.client.host if request.client else "unknown"

        # Skip rate limiting on health checks
        if request.url.path == "/health":
            return await call_next(request)

        allowed, info = limiter.check(f"ip:{client_ip}")

        if not allowed:
            return JSONResponse(
                status_code=429,
                content={
                    "error": "Too many requests",
                    "retry_after_seconds": info["retry_after"]
                },
                headers={
                    "X-RateLimit-Limit": str(info["limit"]),
                    "X-RateLimit-Remaining": "0",
                    "X-RateLimit-Reset": str(int(time.time() + info["reset_in"])),
                    "Retry-After": str(int(info["retry_after"])),
                }
            )

        response = await call_next(request)
        response.headers["X-RateLimit-Limit"] = str(info["limit"])
        response.headers["X-RateLimit-Remaining"] = str(info["tokens_remaining"])
        response.headers["X-RateLimit-Reset"] = str(int(time.time() + info["reset_in"]))
        return response


app = FastAPI(title="Rate Limited API")
app.add_middleware(RateLimitMiddleware)


@app.get("/health")
def health():
    return {"status": "ok"}


@app.get("/products")
def list_products():
    return {"products": [f"product-{i}" for i in range(10)]}


@app.get("/products/{product_id}")
def get_product(product_id: int):
    return {"id": product_id, "name": f"Product {product_id}"}

Testing the rate limiter

uvicorn app:app --reload

# Test 1: a normal request
curl -i http://localhost:8000/products | head -10
# HTTP/1.1 200 OK
# X-RateLimit-Limit: 60
# X-RateLimit-Remaining: 59
# X-RateLimit-Reset: 1714069262

# Test 2: 70 quick requests
for i in {1..70}; do
  echo "Request $i: $(curl -s -o /dev/null -w "%{http_code}" http://localhost:8000/products)"
done

# Expected output:
# Request 1: 200
# Request 2: 200
# ...
# Request 60: 200
# Request 61: 429
# Request 62: 429
# ...

The headers in the 429 response:

curl -i http://localhost:8000/products | head -10
# HTTP/1.1 429 Too Many Requests
# X-RateLimit-Limit: 60
# X-RateLimit-Remaining: 0
# X-RateLimit-Reset: 1714069262
# Retry-After: 1

This is professional rate limiting:

• HTTP 429 (Too Many Requests) — the standard code
• X-RateLimit-* headers — clients can adapt their behavior
• Retry-After — clients know when to retry

Troubleshooting

Problem 1: The counter "skips" values under concurrency

Cause: This can be surprising at first — but INCR is atomic. There's no race condition with counters.

Solution: If you see strange behavior, it's probably because you have multiple uvicorn workers each applying the rate limit on their own. Make sure they ALL point to the same Redis.

Problem 2: The TTL isn't reset in each window

Cause: You're using SET ... EX every time. If the counter is already at 50 and you add another request, the SET resets the TTL back to the maximum.

Solution: Use INCR (which does NOT touch the TTL) and only set the TTL on the FIRST request:

current = r.incr(key)
if current == 1:
    r.expire(key, window)  # only here

Problem 3: The token bucket "forgets" tokens between requests

Cause: The key's TTL is too short. If more seconds go by than the TTL, the bucket's data is deleted.

Solution: The bucket's TTL should be > the time it takes to fill + a margin:

r.expire(bucket_key, int(self.capacity / self.refill_rate) + 60)

For capacity=60, refill_rate=1: TTL = 120 seconds. Plenty of margin.

Problem 4: Rate limiting by IP is bypassable with proxies

Cause: The attacker uses multiple IPs.

Solution: An additional layer of rate limiting:

1. Rate limit by IP: the first filter (this module)
2. Rate limit by user_id: after auth (stricter)
3. A WAF at the CDN level: Cloudflare, AWS WAF (out of scope)
4. A CAPTCHA after N suspicious requests

Combine layers — none of them is perfect on its own.

Problem 5: The headers don't show up on successful responses

Cause: The middleware only modifies the response in some cases.

Solution: Add the headers to ALL the responses (200 and 429):

async def dispatch(self, request, call_next):
    allowed, info = limiter.check(...)

    if not allowed:
        return JSONResponse(status_code=429, headers={...})

    response = await call_next(request)
    # ✨ Headers on the 200 as well
    response.headers["X-RateLimit-Limit"] = str(info["limit"])
    response.headers["X-RateLimit-Remaining"] = str(info["tokens_remaining"])
    return response

Problem 6: The fixed window bug still happens with my token bucket

Cause: If your "token bucket" is really INCR + EXPIRE, then yes, it has the bug. A REAL token bucket stores a timestamp + tokens; it doesn't reset per window.

Solution: Use the HSET implementation we saw above. Or better, skip ahead to the sliding window (capsule 03) — simpler and bug-free.

Exercises

Exercise 1: A basic counter (Easy)

Implement is_allowed(user_id, limit=10, window=60) with INCR + EXPIRE. Test: 12 quick requests, the first 10 allowed, the last 2 blocked.

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)


def is_allowed(user_id, limit=10, window=60):
    key = f"rate:{user_id}"
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)
    return current <= limit, max(0, limit - current)


# Test
r.delete("rate:user42")

for i in range(12):
    allowed, remaining = is_allowed("user42")
    status = "✓" if allowed else "✗"
    print(f"Request {i+1}: {status} (remaining={remaining})")

Request 1: ✓ (remaining=9)
Request 2: ✓ (remaining=8)
...
Request 10: ✓ (remaining=0)
Request 11: ✗ (remaining=0)
Request 12: ✗ (remaining=0)

Exercise 2: A complete token bucket (Medium)

Implement TokenBucket with capacity=5 and refill_rate=2 (2 tokens/sec). Make 10 quick requests, wait 3 seconds, make 8 more. Report which ones went through and which didn't.

import time
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)


class TokenBucket:
    def __init__(self, capacity, refill_rate):
        self.capacity = capacity
        self.refill_rate = refill_rate

    def allow(self, key):
        bucket_key = f"bucket:{key}"
        now = time.time()
        state = r.hgetall(bucket_key)

        if not state:
            r.hset(bucket_key, mapping={"tokens": self.capacity - 1, "last_refill": now})
            r.expire(bucket_key, 600)
            return True, self.capacity - 1

        last_refill = float(state["last_refill"])
        elapsed = now - last_refill
        current = min(self.capacity, float(state["tokens"]) + elapsed * self.refill_rate)

        if current >= 1:
            new_tokens = current - 1
            r.hset(bucket_key, mapping={"tokens": new_tokens, "last_refill": now})
            return True, int(new_tokens)
        return False, 0


bucket = TokenBucket(capacity=5, refill_rate=2)
r.delete("bucket:test")

print("=== A burst of 10 ===")
for i in range(10):
    allowed, tokens = bucket.allow("test")
    status = "✓" if allowed else "✗"
    print(f"Req {i+1}: {status} (tokens={tokens})")

print("\n=== Wait 3 sec (expected refill: 6 tokens, capped at 5) ===")
time.sleep(3)

print("\n=== 8 more requests ===")
for i in range(8):
    allowed, tokens = bucket.allow("test")
    status = "✓" if allowed else "✗"
    print(f"Req {i+11}: {status} (tokens={tokens})")

=== A burst of 10 ===
Req 1: ✓ (tokens=4)
Req 2: ✓ (tokens=3)
Req 3: ✓ (tokens=2)
Req 4: ✓ (tokens=1)
Req 5: ✓ (tokens=0)
Req 6: ✗ (tokens=0)
Req 7: ✗ (tokens=0)
Req 8: ✗ (tokens=0)
Req 9: ✗ (tokens=0)
Req 10: ✗ (tokens=0)

=== Wait 3 sec ===

=== 8 more requests ===
Req 11: ✓ (tokens=4)   # 5 tokens available (capped at capacity)
Req 12: ✓ (tokens=3)
Req 13: ✓ (tokens=2)
Req 14: ✓ (tokens=1)
Req 15: ✓ (tokens=0)
Req 16: ✗ (tokens=0)
Req 17: ✗ (tokens=0)
Req 18: ✗ (tokens=0)

Exercise 3: A leaky bucket vs token bucket comparison (Medium)

The same workload (15 requests in 1 second, wait 5 sec, 5 more requests): compare token bucket vs leaky bucket. Which one allowed more requests? Explain why.

# Test both
import time

def benchmark(bucket, name, key):
    r.delete(f"bucket:{key}")
    r.delete(f"leaky:{key}")

    print(f"\n=== {name} ===")
    allowed_count = 0

    # The initial burst
    for i in range(15):
        allowed, _ = bucket.allow(key)
        if allowed:
            allowed_count += 1

    # Wait
    time.sleep(5)

    # 5 more
    for i in range(5):
        allowed, _ = bucket.allow(key)
        if allowed:
            allowed_count += 1

    print(f"Total allowed: {allowed_count}/20")


benchmark(TokenBucket(10, 1), "Token Bucket", "tb_test")
benchmark(LeakyBucket(10, 1), "Leaky Bucket", "lb_test")

=== Token Bucket ===
Total allowed: 15/20
# 10 from the initial burst + 5 after the 5 seconds (refill)

=== Leaky Bucket ===
Total allowed: 15/20
# 10 from the initial burst + 5 afterwards (it drained 5)

Exercise 4: Demonstrating the fixed window bug (Hard)

Implement the "double window" attack: with an INCR + EXPIRE counter of 60s and a limit of 100, show that you can make 200 requests within a real 60-second window by timing it with the reset.

import time
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)


def is_allowed_fixed(user_id, limit=100, window=60):
    key = f"rate:{user_id}"
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)
    return current <= limit


# Setup: simulate the attack
r.delete("rate:attacker")

# For a real simulation, you'd have to wait for the window's natural boundary.
# Here we simulate it manually:

# Step 1: make 100 requests at the end of window 1 (they all go through)
print("Filling window 1 with 100 requests...")
for i in range(100):
    is_allowed_fixed("attacker")
print(f"The counter is now: {r.get('rate:attacker')}")  # 100

# Step 2: wait for the TTL to expire (simulating the exact moment of the reset)
ttl_remaining = r.ttl("rate:attacker")
print(f"TTL remaining: {ttl_remaining}s — waiting for it to expire...")
time.sleep(ttl_remaining + 1)

# Step 3: now we're in window 2, we can make 100 more
print("\nWindow 2 (the counter was reset):")
for i in range(100):
    allowed = is_allowed_fixed("attacker")
print(f"The counter is now: {r.get('rate:attacker')}")  # 100

# RESULT: 200 requests within a real time window of ~10 seconds
# (5 sec at the end of window 1 + 5 sec at the start of window 2)
print("\n⚠️ The user made 200 requests but the limit was 100/min!")

Exercise 5: Rate limiting per user tier (Medium)

Implement rate limiting with 3 tiers: free (10 req/min), pro (100 req/min), enterprise (1000 req/min). Use a token bucket. Read the user's tier from a simulated dict.

import time
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)


# We simulate a "users table"
user_tiers = {
    "user:free_alice": "free",
    "user:pro_bob": "pro",
    "user:enterprise_carol": "enterprise",
}


TIER_CONFIG = {
    "free":       {"capacity": 10,   "refill_rate": 10/60},    # 10/min
    "pro":        {"capacity": 100,  "refill_rate": 100/60},   # 100/min
    "enterprise": {"capacity": 1000, "refill_rate": 1000/60},  # 1000/min
}


class TieredTokenBucket:
    def allow(self, user_id):
        tier = user_tiers.get(user_id, "free")
        config = TIER_CONFIG[tier]

        bucket_key = f"bucket:{user_id}"
        now = time.time()
        state = r.hgetall(bucket_key)

        if not state:
            r.hset(bucket_key, mapping={"tokens": config["capacity"] - 1, "last_refill": now})
            r.expire(bucket_key, 600)
            return True, {"tier": tier, "tokens": config["capacity"] - 1, "limit": config["capacity"]}

        last_refill = float(state["last_refill"])
        elapsed = now - last_refill
        current = min(config["capacity"], float(state["tokens"]) + elapsed * config["refill_rate"])

        if current >= 1:
            new_tokens = current - 1
            r.hset(bucket_key, mapping={"tokens": new_tokens, "last_refill": now})
            return True, {"tier": tier, "tokens": int(new_tokens), "limit": config["capacity"]}
        return False, {"tier": tier, "tokens": 0, "limit": config["capacity"]}


limiter = TieredTokenBucket()

# Test
for user in ["user:free_alice", "user:pro_bob", "user:enterprise_carol"]:
    r.delete(f"bucket:{user}")
    print(f"\n=== {user} (tier: {user_tiers.get(user)}) ===")

    # Make N+1 requests where N is their capacity
    config = TIER_CONFIG[user_tiers[user]]
    n = config["capacity"]

    for i in range(n + 2):
        allowed, info = limiter.allow(user)
        status = "✓" if allowed else "✗"
        if i < 3 or i > n - 2:  # only print the first 3 and the last 3
            print(f"  Req {i+1}: {status} (tokens={info['tokens']}, limit={info['limit']})")
        elif i == 3:
            print("  ...")

=== user:free_alice (tier: free) ===
  Req 1: ✓ (tokens=9, limit=10)
  Req 2: ✓ (tokens=8, limit=10)
  Req 3: ✓ (tokens=7, limit=10)
  ...
  Req 10: ✓ (tokens=0, limit=10)
  Req 11: ✗ (tokens=0, limit=10)
  Req 12: ✗ (tokens=0, limit=10)

=== user:pro_bob (tier: pro) ===
  Req 1: ✓ (tokens=99, limit=100)
  ...
  Req 100: ✓ (tokens=0, limit=100)
  Req 101: ✗ (tokens=0, limit=100)

=== user:enterprise_carol (tier: enterprise) ===
  Req 1: ✓ (tokens=999, limit=1000)
  ...

Exercise 6: FastAPI middleware with correct headers (Medium-Hard)

Integrate the TokenBucket into a FastAPI middleware. Make sure to include the X-RateLimit-* headers on ALL the responses (200 and 429). Test with curl that the headers show up.

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import time, redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)


class TokenBucket:
    def __init__(self, capacity=10, refill_rate=1/6):  # 10/min
        self.capacity = capacity
        self.refill_rate = refill_rate

    def check(self, key):
        bucket_key = f"bucket:{key}"
        now = time.time()
        state = r.hgetall(bucket_key)

        if not state:
            r.hset(bucket_key, mapping={"tokens": self.capacity - 1, "last_refill": now})
            r.expire(bucket_key, 600)
            return True, self.capacity - 1

        elapsed = now - float(state["last_refill"])
        current = min(self.capacity, float(state["tokens"]) + elapsed * self.refill_rate)

        if current >= 1:
            r.hset(bucket_key, mapping={"tokens": current - 1, "last_refill": now})
            return True, int(current - 1)
        return False, 0


limiter = TokenBucket(capacity=10, refill_rate=10/60)


class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        ip = request.client.host if request.client else "unknown"
        allowed, tokens = limiter.check(f"ip:{ip}")

        headers = {
            "X-RateLimit-Limit": "10",
            "X-RateLimit-Remaining": str(tokens),
            "X-RateLimit-Reset": str(int(time.time() + 60)),
        }

        if not allowed:
            return JSONResponse(
                status_code=429,
                content={"error": "Too many requests"},
                headers={**headers, "Retry-After": "6"}
            )

        response = await call_next(request)
        for k, v in headers.items():
            response.headers[k] = v
        return response


app = FastAPI()
app.add_middleware(RateLimitMiddleware)


@app.get("/products")
def products():
    return {"products": []}

uvicorn app:app --reload &
curl -i http://localhost:8000/products | head -8
# HTTP/1.1 200 OK
# X-RateLimit-Limit: 10
# X-RateLimit-Remaining: 9
# X-RateLimit-Reset: 1714069262

# 11 requests
for i in {1..11}; do
  curl -s -o /dev/null -w "Status: %{http_code}, Remaining: %{header_x-ratelimit-remaining}\n" \
    http://localhost:8000/products
done

Status: 200, Remaining: 9
Status: 200, Remaining: 8
...
Status: 200, Remaining: 0
Status: 429, Remaining: 0

Summary

In this capsule you learned:

• Token bucket: a bucket with N tokens, refilled at a fixed rate, which allows bursts up to capacity
• Leaky bucket: a bucket that drains at a constant rate, smoothing traffic, with no bursts
• The decision: token bucket for public APIs (UX friendly), leaky bucket for downstream systems with a strict quota
• The implementation: HSET with tokens + a last_refill timestamp, computing the refill on the fly
• The fixed window bug: an INCR + EXPIRE counter lets you double the limit by timing the reset
• A "real" token bucket vs a naive counter: persistent state vs a counter with a TTL
• Integration with FastAPI: middleware that applies the rate limit + adds X-RateLimit-* headers
• Standard HTTP headers: X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset, Retry-After, and the 429 Too Many Requests status

The critical part for capsule 03: you now understand the fixed window bug. A sliding window with sorted sets solves it, and it's the professional algorithm. Without having seen the bug first, you wouldn't appreciate why a sliding window is worth the extra complexity.

Additional resources

1. Token Bucket on Wikipedia — The algorithm's formal definition
2. Leaky Bucket on Wikipedia — Variants of the algorithm
3. Stripe API Rate Limiting — How Stripe implements rate limiting (it uses a token bucket with Redis)
4. GitHub API Rate Limiting — Standard headers and behavior
5. Redis: Rate Limiting Patterns — The official patterns with redis-py
6. System Design: Rate Limiter — A deep analysis of the algorithms

What's next?

In Capsule 03 you get into Sliding Window with sorted sets — the professional rate limiting algorithm. You'll take ZADD, ZRANGEBYSCORE, and ZREMRANGEBYSCORE from module 1 and build the rate limiter that does NOT have the fixed window bug. You'll also cover granular rate limiting (by IP, by user, by endpoint) and advanced HTTP headers.

Keep your workspace and Redis running. Let's go.