FastAPI 標準限流器為何難以應付 LLM 與 SSE 串流

If you are building an AI chat assistant, real-time dashboard, or financial ticker in Python using FastAPI, chances are you are using Server-Sent Events (StreamingResponse) or WebSockets.

And if you care about uptime and cloud costs, you probably installed a rate-limiting middleware like slowapi or configured a Redis token bucket.

Here is the problem: standard rate limiters are designed for request frequency, not connection duration.

In this post, we'll break down why traditional rate limiters collapse under streaming workloads, and how to implement a distributed lease pattern using Redis to protect your workers.

The Root Cause: Frequency vs. Duration

Traditional HTTP endpoints are transactional:

1. Client sends request.
2. Server queries DB (50ms).
3. Server returns JSON response.
4. Connection closes.

In this model, limiting by frequency (10 requests per minute) works reliably.

What Happens with Streaming (SSE / LLM Tokens)?

An LLM stream (OpenAI, Claude, Ollama) stays open for 30 to 60 seconds while tokens are yielded chunk by chunk.

If you have a rate limit of 10 requests per minute:

• A malicious or impatient user opens 10 browser tabs simultaneously.
• Each tab requests an SSE stream.
• The rate limiter inspects the counter: "10 requests in 60s? Allowed!"
• The result: That single user now holds 10 open sockets concurrently for a full minute.

Standard Rate Limiting (Permissive Failure):
User Request 1 ──► [OK] ──► Holding Socket (60s) ──────┐
User Request 2 ──► [OK] ──► Holding Socket (60s) ──────┤
User Request 3 ──► [OK] ──► Holding Socket (60s) ──────┼──► ASGI Worker Pool Exhausted!
User Request 4 ──► [OK] ──► Holding Socket (60s) ──────┤    (HTTP 502 / Socket Starvation)
User Request 5 ──► [OK] ──► Holding Socket (60s) ──────┘

If your Uvicorn worker pool has 100 concurrent connection limits, just 10 users doing this simultaneously will starve the worker, causing HTTP 502/504 timeouts for all regular API endpoints.

The Solution: Connection Leases, Not Request Counts

To protect streaming endpoints, you need to answer a different question:

"How many active streams is this user holding RIGHT NOW?"

Instead of an incrementing counter with a TTL, you need a distributed lease mechanism:

1. Before streaming: Try to acquire an atomic slot for user_id.
2. If active_slots >= limit: Reject immediately with HTTP 429 Too Many Requests before triggering the LLM generation.
3. While streaming: Hold the lease and periodically refresh it.
4. On completion or client disconnect: Release the slot immediately.

Implementing Atomic Leases in Redis (Lua CAS)

A naive implementation using multiple Redis commands (GET then INCR) introduces race conditions. Two parallel requests could both read count=1 and both succeed, violating your concurrency limit.

To make lease acquisition atomic, we use Redis Sorted Sets (ZSET) where:

• Key: stream_lease:{user_id}
• Score: Current Unix epoch timestamp
• Member: Unique lease_id (UUID)

Here is the atomic Lua script pattern ($O(\log N)$):

local key = KEYS[1]
local now = tonumber(ARGV[1])
local max_leases = tonumber(ARGV[2])
local ttl = tonumber(ARGV[3])
local lease_id = ARGV[4]

-- 1. Remove expired/stale leases
redis.call('ZREMRANGEBYSCORE', key, '-inf', now - ttl)

-- 2. Check current active lease count
local current_count = redis.call('ZCARD', key)
if current_count >= max_leases then
    return 0 -- Rejected (Concurrency limit reached)
end

-- 3. Acquire lease
redis.call('ZADD', key, now, lease_id)
redis.call('EXPIRE', key, ttl * 2)
return 1 -- Allowed

When the client disconnects or the generator completes, a cleanup script removes the lease_id:

redis.call('ZREM', KEYS[1], ARGV[1])

Handling the Edge Cases in Production

In real-world microservices, three critical failure modes occur:

1. Zombie Tasks on Abrupt Disconnections

If a user closes their laptop mid-stream, Starlette detects the socket disconnect via request.is_disconnected(). But if your generator doesn't actively poll this flag or catch the CancelledError, the background task keeps running, consuming downstream AI tokens and keeping the Redis lease alive until expiration.

2. Redis Sentinel Failover (ReadOnlyError)

When Redis switches from primary to replica, pending writes raise ReadOnlyError. If your lease manager doesn't catch this, your entire API crashes. You need an automated fail-open or fail-closed fallback policy.

Enter fastapi-stream-lease

Instead of writing custom Lua scripts, disconnect wrappers, and metrics hooks in every microservice, I packaged this into a small, dependency-light library:

👉 fastapi-stream-lease on GitHub (MIT, 100% test coverage)

1-Line Integration Example

from fastapi import FastAPI
from fastapi_stream_lease import StreamLeaseManager, LeaseConfig

app = FastAPI()
manager = StreamLeaseManager(redis_url="redis://localhost:6379")

# Register 1-line standard 429 exception handler
manager.register_exception_handlers(app)

@app.get("/stream")
async def chat_stream(user_id: str):
    async def token_generator():
        for i in range(10):
            yield f"Token {i} "

    # Acquires atomic Redis lease, wraps StreamingResponse,
    # detects disconnects, and releases slot on finish or error.
    return await manager.stream(
        user_id=user_id,
        content=token_generator(),
        media_type="text/event-stream",
        config=LeaseConfig(user_limit=2, ttl_seconds=30),
    )

What You Get Out of the Box:

• 1-Line Integration: Wrap any async generator in manager.stream().
• Atomic Concurrency Bounds: Backed by Redis Lua CAS ($O(\log N)$).
• Graceful Fallback: In-memory circuit breaker handles Redis failovers safely.
• Production Observability: Native Prometheus and OpenTelemetry gauges (fastapi_stream_lease_active_streams).
• Resilient Reconnection: Disconnect hooks clean up slots immediately.

Summary

When designing systems for LLM streaming and SSE, remember:

• Rate limiters control how fast users knock on your door.
• Lease managers control how many rooms they can occupy at once.

If you are running streaming endpoints in production, you need both.

• Repo: github.com/agustin18/fastapi-stream-lease
• PyPI: pypi.org/project/fastapi-stream-lease

(Note: English is not my native language; drafted with some AI assistance for translation and phrasing, but the architecture and code come directly from our production experience).

How are you currently handling streaming concurrency in your stack? Feedback and architecture critiques are welcome in the comments!