Sliding Window with Sorted Sets

Overview

In the previous capsule you saw the fixed window bug: INCR + EXPIRE lets you double the rate limit by timing it with the window's reset. A token bucket with persistent state avoids it, but it isn't the most-used algorithm in serious APIs. Sliding window is the professional pattern — the one APIs like Cloudflare, GitHub, and most cloud gateways use when they need reliable rate limiting.

The idea is elegant: instead of "a fixed 1-minute window that resets every natural minute," we use "a moving 60-second window from RIGHT NOW." Every request is recorded with its timestamp as the score in a sorted set. To check the limit, we count how many requests are in [now - 60s, now]. To clean up, we remove the oldest requests with ZREMRANGEBYSCORE. The window literally moves with time.

This is the moment where module 1's sorted sets take on their purpose. ZADD to record, ZRANGEBYSCORE or ZCARD to count, ZREMRANGEBYSCORE to clean up. Three commands make up a bug-free, mathematically precise rate limiter. You'll learn to integrate it with FastAPI middleware with advanced HTTP headers (X-RateLimit-*, Retry-After), apply it granularly (by IP, by user, by endpoint), and test it with concurrent requests to verify the limit holds under real load.

Why a sliding window fixes the bug

A recap of the problem (capsule 02):

Fixed window (an INCR + EXPIRE counter):

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!

Sliding window (with a sorted set):

At 10:01:05, the moving window = [10:00:05, 10:01:05]

If there were 100 requests between 10:00:50 and 10:00:59,
and 50 requests between 10:01:00 and 10:01:05,
then the current window = 100 + 50 = 150 reqs
                                       └── limit exceeded!

The window moves. There's no reset to exploit. Every request is measured against the last 60 seconds, whatever the exact moment.

The algorithm

Pseudocode

function is_allowed(user_id, limit=100, window_seconds=60):
    now = current_timestamp()
    window_start = now - window_seconds
    key = f"rate:{user_id}"

    # 1. Clean out requests older than the window
    redis.zremrangebyscore(key, 0, window_start)

    # 2. Count the requests in the current window
    current_count = redis.zcard(key)

    # 3. If it's under the limit, record the new request
    if current_count < limit:
        redis.zadd(key, {f"req:{now}:{uuid}": now})
        redis.expire(key, window_seconds)  # safety TTL
        return True

    return False

3 commands to Redis: ZREMRANGEBYSCORE, ZCARD, ZADD (+ EXPIRE). About ~3 ms total. Atomic with a pipeline.

Implementation with redis-py

"""
A sliding window rate limiter with sorted sets.
"""
import time
import uuid
import redis

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


class SlidingWindowRateLimiter:
    def __init__(self, limit=100, window_seconds=60):
        self.limit = limit
        self.window_seconds = window_seconds

    def check(self, user_id: str) -> tuple[bool, dict]:
        """
        ⚠️ AN INITIAL VERSION WITH A SUBTLE BUG — just to understand the pattern.
        The rollback generates a new UUID instead of reusing the one from the ZADD,
        so it does NOT delete the member you just inserted.
        Use `check_v2` (below) in real code.

        Returns: (allowed, info)
        info: {limit, remaining, reset_in_seconds, retry_after_seconds}
        """
        key = f"rate:{user_id}"
        now = time.time()
        window_start = now - self.window_seconds

        # A pipeline for atomicity
        pipe = r.pipeline()
        pipe.zremrangebyscore(key, 0, window_start)  # cleanup
        pipe.zcard(key)                                # count
        pipe.zadd(key, {f"req:{now}:{uuid.uuid4().hex}": now})  # add
        pipe.expire(key, self.window_seconds + 60)     # safety TTL
        _, current_count, _, _ = pipe.execute()

        # current_count is AFTER the remove but BEFORE the add
        # If it was < limit before, there's room for the new one
        allowed = current_count < self.limit

        if not allowed:
            # ⚠️ BUG: this UUID is NOT the same one from the ZADD, so the ZREM deletes nothing.
            # I'm leaving it here on purpose so you see why `check_v2` captures `member` in a variable.
            r.zrem(key, f"req:{now}:{uuid.uuid4().hex}")
            # Calculate when there will be room
            oldest = r.zrange(key, 0, 0, withscores=True)
            if oldest:
                oldest_score = oldest[0][1]
                retry_after = max(0, oldest_score + self.window_seconds - now)
            else:
                retry_after = self.window_seconds
        else:
            retry_after = 0

        remaining = max(0, self.limit - current_count - 1)

        return (allowed, {
            "limit": self.limit,
            "remaining": remaining,
            "reset_in_seconds": int(self.window_seconds),
            "retry_after_seconds": round(retry_after, 2),
        })


if __name__ == "__main__":
    limiter = SlidingWindowRateLimiter(limit=10, window_seconds=10)
    user_id = "user:42"
    r.delete(f"rate:{user_id}")

    print("=== A burst of 12 requests ===")
    for i in range(12):
        allowed, info = limiter.check(user_id)
        status = "✓" if allowed else "✗"
        print(f"Req {i+1}: {status} (remaining={info['remaining']}, retry_after={info['retry_after_seconds']}s)")

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

    print("\n=== 5 more requests ===")
    for i in range(5):
        allowed, info = limiter.check(user_id)
        status = "✓" if allowed else "✗"
        print(f"Req {i+13}: {status} (remaining={info['remaining']}, retry_after={info['retry_after_seconds']}s)")

Expected output:

=== A burst of 12 requests ===
Req 1:  ✓ (remaining=9, retry_after=0s)
Req 2:  ✓ (remaining=8, retry_after=0s)
...
Req 10: ✓ (remaining=0, retry_after=0s)
Req 11: ✗ (remaining=0, retry_after=10.0s)
Req 12: ✗ (remaining=0, retry_after=10.0s)

=== Wait 5 seconds ===

=== 5 more requests ===
Req 13: ✗ (remaining=0, retry_after=5.0s)   # the older ones still count
Req 14: ✗ (remaining=0, retry_after=5.0s)
Req 15: ✗ (remaining=0, retry_after=5.0s)

The critical part: after 5 seconds, the old requests (from 5 sec ago) have NOT been cleaned out yet — only the ones older than 10 sec. A sliding window literally honors "10 requests in any 10-second window."

How it looks in Redis

While the rate limiter runs, you can inspect the sorted set:

127.0.0.1:6379> ZRANGE rate:user:42 0 -1 WITHSCORES
1) "req:1714069200.123:a1b2c3"
2) "1714069200.123"
3) "req:1714069200.456:d4e5f6"
4) "1714069200.456"
...

127.0.0.1:6379> ZCARD rate:user:42
(integer) 10

Each request is a unique member (with a UUID), and its score is the Unix timestamp. Visually, you can see exactly when each request was made.

An implementation with a properly atomic pipeline

There's a subtle detail: in my implementation above, the ZADD and the rollback aren't fully atomic. A better approach is to use the "ZADD-then-check" pattern:

def check_v2(self, user_id: str) -> tuple[bool, dict]:
    """A cleaner implementation."""
    key = f"rate:{user_id}"
    now = time.time()
    window_start = now - self.window_seconds
    member = f"req:{now}:{uuid.uuid4().hex}"

    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, window_start)
    pipe.zadd(key, {member: now})
    pipe.zcard(key)
    pipe.expire(key, self.window_seconds + 60)
    _, _, count_after, _ = pipe.execute()

    # count_after includes the request we just added
    if count_after > self.limit:
        # It exceeds the limit: roll back
        r.zrem(key, member)
        return False, {"limit": self.limit, "remaining": 0, ...}

    return True, {"limit": self.limit, "remaining": self.limit - count_after, ...}

Why this one is better: the "allowed" decision is based on the state AFTER the add, so there's no race condition between two concurrent workers.

Granular rate limiting: 3 dimensions

In production, you'll want to apply rate limits at different levels depending on the context. There are 3 common dimensions:

1. By IP (anti-DDoS, no auth)

def rate_limit_by_ip(request_ip: str):
    return limiter.check(f"ip:{request_ip}")

Useful for:

• Public endpoints with no authentication (login, signup)
• General DDoS protection
• Bots and scrapers

The limitation: bypassable with multiple IPs (proxies, VPNs).

2. By authenticated user (fair use)

def rate_limit_by_user(user_id: str):
    return limiter.check(f"user:{user_id}")

Useful for:

• Applying quotas per plan (free/pro/enterprise)
• Detecting abuse from individual accounts
• Usage tracking for billing

Stricter than IP-based because the user_id is unique and identifiable.

3. By specific endpoint (protecting the expensive ones)

def rate_limit_by_endpoint(user_id: str, endpoint: str):
    return limiter.check(f"endpoint:{endpoint}:user:{user_id}")

Useful when some endpoints are MUCH more expensive than others:

RATE_LIMITS = {
    "GET:/products": (100, 60),       # 100 req/min
    "GET:/search": (20, 60),          # 20 req/min (search is expensive)
    "POST:/orders": (10, 60),         # 10 req/min (critical writes)
    "POST:/reports/generate": (5, 3600),  # 5 per hour (very expensive)
}

def get_limit_for_endpoint(method, path, user_id):
    key = f"{method}:{path}"
    if key in RATE_LIMITS:
        limit, window = RATE_LIMITS[key]
        return SlidingWindowRateLimiter(limit, window).check(f"{key}:user:{user_id}")
    # Default
    return default_limiter.check(f"user:{user_id}")

Combining them: layered rate limiting

The 3 dimensions aren't mutually exclusive — combine them:

def check_all_limits(request_ip, user_id, endpoint):
    # 1. The global layer by IP (anti-DDoS)
    allowed_ip, info_ip = ip_limiter.check(f"ip:{request_ip}")
    if not allowed_ip:
        return False, info_ip, "IP rate limit exceeded"

    # 2. The per-user layer (fair use)
    allowed_user, info_user = user_limiter.check(f"user:{user_id}")
    if not allowed_user:
        return False, info_user, "User rate limit exceeded"

    # 3. The per-endpoint layer (protecting expensive endpoints)
    if endpoint in EXPENSIVE_ENDPOINTS:
        allowed_ep, info_ep = endpoint_limiter.check(f"ep:{endpoint}:user:{user_id}")
        if not allowed_ep:
            return False, info_ep, "Endpoint rate limit exceeded"

    return True, info_user, "OK"

Layer by layer: if one fails, the request is blocked. This is defense-in-depth.

Advanced HTTP headers

Professional APIs tell the client about the rate limit's state through headers. This allows:

• Client-side self-throttling (it reduces its requests before hitting the limit)
• Intelligent backoff when it gets a 429
• Debugging and monitoring on the client's side

The standard headers

The status code

• 429 Too Many Requests when the limit is exceeded
• Include a body with the details:

{
  "error": "rate_limit_exceeded",
  "message": "Too many requests. Retry after 30 seconds.",
  "retry_after_seconds": 30
}

A complete implementation with FastAPI middleware

"""
FastAPI middleware with a sliding window + professional headers.
"""
import time
import uuid
from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import redis


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


class SlidingWindowLimiter:
    def __init__(self, limit=100, window=60):
        self.limit = limit
        self.window = window

    def check(self, key: str) -> tuple[bool, int, float]:
        """Returns: (allowed, remaining, retry_after_seconds)"""
        full_key = f"rate:{key}"
        now = time.time()
        window_start = now - self.window
        member = f"req:{now}:{uuid.uuid4().hex}"

        pipe = r.pipeline()
        pipe.zremrangebyscore(full_key, 0, window_start)
        pipe.zadd(full_key, {member: now})
        pipe.zcard(full_key)
        pipe.expire(full_key, self.window + 60)
        _, _, count, _ = pipe.execute()

        if count > self.limit:
            # Rollback
            r.zrem(full_key, member)

            # Calculate retry_after from the oldest entry
            oldest = r.zrange(full_key, 0, 0, withscores=True)
            if oldest:
                oldest_score = oldest[0][1]
                retry_after = max(0, oldest_score + self.window - now)
            else:
                retry_after = self.window
            return (False, 0, round(retry_after, 2))

        remaining = self.limit - count
        return (True, remaining, 0)


# Limiters per dimension
ip_limiter = SlidingWindowLimiter(limit=600, window=60)        # 600 reqs/min per IP
user_limiter = SlidingWindowLimiter(limit=100, window=60)      # 100 reqs/min per user
expensive_limiter = SlidingWindowLimiter(limit=10, window=60)  # 10 reqs/min for expensive endpoints


EXPENSIVE_ENDPOINTS = {"/search", "/reports", "/export"}


class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Skip rate limiting on health checks
        if request.url.path in ("/health", "/metrics"):
            return await call_next(request)

        # 1. The IP-based limit
        client_ip = request.client.host if request.client else "unknown"
        ip_allowed, ip_remaining, ip_retry = ip_limiter.check(f"ip:{client_ip}")
        if not ip_allowed:
            return self._rate_limit_response(
                limit=ip_limiter.limit,
                remaining=0,
                retry_after=ip_retry,
                reason="IP rate limit"
            )

        # 2. The user-based limit (if there's an authenticated user)
        # In production, extract the user_id from the JWT
        user_id = request.headers.get("X-User-Id", "anonymous")
        user_allowed, user_remaining, user_retry = user_limiter.check(f"user:{user_id}")
        if not user_allowed:
            return self._rate_limit_response(
                limit=user_limiter.limit,
                remaining=0,
                retry_after=user_retry,
                reason="User rate limit"
            )

        # 3. The endpoint-based limit (for expensive endpoints only)
        if request.url.path in EXPENSIVE_ENDPOINTS:
            ep_key = f"endpoint:{request.url.path}:user:{user_id}"
            ep_allowed, ep_remaining, ep_retry = expensive_limiter.check(ep_key)
            if not ep_allowed:
                return self._rate_limit_response(
                    limit=expensive_limiter.limit,
                    remaining=0,
                    retry_after=ep_retry,
                    reason=f"Endpoint {request.url.path} rate limit"
                )

        # All the checks passed — proceed with the request
        response = await call_next(request)

        # Add the informational headers (use the strictest limit that applies)
        response.headers["X-RateLimit-Limit"] = str(user_limiter.limit)
        response.headers["X-RateLimit-Remaining"] = str(user_remaining)
        response.headers["X-RateLimit-Reset"] = str(int(time.time() + user_limiter.window))
        return response

    def _rate_limit_response(self, limit, remaining, retry_after, reason):
        return JSONResponse(
            status_code=429,
            content={
                "error": "rate_limit_exceeded",
                "message": f"{reason}. Retry after {int(retry_after)} seconds.",
                "retry_after_seconds": int(retry_after),
            },
            headers={
                "X-RateLimit-Limit": str(limit),
                "X-RateLimit-Remaining": str(remaining),
                "X-RateLimit-Reset": str(int(time.time() + retry_after)),
                "Retry-After": str(int(retry_after)),
            }
        )


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


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


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


@app.get("/search")
def search(q: str = ""):
    """An expensive endpoint — a stricter rate limit."""
    return {"query": q, "results": []}

A load test

# Create the test_load.py file
cat > test_load.py << 'EOF'
import asyncio
import httpx


async def make_request(client, i, headers):
    try:
        r = await client.get("http://localhost:8000/products", headers=headers)
        return r.status_code, r.headers.get("x-ratelimit-remaining")
    except Exception as e:
        return None, None


async def main():
    headers = {"X-User-Id": "user42"}

    async with httpx.AsyncClient() as client:
        tasks = [make_request(client, i, headers) for i in range(150)]
        results = await asyncio.gather(*tasks)

    statuses = {}
    for status, remaining in results:
        statuses[status] = statuses.get(status, 0) + 1

    print("=== Results ===")
    for status, count in sorted(statuses.items(), key=lambda x: x[0] or 0):
        print(f"  {status}: {count}")

asyncio.run(main())
EOF

uvicorn app:app --reload &
sleep 2
python test_load.py

Expected output:

=== Results ===
  200: 100      ← the user_limiter allowed 100
  429: 50       ← the remaining 50 were blocked

✅ It works: 100 requests went through, 50 were rate-limited. The headers on each response tell you the remaining count.

Troubleshooting

Problem 1: ZCARD returns an old count (it includes requests outside the window)

Cause: You forgot the ZREMRANGEBYSCORE before the count.

Solution: Always clean up first, then count:

pipe = r.pipeline()
pipe.zremrangebyscore(key, 0, window_start)  # ✓ first
pipe.zadd(key, {member: now})
pipe.zcard(key)                                # ✓ now the count is correct
pipe.execute()

Problem 2: Redis's memory grows without limit

Cause: Sorted sets that never empty out (the TTL isn't applied, or the keys are orphaned).

Solution:

1. Always EXPIRE after the ZADD:

   pipe.expire(key, window + 60)  # safety

2. Audit with INFO memory:

   redis-cli INFO memory | grep used_memory_human

3. If you have a huge number of users, consider partitioning:

   # Instead of rate:user:1, rate:user:2, ...
   # Use rate:user:hash(user_id) % 100 — it caps the number of keys

Problem 3: Rate limiting works locally but fails in production with multiple workers

Cause: Each worker has its own connection to Redis. That's fine — they all share the same Redis. But if your Redis is on another machine, the latency adds up.

Solution:

1. Check that ALL the workers point to the same Redis:

   r = redis.Redis(host=os.getenv('REDIS_HOST', 'localhost'), port=6379)

2. Measure the latency with a remote Redis:

   start = time.time()
   r.ping()
   print(f"Redis latency: {(time.time() - start) * 1000:.1f}ms")

3. If the latency is >5 ms, consider a local Redis (a local DB) or a regional Redis cache

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

Cause: The middleware only modifies the response when there's a 429.

Solution: As in capsule 02, add the headers to ALL the responses, not just the 429s.

Problem 5: The concurrent test reports more allowed requests than expected

Cause: A race condition: two workers read ZCARD = 99, they both ZADD, they both pass, and the real counter is 101.

Solution: Use ZADD + ZCARD in a pipeline (atomic), and check count > limit with a rollback (like in the implementation above).

Better still: an implementation with a Lua script for strict atomicity:

# A Lua script (atomic)
SLIDING_WINDOW_LUA = """
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window_start = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])
local member = ARGV[4]
local ttl = tonumber(ARGV[5])

redis.call('ZREMRANGEBYSCORE', key, 0, window_start)
local count = redis.call('ZCARD', key)

if count >= limit then
    return {0, 0}
end

redis.call('ZADD', key, now, member)
redis.call('EXPIRE', key, ttl)
return {1, limit - count - 1}
"""

# Call it
result = r.eval(
    SLIDING_WINDOW_LUA,
    1,  # 1 key
    key,                    # KEYS[1]
    str(now),               # ARGV[1]
    str(window_start),      # ARGV[2]
    str(limit),             # ARGV[3]
    member,                 # ARGV[4]
    str(window + 60),       # ARGV[5]
)
allowed = bool(result[0])
remaining = result[1]

Lua scripts run atomically in Redis. If you need strict guarantees under very high concurrency, this is the approach.

Problem 6: Retry-After shows strange values

Cause: You're calculating the retry from a request that was already rolled back.

Solution: Compute retry_after based on the oldest score in the sorted set:

oldest = r.zrange(key, 0, 0, withscores=True)
if oldest:
    oldest_score = oldest[0][1]
    retry_after = max(0, oldest_score + window - now)

That gives you the exact moment when the oldest request will leave the window.

Exercises

Exercise 1: Implementing a basic sliding window (Easy)

Implement SlidingWindow with limit=5 and window=10 seconds. Test: 8 quick requests (5 go through, 3 fail), wait 11 seconds, then 5 more (they all go through).

import time, uuid, redis

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


class SlidingWindow:
    def __init__(self, limit, window):
        self.limit = limit
        self.window = window

    def check(self, key):
        full_key = f"sw:{key}"
        now = time.time()
        window_start = now - self.window
        member = f"req:{now}:{uuid.uuid4().hex}"

        pipe = r.pipeline()
        pipe.zremrangebyscore(full_key, 0, window_start)
        pipe.zadd(full_key, {member: now})
        pipe.zcard(full_key)
        pipe.expire(full_key, self.window + 60)
        _, _, count, _ = pipe.execute()

        if count > self.limit:
            r.zrem(full_key, member)
            return False
        return True


limiter = SlidingWindow(limit=5, window=10)
r.delete("sw:test")

print("=== 8 quick requests ===")
for i in range(8):
    print(f"Req {i+1}: {'✓' if limiter.check('test') else '✗'}")

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

print("\n=== 5 more requests ===")
for i in range(5):
    print(f"Req {i+9}: {'✓' if limiter.check('test') else '✗'}")

=== 8 quick requests ===
Req 1: ✓
Req 2: ✓
Req 3: ✓
Req 4: ✓
Req 5: ✓
Req 6: ✗
Req 7: ✗
Req 8: ✗

=== Wait 11 seconds ===

=== 5 more requests ===
Req 9: ✓
Req 10: ✓
Req 11: ✓
Req 12: ✓
Req 13: ✓

Exercise 2: Verifying that a sliding window does NOT have the fixed window bug (Medium)

Reproduce the "double window" attack from exercise 4 of the previous capsule, but with a sliding window. Verify that the attack fails.

import time
import redis

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


# A sliding window
limiter = SlidingWindow(limit=10, window=10)
r.delete("sw:attacker")

# Step 1: make 10 requests at the end of "window 1"
print("Step 1: 10 quick requests (filling it up)")
for i in range(10):
    allowed = limiter.check("attacker")
    print(f"  Req {i+1}: {'✓' if allowed else '✗'}")

# Step 2: wait 5 seconds (half the window)
print("\nStep 2: Wait 5 sec (the window is still active)")
time.sleep(5)

# Step 3: try 10 more
# With a FIXED window: they'd go through (the counter resets every 10s)
# With a SLIDING window: they do NOT (the 10 old ones are still in the "last 10s" window)
print("\nStep 3: 10 more requests (trying to exploit it)")
allowed_count = 0
for i in range(10):
    if limiter.check("attacker"):
        allowed_count += 1

print(f"\n✅ Only {allowed_count} out of 10 went through — the sliding window protected us.")
print(f"   Total in the last 5 sec: 10 + {allowed_count} = {10 + allowed_count}")
print(f"   If it were a fixed window: 10 + 10 = 20 (DOUBLE the limit)")

Step 1: 10 quick requests (filling it up)
  Req 1: ✓
  ...
  Req 10: ✓

Step 2: Wait 5 sec

Step 3: 10 more requests
  (all failed)

✅ Only 0 out of 10 went through — the sliding window protected us.

Exercise 3: Correct HTTP headers (Medium)

Implement a FastAPI middleware with a sliding window. Verify with curl that:

1. The 200 response includes a decreasing X-RateLimit-Remaining
2. The 429 response includes Retry-After with the correct value
3. After the retry-after, the request goes through

uvicorn app:app --reload &
sleep 2

# Check the headers on the 200
curl -i -H "X-User-Id: testuser" http://localhost:8000/products | head -5
# Expected:
# HTTP/1.1 200 OK
# X-RateLimit-Limit: 100
# X-RateLimit-Remaining: 99
# X-RateLimit-Reset: 1714069320

# Force a 429 (make 100 reqs)
for i in {1..100}; do
  curl -s -o /dev/null -H "X-User-Id: testuser" http://localhost:8000/products
done

# Request 101 → 429
curl -i -H "X-User-Id: testuser" http://localhost:8000/products
# HTTP/1.1 429 Too Many Requests
# X-RateLimit-Limit: 100
# X-RateLimit-Remaining: 0
# Retry-After: 60

# Wait Retry-After seconds
sleep 60

# Now it goes through
curl -i -H "X-User-Id: testuser" http://localhost:8000/products | head -3
# HTTP/1.1 200 OK

Exercise 4: Rate limiting for an expensive endpoint (Medium)

Configure differentiated rate limits:

• /products: 100/min
• /search: 20/min
• /reports/generate: 5/min

Implement a configuration dict and middleware that applies the right limit per endpoint.

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


ENDPOINT_LIMITS = {
    "GET:/products": (100, 60),
    "GET:/search": (20, 60),
    "POST:/reports/generate": (5, 60),
}

# The default if it isn't in the dict
DEFAULT_LIMIT = (60, 60)


def get_limit_for(method, path):
    key = f"{method}:{path}"
    return ENDPOINT_LIMITS.get(key, DEFAULT_LIMIT)


class EndpointRateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        method = request.method
        path = request.url.path
        user_id = request.headers.get("X-User-Id", "anonymous")

        limit, window = get_limit_for(method, path)
        sw = SlidingWindow(limit, window)

        key = f"{method}:{path}:user:{user_id}"
        if not sw.check(key):
            return JSONResponse(
                status_code=429,
                content={"error": f"Rate limit exceeded for {method} {path}"},
                headers={"Retry-After": str(window)}
            )

        return await call_next(request)


app = FastAPI()
app.add_middleware(EndpointRateLimitMiddleware)


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


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


@app.post("/reports/generate")
def generate_report():
    return {"report_id": "abc123"}

# /products: 100 reqs/min
for i in {1..101}; do
  curl -s -o /dev/null -w "%{http_code}\n" http://localhost:8000/products
done | sort | uniq -c
# 100 200
#   1 429

# /search: 20 reqs/min (stricter)
for i in {1..21}; do
  curl -s -o /dev/null -w "%{http_code}\n" http://localhost:8000/search
done | sort | uniq -c
# 20 200
#  1 429

Exercise 5: A concurrent test to verify atomicity (Hard)

Fire 200 concurrent requests (not sequential) with limit=100. Verify with asyncio + httpx that exactly 100 go through and 100 fail, with no race conditions.

import asyncio
import httpx


async def make_request(client, i):
    try:
        r = await client.get(
            "http://localhost:8000/products",
            headers={"X-User-Id": "test_concurrent"}
        )
        return r.status_code
    except Exception:
        return None


async def main():
    async with httpx.AsyncClient(timeout=30) as client:
        # 200 concurrent requests
        tasks = [make_request(client, i) for i in range(200)]
        results = await asyncio.gather(*tasks)

    counts = {}
    for status in results:
        counts[status] = counts.get(status, 0) + 1

    print("Results of 200 concurrent requests:")
    for status, count in sorted(counts.items(), key=lambda x: x[0] or 0):
        print(f"  {status}: {count}")

    # Verification
    if counts.get(200, 0) == 100 and counts.get(429, 0) == 100:
        print("\n✅ Atomicity verified: exactly 100 went through, 100 were blocked")
    else:
        print(f"\n⚠️ Race condition detected: {counts}")


asyncio.run(main())

Results of 200 concurrent requests:
  200: 100
  429: 100

✅ Atomicity verified: exactly 100 went through, 100 were blocked

Exercise 6: A cleanup script (Hard)

Implement a script that runs periodically (every hour) to clean out rate limiting sorted sets that are no longer in use (orphaned keys). Use SCAN to iterate without blocking.

import redis

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


def cleanup_orphaned_rate_keys(prefix="rate:", inactivity_threshold=86400):
    """
    Deletes rate limiting sorted sets that haven't had entries
    in the last N hours.
    """
    cleaned = 0
    cursor = 0

    while True:
        cursor, keys = r.scan(cursor=cursor, match=f"{prefix}*", count=100)

        for key in keys:
            # Get the latest score (timestamp) in the sorted set
            latest = r.zrange(key, -1, -1, withscores=True)
            if not latest:
                # An empty sorted set, delete it
                r.delete(key)
                cleaned += 1
                continue

            latest_score = latest[0][1]
            age_seconds = time.time() - latest_score

            if age_seconds > inactivity_threshold:
                r.delete(key)
                cleaned += 1

        if cursor == 0:
            break

    return cleaned


if __name__ == "__main__":
    print("Starting the cleanup of orphaned rate limiting keys...")
    cleaned = cleanup_orphaned_rate_keys()
    print(f"✅ {cleaned} keys deleted")


# Schedule it with cron on Linux:
# 0 * * * * /path/to/.venv/bin/python /path/to/cleanup.py

from contextlib import asynccontextmanager
import asyncio


async def periodic_cleanup():
    while True:
        await asyncio.sleep(3600)  # every hour
        try:
            cleaned = cleanup_orphaned_rate_keys()
            logger.info(f"Cleanup: {cleaned} keys removed")
        except Exception as e:
            logger.error(f"Cleanup failed: {e}")


@asynccontextmanager
async def lifespan(app):
    cleanup_task = asyncio.create_task(periodic_cleanup())
    yield
    cleanup_task.cancel()


app = FastAPI(lifespan=lifespan)

Summary

In this capsule you learned:

• Sliding window solves the fixed window bug with sorted sets
• The 3 key commands: ZREMRANGEBYSCORE (cleanup), ZADD (record), ZCARD (count)
• A pipeline for atomicity: the 3 operations in one round-trip
• A rollback when it exceeds: if the count > limit after the ZADD, remove the member you added
• Granularity: rate limiting by IP, by user, by endpoint — combinable in layers
• Standard HTTP headers: X-RateLimit-*, Retry-After, the 429 status
• A body in the 429 response: JSON with an error and retry_after_seconds
• Lua scripts: for strict atomicity under very high concurrency
• A concurrent load test: verifying atomicity with 200 parallel requests
• Periodic cleanup: SCAN + an activity check to remove orphaned keys

The critical part: a sliding window isn't optional for serious APIs. A fixed window has the "double window" bug that a sliding window mathematically avoids. Capsule 02's token bucket also works, but a sliding window is more precise for strict rate limiting.

Additional resources

1. Cloudflare: Sliding Window — A sliding window implementation in Cloudflare Workers
2. Redis: ZRANGEBYSCORE — The complete range syntax
3. Redis: EVAL (Lua scripting) — Atomic scripts for critical cases
4. GitHub Engineering: Rate Limiting — How GitHub implements rate limiting
5. Stripe Engineering: Rate Limiters — Lessons from production
6. Redis Best Practices: Rate Limiting — The official guide with several patterns

What's next?

In Capsule 04 you get into Sessions with Redis — the perfect complement to JWT. You'll learn why JWT alone doesn't allow revocation, how to combine JWT + Redis sessions, implement global logout ("log out on every device"), and list a user's active sessions. It's the module's most practical topic from a security perspective.

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