Let's Learn GenAI
  • Learn0 topics
  • Techniques0 topics
  • Courses0 courses
    View all courses →
  • GenAI Guide
    AI Career Path
    Paid AI Models & Tools
    Free AI Models & Tools
    Interview Preparation
    AI Career Path0 items
    View all paths →
  • Resources
    ArticlesResearch, releases & insight.NewsletterCurated AI, to your inbox.BenchmarkTop models, ranked.
Newsletter
  1. Home
  2. /Interview Prep
  3. /Coding
  4. /Rate Limiter

Rate Limiter

Design a thread-safe sliding-window rate limiter with per-user and global call limits using timestamp lists and a threading lock.

Last updated: September 2026

Design a thread-safe sliding-window rate limiter that enforces both per-user and global call limits. The key insight: use a timestamp list per identity - prune entries older than the window, check the length against the limit, then append. Thread safety requires a lock to make the check-then-append sequence atomic.

“Two gates matter: per-user AND global. Check global first - it is the cheaper gate. The lock wraps the entire check-then-append as one atomic unit. Sliding window beats fixed window: no artificial burst at the boundary.”

Clarifying Questions (Ask These First)

QuestionWhy it matters
What is the identity key - per user, API key, or IP?Determines the dictionary key for per-identity tracking
Per-endpoint limits or global across all endpoints?Decides whether you need one limiter or one per route
Fixed window (per minute) or sliding window (rolling 60s)?Sliding window prevents burst at boundary - more complex but fairer
Same limit for all users, or configurable per tier?Tiered limits need per-identity config lookup, not just constants

Implementation

import time
import threading
from collections import defaultdict

class RateLimiter:
    def __init__(self, global_limit: int, user_limit: int, window_seconds: int = 60):
        self.global_limit = global_limit
        self.user_limit = user_limit
        self.window = window_seconds
        self.global_calls = []
        self.user_calls = defaultdict(list)
        self._lock = threading.Lock()

    def _prune(self, call_list: list) -> list:
        now = time.time()
        return [t for t in call_list if now - t < self.window]

    def is_allowed(self, user_id: str) -> bool:
        with self._lock:
            self.global_calls = self._prune(self.global_calls)
            self.user_calls[user_id] = self._prune(self.user_calls[user_id])

            if len(self.global_calls) >= self.global_limit:
                return False
            if len(self.user_calls[user_id]) >= self.user_limit:
                return False

            now = time.time()
            self.global_calls.append(now)
            self.user_calls[user_id].append(now)
        return True


# Usage
limiter = RateLimiter(global_limit=5, user_limit=3, window_seconds=60)
for i in range(1, 7):
    result = limiter.is_allowed("yash")
    status = "ALLOWED" if result else "BLOCKED"
    print(f"Call {i}: {status} | user={len(limiter.user_calls['yash'])}/3 | global={len(limiter.global_calls)}/5")

Key Design Decisions

  • 1Sliding Window via Timestamp List - Store a list of Unix timestamps per identity. On each request, prune entries older than the window, then check length. This gives a true rolling window - no artificial burst at the minute boundary like fixed-window approaches allow.
  • 2Two-Level Limits - User and Global - Check global first (cheaper gate). If global passes, check per-user. This ordering ensures a single abusive user cannot exhaust the global budget before other users are checked.
  • 3Thread Safety with Lock - The lock wraps the entire check-then-append sequence. Without it, two threads can both pass the length check before either appends - a classic TOCTOU race condition. One lock makes it one atomic operation.
  • 4defaultdict for User State - defaultdict(list) auto-initializes an empty list for new user IDs. No need to check if the key exists before appending - clean and Pythonic.

Frequently asked questions

What is the race condition here, and how does the lock fix it?

Without the lock, two threads from the same user can both execute len(user_calls[user_id]) >= limit and both pass (say count is 2, limit is 3). Then both append, pushing count to 4 - exceeding the limit silently. The lock makes check + append one atomic unit: only one thread can be inside that block at a time.

What is the memory complexity as users grow?

O(U x W) where U is the number of unique users and W is calls-per-window. For 10,000 users each making 100 calls per minute, that is 1M timestamps in memory. Each Python float is 28 bytes, so roughly 28MB - fine for most services. For millions of users, move to Redis with sorted sets.

How would you scale this across multiple servers?

Replace the in-process dict with Redis sorted sets. Use ZADD to add timestamps, ZREMRANGEBYSCORE to prune, ZCARD to count. Wrap in a Lua script or MULTI/EXEC transaction to make it atomic across servers. Same sliding window logic, now distributed.

What is the difference between sliding window and fixed window?

Fixed window resets at each interval boundary (every minute at :00). A user can send limit calls at :58, the window resets at :00, and they send limit calls again - double the rate in 2 seconds. Sliding window tracks the rolling last-60-seconds, so this burst is impossible.

How would you add token bucket semantics instead?

Track tokens (float) and last_refill_time per user. On each request, calculate elapsed time, add (elapsed * rate) tokens capped at bucket capacity, then check if tokens >= 1. If yes, subtract 1 and allow. Token bucket allows smooth bursts up to capacity rather than hard windowed counts.

How do you tell the client when they can retry?

Return a Retry-After header with the timestamp of the oldest call in the window plus the window size. That is when the oldest call expires and one slot opens. Standard practice: return HTTP 429 with that header.

Where this skill is used

AI roles that rely on this day to day, with salaries and the path in.

  • AI/ML Engineer
  • LLM Engineer
  • Generative AI Developer
  • AI Reliability Engineer
  • AI Infrastructure Optimizer
  • AI Copilot Engineer

Related coding challenges

  • RAG PipelineImplement a retrieval-augmented generation pipeline: ingest documents, chunk with overlap, embed, and retrieve top-K similar chunks via dot-product similarity.
  • Retry HandlerImplement a configurable retry handler with exponential backoff, jitter, and a max-delay cap to prevent thundering herds and unbounded blocking.

Newsletter

Four editions, one inbox

The Build Layer for developers, The Strategy Signal for managers, The Executive Brief for executives.

Pick your edition

Guided courses

Get certified, not just informed

Guided courses from beginner to advanced, each with a named certificate. Free to take, yours to keep.

Browse courses
Let's Learn GenAI

Your guided portal to AI fundamentals, advanced techniques, and industry resources.

Learn

FundamentalsTechniquesCareersPaid AI ToolsFree AI ToolsBenchmarks

Explore

ArticlesNewsletterAbout

Legal

Terms of ServicePrivacy Policy

© 2026 Let's Learn GenAI. All rights reserved.