Have you ever found your high-concurrency async Python application silently failing under load, struggling with unexpected API rate limits, or worse, accidentally DDoSing an external service? I certainly have. Building robust, scalable services means anticipating and controlling traffic, but implementing a truly production-grade distributed rate limiter is far more complex than a simple in-memory counter. While many senior engineers quickly reach for Redis, the devil is in the details: without atomic operations, you're still vulnerable to race conditions that can lead to subtle, yet critical, rate-limit bypasses. In this post, I'll walk you through how I engineered a resilient, distributed token bucket rate limiter using Python's asyncio and atomic Redis Lua scripts, simulating real concurrent requests by fetching live data from the Stripe blog RSS feed. You'll learn how to prevent those elusive race conditions, manage high-throughput async workloads gracefully, and build a system that stands up to production demands. My core judgment? For distributed Python services, atomic Redis Lua scripts are not just an optimization; they are a fundamental requirement for reliable rate limiting.
Key Takeaways
- In-memory rate limiters are insufficient for distributed, multi-worker Python environments due to lack of shared state, making Redis a mandatory component for consistent enforcement.
- Atomic Redis Lua scripts are essential to eliminate race conditions in distributed rate limiting by performing token checks, refills, and decrements as a single, indivisible operation.
- Leveraging
asyncio.gatherandasyncio.Semaphoreallows for orchestrating high-throughput concurrent requests, ensuring your application gracefully handles backpressure without starving the event loop. - Robust error handling for Redis connection pooling, network partitions, and fail-open/fail-closed strategies is critical to prevent your rate limiter from becoming a single point of failure.
The Problem: Race Conditions in Distributed Rate Limiting
The challenge I often see in building concurrent async applications is the temptation to implement naive rate limiting. An in-memory token bucket might seem sufficient for a single process, but it immediately falls apart in a distributed environment with multiple Python workers or instances. Each worker would maintain its own bucket, leading to a complete bypass of the rate limit. Moving to Redis solves the distributed state problem, but introduces another: race conditions. If one worker checks the token count, finds it sufficient, and then another worker performs the same check before the first one can decrement the tokens, both requests might be allowed, exceeding the intended limit. This "check-then-set" pattern is a classic concurrency pitfall, leading to silent rate-limit breaches under heavy load and potentially impacting external services.
Data and Sources
For this exercise, we'll simulate a high-concurrency workload by fetching and processing entries from the public Stripe Blog RSS feed. Each entry will represent an incoming request that needs to be rate-limited before further processing.
- Stripe Blog RSS Feed:
https://stripe.com/blog/feed.rss(parsed usingfeedparserand fetched viahttpx). - Redis documentation on EVAL and Lua scripting atomicity:
https://redis.io/docs/manual/programmability/eval-intro/ - Python
asynciosynchronization primitives documentation:https://docs.python.org/3/library/asyncio-sync.html
Step 1 — Designing the Atomic Token Bucket Lua Script
To eliminate the insidious race conditions inherent in application-layer check-then-set logic, the entire token bucket algorithm must execute atomically. Redis's Lua scripting feature is perfect for this. When a Lua script runs, it's guaranteed to complete without interruption from other Redis commands, effectively acting as a single, atomic transaction. Our script will encapsulate the logic for calculating token refills, decrementing the bucket, and returning the current state.
The Lua script takes parameters for the bucket key, capacity, refill rate, current timestamp, cost per request, and an expiry time for the Redis key. It retrieves the last refill time and current tokens, calculates any tokens gained since the last check, updates the bucket (capped at capacity), and then attempts to decrement for the current request. If successful, it stores the new state and sets a TTL; otherwise, it returns that the request is denied.
-- KEYS[1] = bucket_key (e.g., "rate_limit:global_app")
-- ARGV[1] = capacity
-- ARGV[2] = refill_rate (tokens per second)
-- ARGV[3] = current_timestamp (in seconds)
-- ARGV[4] = cost (tokens per request, usually 1)
-- ARGV[5] = expiry_time (in seconds, for EXPIRE)
local bucket_key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local current_timestamp = tonumber(ARGV[3])
local cost = tonumber(ARGV[4])
local expiry_time = tonumber(ARGV[5])
local last_refill_time = tonumber(redis.call('HGET', bucket_key, 'last_refill_time'))
local tokens = tonumber(redis.call('HGET', bucket_key, 'tokens'))
if not last_refill_time then
last_refill_time = current_timestamp
tokens = capacity
end
-- Calculate tokens to add since last refill
local time_passed = current_timestamp - last_refill_time
local tokens_to_add = time_passed * refill_rate
-- Update tokens and refill time
tokens = math.min(capacity, tokens + tokens_to_add)
last_refill_time = current_timestamp
local allowed = 0
local new_tokens = tokens
if tokens >= cost then
new_tokens = tokens - cost
allowed = 1
end
-- Store updated state
redis.call('HSET', bucket_key, 'tokens', new_tokens)
redis.call('HSET', bucket_key, 'last_refill_time', last_refill_time)
redis.call('EXPIRE', bucket_key, expiry_time)
return {allowed, new_tokens}
This script is the heart of our distributed rate limiter. By executing it entirely within Redis, we guarantee that no other client can interfere between reading the current state and writing the new state, thus preventing any race conditions.
Step 2 — Asynchronous Redis Client Integration
With our atomic Lua script ready, the next step is to integrate it efficiently with our Python asyncio application. Traditional Redis clients might block the event loop, defeating the purpose of asynchronous programming. The redis.asyncio library provides a non-blocking interface, crucial for high-performance services. We'll set up a client, load our Lua script once (getting its SHA-1 hash), and then execute it using EVALSHA, which is more efficient than sending the full script every time.
import redis.asyncio as redis
import time
import logging
class RedisRateLimiter:
LUA_SCRIPT = """
-- (Lua script from Step 1 goes here)
"""
def __init__(self, redis_client: redis.Redis, capacity: int, refill_rate: float, expiry_time: int):
self.redis_client = redis_client
self.capacity = capacity
self.refill_rate = refill_rate
self.expiry_time = expiry_time
self.script_sha = None
logging.getLogger(__name__).info(f"Rate Limiter initialized: capacity={capacity}, refill_rate={refill_rate} tokens/sec")
async def _load_script(self):
if not self.script_sha:
self.script_sha = await self.redis_client.script_load(self.LUA_SCRIPT)
logging.getLogger(__name__).debug(f"Lua script loaded, SHA: {self.script_sha}")
async def allow_request(self, key: str, cost: int = 1) -> tuple[bool, int]:
if not self.script_sha:
await self._load_script() # Ensure script is loaded on first use
timestamp = int(time.time())
try:
result = await self.redis_client.evalsha(
self.script_sha,
1, # Number of keys
key,
self.capacity,
self.refill_rate,
timestamp,
cost,
self.expiry_time
)
allowed, tokens_left = result
return bool(allowed), int(tokens_left)
except redis.RedisError as e:
logging.getLogger(__name__).error(f"Redis error during rate limit check for key '{key}': {e}")
# Fail-open strategy: allow request if Redis is down
return True, self.capacity # Assume full bucket if Redis fails
The RedisRateLimiter class wraps the Redis interactions. The _load_script method ensures the Lua script is loaded into Redis's script cache only once, and allow_request then uses evalsha for subsequent calls. This minimizes network overhead and leverages Redis's efficient script execution. Notice the basic error handling: if Redis itself is unreachable or throws an error, we default to a "fail-open" strategy, allowing the request. This prevents the rate limiter from becoming a single point of failure that takes down your entire service.
Step 3 — Orchestrating Concurrent Workers with Asyncio
Now that we have an atomic rate limiter, we need to orchestrate our concurrent "requests" to demonstrate its effectiveness. We'll simulate fetching and processing entries from the Stripe blog RSS feed. To control the degree of concurrency and prevent overwhelming our application's resources or the network, we'll use asyncio.Semaphore. This primitive limits the number of concurrent tasks running at any given time, providing a crucial layer of backpressure management.
import asyncio
import httpx
import feedparser
async def fetch_stripe_feed(url: str) -> list:
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, timeout=10)
response.raise_for_status()
feed = feedparser.parse(response.text)
return feed.entries
except httpx.RequestError as e:
logging.getLogger(__name__).error(f"Error fetching RSS feed from {url}: {e}")
return []
async def worker(
item: dict,
limiter: RedisRateLimiter,
semaphore: asyncio.Semaphore,
worker_id: int,
request_idx: int
):
async with semaphore:
request_key = "app:global_rate_limit" # Applying a global rate limit for demonstration
allowed, tokens_left = await limiter.allow_request(request_key)
if allowed:
logging.getLogger(__name__).info(
f"[Worker {worker_id}, Req {request_idx}] Allowed. Tokens left: {tokens_left:.2f}. "
f"Processing '{item.title}'"
)
# Simulate some async work
await asyncio.sleep(0.1)
else:
logging.getLogger(__name__).warning(
f"[Worker {worker_id}, Req {request_idx}] Rate limited! Tokens left: {tokens_left:.2f}. "
f"Skipping '{item.title}'"
)
The fetch_stripe_feed function asynchronously retrieves and parses the RSS data. The worker function is where the magic happens: it acquires a semaphore slot, calls our rate limiter, and then either processes the item (simulated with a short asyncio.sleep) or logs a rate-limit rejection. By passing the same RedisRateLimiter instance to all workers, they all contend for tokens from the same distributed bucket in Redis, accurately simulating a shared resource under load.
Step 4 — Production Error Handling and Circuit Breaking
In a production environment, network failures, Redis outages, or unexpected errors are inevitable. Our rate limiter needs to be resilient. I've already touched on the "fail-open" strategy within the allow_request method for Redis errors. This means if Redis is unreachable, we temporarily disable rate limiting to ensure core service functionality remains available, albeit without protection. This is a crucial design decision and depends on your business requirements; some systems might prefer "fail-closed" (deny all requests if the limiter is down).
Beyond Redis-specific errors, we also need to handle issues when fetching external data like the RSS feed. The fetch_stripe_feed function includes a try/except httpx.RequestError block to catch network issues or invalid responses. In a real application, you might add retry logic, exponential backoff, or even a circuit breaker pattern (like the one I discussed previously for FastAPI wrappers) for the external API calls themselves, separate from the rate limiter.
The main orchestrator will need to gracefully handle potential empty results from fetch_stripe_feed and ensure all tasks complete or are cancelled properly.
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import asyncio
import httpx
import feedparser
import redis.asyncio as redis
import time
import logging
# --- Configuration ---
REDIS_HOST = "localhost"
REDIS_PORT = 6379
STRIPE_RSS_FEED_URL = "https://stripe.com/blog/feed.rss"
# Rate limit configuration: 5 requests per 10 seconds (0.5 tokens/sec)
RATE_LIMIT_CAPACITY = 5
RATE_LIMIT_REFILL_RATE = 0.5 # tokens per second
RATE_LIMIT_EXPIRY_TIME = int(RATE_LIMIT_CAPACITY / RATE_LIMIT_REFILL_RATE) + 5 # TTL for Redis key, slightly longer than refill time
MAX_CONCURRENT_WORKERS = 10 # Max concurrent Python workers
SIMULATED_REQUESTS_PER_ENTRY = 4 # How many times each RSS entry is processed to stress the limiter
# --- Logger Setup ---
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
datefmt='%H:%M:%S'
)
logger = logging.getLogger(__name__)
# --- Redis Lua Script for Atomic Token Bucket ---
RATE_LIMIT_LUA_SCRIPT = """
-- KEYS[1] = bucket_key (e.g., "rate_limit:global_app")
-- ARGV[1] = capacity
-- ARGV[2] = refill_rate (tokens per second)
-- ARGV[3] = current_timestamp (in seconds)
-- ARGV[4] = cost (tokens per request, usually 1)
-- ARGV[5] = expiry_time (in seconds, for EXPIRE)
local bucket_key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local current_timestamp = tonumber(ARGV[3])
local cost = tonumber(ARGV[4])
local expiry_time = tonumber(ARGV[5])
local last_refill_time = tonumber(redis.call('HGET', bucket_key, 'last_refill_time'))
local tokens = tonumber(redis.call('HGET', bucket_key, 'tokens'))
if not last_refill_time then
last_refill_time = current_timestamp
tokens = capacity
end
-- Calculate tokens to add since last refill
local time_passed = current_timestamp - last_refill_time
local tokens_to_add = time_passed * refill_rate
-- Update tokens and refill time
tokens = math.min(capacity, tokens + tokens_to_add)
last_refill_time = current_timestamp
local allowed = 0
local new_tokens = tokens
if tokens >= cost then
new_tokens = tokens - cost
allowed = 1
end
-- Store updated state
redis.call('HSET', bucket_key, 'tokens', new_tokens)
redis.call('HSET', bucket_key, 'last_refill_time', last_refill_time)
redis.call('EXPIRE', bucket_key, expiry_time)
return {allowed, new_tokens}
"""
class RedisRateLimiter:
def __init__(self, redis_client: redis.Redis, capacity: int, refill_rate: float, expiry_time: int):
self.redis_client = redis_client
self.capacity = capacity
self.refill_rate = refill_rate
self.expiry_time = expiry_time
self.script_sha = None
logger.info(f"Rate Limiter initialized: capacity={capacity}, refill_rate={refill_rate} tokens/sec, expiry={expiry_time}s")
async def _load_script(self):
if not self.script_sha:
try:
self.script_sha = await self.redis_client.script_load(RATE_LIMIT_LUA_SCRIPT)
logger.debug(f"Lua script loaded, SHA: {self.script_sha}")
except redis.RedisError as e:
logger.critical(f"Failed to load Redis Lua script: {e}. Rate limiter cannot function.")
raise
async def allow_request(self, key: str, cost: int = 1) -> tuple[bool, int]:
if not self.script_sha:
try:
await self._load_script()
except Exception: # If script load failed, we can't proceed with atomic limiting
return True, self.capacity # Fail-open if script cannot be loaded
timestamp = int(time.time())
try:
result = await self.redis_client.evalsha(
self.script_sha,
1, # Number of keys
key,
self.capacity,
self.refill_rate,
timestamp,
cost,
self.expiry_time
)
allowed, tokens_left = result
return bool(allowed), int(tokens_left)
except redis.RedisError as e:
logger.error(f"Redis error during rate limit check for key '{key}': {e}. Falling back to fail-open.")
# Fail-open strategy: allow request if Redis interaction fails
return True, self.capacity # Assume full bucket if Redis fails
async def fetch_stripe_feed(url: str) -> list:
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, timeout=10)
response.raise_for_status() # Raise an exception for 4xx/5xx responses
feed = feedparser.parse(response.text)
logger.info(f"Successfully fetched {len(feed.entries)} entries from Stripe RSS feed.")
return feed.entries
except httpx.RequestError as e:
logger.error(f"Network or request error fetching RSS feed from {url}: {e}")
return []
except Exception as e:
logger.error(f"Unexpected error parsing RSS feed from {url}: {e}")
return []
async def worker(
item: dict,
limiter: RedisRateLimiter,
semaphore: asyncio.Semaphore,
worker_id: int,
request_idx: int
):
async with semaphore:
request_key = "app:global_rate_limit" # Using a single global key for this example
allowed, tokens_left = await limiter.allow_request(request_key)
if allowed:
logger.info(
f"[Worker {worker_id}, Req {request_idx}] Allowed. Tokens left: {tokens_left:.2f}. "
f"Processing: '{item.title}'"
)
# Simulate some async work that would use the fetched data
await asyncio.sleep(0.1)
else:
logger.warning(
f"[Worker {worker_id}, Req {request_idx}] Rate limited! Tokens left: {tokens_left:.2f}. "
f"Skipping: '{item.title}'"
)
async def main():
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)
try:
# Ping Redis to ensure connection
await redis_client.ping()
logger.info("Successfully connected to Redis.")
except redis.RedisError as e:
logger.critical(f"Could not connect to Redis at {REDIS_HOST}:{REDIS_PORT}: {e}. Exiting.")
await redis_client.close()
return
rate_limiter = RedisRateLimiter(redis_client, RATE_LIMIT_CAPACITY, RATE_LIMIT_REFILL_RATE, RATE_LIMIT_EXPIRY_TIME)
# Pre-load the script once
try:
await rate_limiter._load_script()
except Exception: # _load_script raises if critical failure
logger.critical("Failed to initialize rate limiter due to script loading error. Shutting down.")
await redis_client.close()
return
feed_entries = await fetch_stripe_feed(STRIPE_RSS_FEED_URL)
if not feed_entries:
logger.error("No RSS feed entries to process. Exiting.")
await redis_client.close()
return
semaphore = asyncio.Semaphore(MAX_CONCURRENT_WORKERS)
tasks = []
request_counter = 0
# Create multiple tasks to stress the rate limiter
for _ in range(SIMULATED_REQUESTS_PER_ENTRY):
for i, entry in enumerate(feed_entries[:5]): # Use first 5 entries for simulation
request_counter += 1
tasks.append(
asyncio.create_task(
worker(entry, rate_limiter, semaphore, i, request_counter)
)
)
logger.info(f"Starting {len(tasks)} simulated requests with {MAX_CONCURRENT_WORKERS} concurrent Python workers...")
start_time = time.time()
await asyncio.gather(*tasks)
end_time = time.time()
logger.info(f"All {len(tasks)} requests completed in {end_time - start_time:.2f} seconds.")
await redis_client.close()
logger.info("Redis client closed.")
if __name__ == '__main__':
asyncio.run(main())
Expected Output
When you run the script (ensure Redis is running on localhost:6379), you should see a stream of log messages. Initially, requests will be "Allowed" until the rate limit bucket is depleted. Then, you'll see "Rate limited!" warnings for subsequent requests within the rate limit window. After a brief pause (as tokens refill), requests will start being "Allowed" again. The output demonstrates how the atomic rate limiter effectively throttles concurrent requests, even when many workers are trying to access the shared resource simultaneously.
HH:MM:SS - INFO - Rate Limiter initialized: capacity=5, refill_rate=0.5 tokens/sec, expiry=15s
HH:MM:SS - INFO - Successfully connected to Redis.
HH:MM:SS - INFO - Successfully fetched XX entries from Stripe RSS feed.
HH:MM:SS - INFO - Starting 20 simulated requests with 10 concurrent Python workers...
HH:MM:SS - INFO - [Worker 0, Req 1] Allowed. Tokens left: 4.00. Processing: 'Trends from Money20/20 2026: The rise of “hidden” financial services'
HH:MM:SS - INFO - [Worker 1, Req 2] Allowed. Tokens left: 3.00. Processing: 'Stripe survey: Four trends reshaping how SaaS platforms compete in the AI era'
HH:MM:SS - INFO - [Worker 2, Req 3] Allowed. Tokens left: 2.00. Processing: 'New ways for fintechs to move money and expand globally'
HH:MM:SS - INFO - [Worker 3, Req 4] Allowed. Tokens left: 1.00. Processing: 'Sync real-time Stripe data to Postgres with Stripe Data Pipeline'
HH:MM:SS - INFO - [Worker 4, Req 5] Allowed. Tokens left: 0.00. Processing: 'The “alien” economics of AI monetization'
HH:MM:SS - WARNING - [Worker 0, Req 6] Rate limited! Tokens left: 0.00. Skipping: 'Trends from Money20/20 2026: The rise of “hidden” financial services'
HH:MM:SS - WARNING - [Worker 1, Req 7] Rate limited! Tokens left: 0.00. Skipping: 'Stripe survey: Four trends reshaping how SaaS platforms compete in the AI era'
... (more rate limited messages) ...
HH:MM:SS - INFO - [Worker 0, Req 11] Allowed. Tokens left: 0.50. Processing: 'Trends from Money20/20 2026: The rise of “hidden” financial services'
... (requests start being allowed again as tokens refill) ...
HH:MM:SS - INFO - All 20 requests completed in X.XX seconds.
HH:MM:SS - INFO - Redis client closed.
Limitations and Tradeoffs
While this Redis-backed token bucket is robust, it's not a silver bullet. This solution introduces a network hop to Redis for every rate limit check, which can add latency. For extremely low-latency trading systems or applications where every microsecond counts, this overhead might be unacceptable. Alternatives like local token buckets synchronized via gossip protocols or more complex distributed consensus mechanisms could be considered, but they introduce significant operational complexity. Furthermore, while Redis is fast, a single Redis instance can still be a bottleneck if your request volume scales beyond its capacity. Horizontal scaling with Redis Cluster or sharding your rate limit keys becomes necessary in such scenarios. Finally, a fail-open strategy provides resilience but temporarily exposes your downstream services to potential overload; a fail-closed approach offers stricter protection but risks service unavailability if Redis goes down.
Frequently Asked Questions
How do you handle Redis cluster key slot routing when rate-limiting across different user IDs?
You must use Redis hash tags by enclosing the dynamic identifier in curly braces, such as {user:42}:rate_limit. This forces Redis Cluster to route all keys with the exact same tag (e.g., user:42) to the same master node, preventing cross-slot evaluation errors during Lua script execution. Our current example uses a single global key, but for per-user limits, this is critical.
What is the memory footprint of storing millions of dynamic rate-limiting keys in Redis, and how do you set TTLs?
Each Redis hash consumes a small base memory overhead (around 100-200 bytes depending on configuration), plus the memory for the fields and their values. To prevent unbounded memory growth from abandoned user sessions, always assign an explicit TTL using the EXPIRE command inside your Lua script, set slightly longer than the time required to fully refill the bucket. Our script uses ARGV[5] for this, ensuring keys expire after a period of inactivity.
How do you test a distributed rate limiter under heavy concurrency in local CI/CD pipelines?
You can use pytest combined with pytest-asyncio and a lightweight containerized Redis instance (e.g., using Docker or Testcontainers) or even fakeredis for unit tests. For integration tests, spin up a dedicated Redis container, then use libraries like locust or even a custom asyncio script similar to our main function to generate a high volume of concurrent requests against your rate-limited service.
When should you choose a Fixed Window vs. Sliding Window vs. Token Bucket algorithm?
The Token Bucket algorithm, which we've implemented, is generally preferred for its smooth rate limiting and ability to handle bursts. Fixed Window is simpler but can allow bursts at the window edges. Sliding Window (often combined with a log of timestamps) offers better burst control than Fixed Window but is more complex and memory-intensive than Token Bucket for Redis implementations.
What I'd Change
If I were to take this into a more complex production environment, my primary change would be to introduce a more granular rate-limiting strategy. While a global application rate limit is useful, most real-world scenarios demand per-user, per-endpoint, or even per-IP rate limits. This would involve making the request_key dynamic (e.g., f"app:user:{user_id}:endpoint:{endpoint_name}") and carefully managing the key space and memory footprint in Redis. I'd also implement a more sophisticated error handling and circuit-breaking mechanism, possibly using a dedicated library, that could dynamically switch between fail-open and fail-closed based on the criticality of the Redis service and the nature of the downstream calls. Finally, for observability, I'd integrate Prometheus metrics to track allowed/denied requests and current token counts, providing real-time insights into the rate limiter's performance and helping tune its parameters more effectively.
Next Steps
Now that you've seen how to build a robust, distributed token bucket rate limiter, try modifying the script to implement per-user rate limits. How would you adjust the Lua script and the Python client to handle millions of unique user keys while efficiently managing Redis memory and ensuring high performance?