API crashes usually happen because one rogue client or a sudden traffic spike slams the backend, so implementing rate limiting is non-negotiable for any stable deployment. Depending on your traffic patterns, some algorithms handle bursts better than others.
This is the most basic approach. You track requests within a rigid time block (e.g., 60 seconds). If the counter hits the limit, further requests are blocked until the next window starts.
To fix the boundary spike, the sliding window log keeps a precise timestamp of every request. It only counts requests that happened within the last X seconds from the current millisecond.
This is my go-to for real-world AI workflow integrations. Imagine a bucket that refills with tokens at a constant rate. Requests consume tokens. If the bucket is empty, the request is rejected.
Strategy 1: Fixed Window Counter #
This is the most basic approach. You track requests within a rigid time block (e.g., 60 seconds). If the counter hits the limit, further requests are blocked until the next window starts.
import time
from collections import defaultdict
class FixedWindowLimiter:
def __init__(self, max_requests=10, window_seconds=60):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = defaultdict(list)
def allow_request(self, user_id):
now = time.time()
window_start = now - self.window_seconds
self.requests[user_id] = [t for t in self.requests[user_id] if t > window_start]
if len(self.requests[user_id]) >= self.max_requests:
return False
self.requests[user_id].append(now)
return True
The main issue here is the "boundary problem." A user could exhaust their limit at the very end of window A and immediately use another full quota at the start of window B, effectively doubling their throughput in a short burst.
Strategy 2: Sliding Window Log #
To fix the boundary spike, the sliding window log keeps a precise timestamp of every request. It only counts requests that happened within the last X seconds from the current millisecond.
import time
from collections import defaultdict
class SlidingWindowLimiter:
def __init__(self, max_requests=10, window_seconds=60):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.logs = defaultdict(list)
def allow_request(self, user_id):
now = time.time()
cutoff = now - self.window_seconds
self.logs[user_id] = [t for t in self.logs[user_id] if t > cutoff]
if len(self.logs[user_id]) >= self.max_requests:
return False
self.logs[user_id].append(now)
return True
This is much smoother, but memory is the trade-off. Storing every single timestamp for millions of users will eat up your RAM quickly.
Strategy 3: Token Bucket #
This is my go-to for real-world AI workflow integrations. Imagine a bucket that refills with tokens at a constant rate. Requests consume tokens. If the bucket is empty, the request is rejected.
import time
class TokenBucket:
def __init__(self, capacity=10, refill_rate=1):
self.capacity = capacity
self.refill_rate = refill_rate # tokens per second
self.tokens = capacity
self.last_refill = time.time()
def allow_request(self):
now = time.time()
elapsed = now - self.last_refill
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
self.last_refill = now
if self.tokens >= 1:
self.tokens -= 1
return True
return False
It's memory-efficient and handles bursts beautifully while maintaining a strict long-term average.
Implementation Tips #
Distributed State: In-memory lists only work for single-instance apps. For any real production deployment, move your counters to Redis.Client Communication: Don't just return a 429 error. Always includeX-RateLimit-Limit
,X-RateLimit-Remaining
, andX-RateLimit-Reset
headers so the client knows when to back off.Cleanup: If you use in-memory maps, you must implement a TTL or a cleanup job, otherwise, your memory usage will grow linearly with your user base.
Next Qwen3 Text Encoder: Fixing FP8 and GGUF Crashes →
All Replies （3） #
M
Saved my app once when a buggy loop in my frontend spammed the API. Essential stuff.
0
A
Don't forget to add a "Retry-After" header so clients know when to try again.
0
N
Do you recommend using Redis for the counter or keeping it in-memory for lower latency?
0