Design a Rate Limiter — Clock Skew Lets 2x Traffic Through

import time
import redis
from datetime import datetime

class FixedWindowRateLimiter:
    def __init__(self, redis_client, limit_per_window, window_seconds):
        self.redis = redis_client
        self.limit = limit_per_window
        self.window_seconds = window_seconds

    def _get_window_key(self, key):
        """Generate key with current window number (UNIX timestamp / window size)."""
        current_window = int(time.time() / self.window_seconds)
        return f"rate_limit:{key}:{current_window}"

    def allow_request(self, key):
        """
        Returns (allowed, current_count).
        WARNING: Not atomic — race condition possible at high concurrency.
        Use Lua script for production (see sliding_window_counter.py).
        """
        window_key = self._get_window_key(key)
        
        # Pipeline: INCR + EXPIRE (if key doesn't exist, EXPIRE sets TTL)
        pipe = self.redis.pipeline()
        pipe.incr(window_key)
        pipe.expire(window_key, self.window_seconds)
        result = pipe.execute()
        
        current_count = result[0]  # INCR result
        
        if current_count <= self.limit:
            return True, current_count
        else:
            return False, current_count

    # ─────────────────────────────────────────────────────────────
    # PRODUCTION-READY VERSION (Atomic Lua script)
    # ─────────────────────────────────────────────────────────────
    
    FIXED_WINDOW_LUA = """
        local key = KEYS[1]
        local limit = tonumber(ARGV[1])
        local window_seconds = tonumber(ARGV[2])
        
        local current = redis.call('INCR', key)
        if current == 1 then
            redis.call('EXPIRE', key, window_seconds)
        end
        
        if current <= limit then
            return {1, current}
        else
            return {0, current}
        end
    """
    
    def allow_request_atomic(self, key):
        """Use Lua script for atomic operation."""
        window_key = self._get_window_key(key)
        allowed, current = self.redis.eval(
            self.FIXED_WINDOW_LUA,
            1,  # number of keys
            window_key,
            self.limit,
            self.window_seconds
        )
        return allowed == 1, current

import time
import redis
import uuid

class SlidingWindowLogRateLimiter:
    def __init__(self, redis_client, limit_per_window, window_seconds):
        self.redis = redis_client
        self.limit = limit_per_window
        self.window_seconds = window_seconds

    def allow_request(self, key):
        """
        Sliding window log using Redis ZSET.
        Memory O(number of requests in window) per key.
        """
        zset_key = f"rate_limit_log:{key}"
        now = time.time()
        window_start = now - self.window_seconds
        
        # Use Lua script for atomic cleanup + count
        lua_script = """
            local key = KEYS[1]
            local now = tonumber(ARGV[1])
            local window_start = tonumber(ARGV[2])
            local limit = tonumber(ARGV[3])
            local request_id = ARGV[4]
            
            -- Remove entries outside current window
            redis.call('ZREMRANGEBYSCORE', key, 0, window_start)
            
            -- Get current count
            local count = redis.call('ZCARD', key)
            
            if count < limit then
                -- Add new request
                redis.call('ZADD', key, now, request_id)
                -- Set TTL to window_seconds + buffer
                redis.call('EXPIRE', key, window_start + 60)
                return {1, count + 1}
            else
                return {0, count}
            end
        """
        
        request_id = str(uuid.uuid4())
        allowed, count = self.redis.eval(
            lua_script,
            1,  # number of keys
            zset_key,
            now,
            window_start,
            self.limit,
            request_id
        )
        return allowed == 1, count
    
    def get_memory_usage(self, key):
        """Estimate memory usage for a key (in bytes)."""
        zset_key = f"rate_limit_log:{key}"
        return self.redis.memory_usage(zset_key) or 0

import time
import redis

class TokenBucketRateLimiter:
    def __init__(self, redis_client, capacity, refill_rate_per_second):
        self.redis = redis_client
        self.capacity = capacity
        self.refill_rate = refill_rate_per_second
    
    def allow_request(self, key):
        """
        Token bucket using Redis Lua script.
        Memory O(1) per key (stores tokens and last_refill timestamp).
        """
        bucket_key = f"rate_limit_token:{key}"
        now = time.time()
        
        lua_script = """
            local key = KEYS[1]
            local now = tonumber(ARGV[1])
            local capacity = tonumber(ARGV[2])
            local refill_rate = tonumber(ARGV[3])
            
            local data = redis.call('HMGET', key, 'tokens', 'last_refill')
            local tokens = tonumber(data[1]) or capacity
            local last_refill = tonumber(data[2]) or now
            
            -- Calculate tokens to add
            local elapsed = now - last_refill
            local tokens_to_add = elapsed * refill_rate
            tokens = math.min(capacity, tokens + tokens_to_add)
            
            local allowed = 0
            if tokens >= 1 then
                allowed = 1
                tokens = tokens - 1
            end
            
            -- Store updated state
            redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
            redis.call('EXPIRE', key, 3600)  -- TTL 1 hour
            
            return {allowed, tokens}
        """
        
        allowed, tokens = self.redis.eval(
            lua_script,
            1,
            bucket_key,
            now,
            self.capacity,
            self.refill_rate
        )
        return allowed == 1, tokens

    def get_token_count(self, key):
        """Get current token count for debugging."""
        bucket_key = f"rate_limit_token:{key}"
        tokens = self.redis.hget(bucket_key, 'tokens')
        return float(tokens) if tokens else self.capacity

import time
import redis

class SlidingWindowCounterRateLimiter:
    def __init__(self, redis_client, limit_per_window, window_seconds):
        self.redis = redis_client
        self.limit = limit_per_window
        self.window_seconds = window_seconds
    
    def allow_request(self, key):
        """
        Sliding window counter using two Redis keys (current + previous windows).
        Memory O(1) per key. Precision ~1% vs sliding window log.
        Uses Redis TIME for consistent cluster-wide timestamp.
        """
        lua_script = """
            local key = KEYS[1]
            local limit = tonumber(ARGV[1])
            local window_seconds = tonumber(ARGV[2])
            local now = tonumber(ARGV[3])
            
            local window_start = now - window_seconds
            -- Each window is 1 second (or window_seconds, but we use 1-second granularity)
            -- Calculate current and previous window identifiers
            local current_window = math.floor(now)
            local previous_window = math.floor(window_start)
            
            local current_key = key .. ':' .. current_window
            local previous_key = key .. ':' .. previous_window
            
            -- Get counts for both windows
            local current_count = redis.call('GET', current_key) or 0
            local previous_count = redis.call('GET', previous_key) or 0
            
            -- Calculate weighted count: previous window weight * (overlap / window_seconds)
            local overlap = window_seconds - (now - previous_window)
            local weight = overlap / window_seconds
            local estimated_count = previous_count * weight + current_count
            
            if estimated_count < limit then
                -- Increment current window counter
                local new_count = redis.call('INCR', current_key)
                if new_count == 1 then
                    redis.call('EXPIRE', current_key, window_seconds * 2)
                end
                return {1, new_count}
            else
                return {0, estimated_count}
            end
        """
        
        # Get consistent timestamp from Redis (not local time)
        redis_time = self.redis.time()  # returns [seconds, microseconds]
        now = redis_time[0] + redis_time[1] / 1000000.0
        
        allowed, count = self.redis.eval(
            lua_script,
            1,
            f"rate_limit_sliding:{key}",
            self.limit,
            self.window_seconds,
            now
        )
        return allowed == 1, count

    def get_estimated_usage(self, key):
        """Get current estimated count without consuming a request."""
        # Similar to allow_request but without INCR
        # ... (omitted for brevity)
        pass

// io.thecodeforge — system-design tutorial

import time

class EdgeRateLimiter:
    """Quick kill at the gateway — IP-based, no business logic."""
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.redis = RedisClient()  # pretend

    def allow_request(self, ip: str) -> bool:
        key = f"ratelimit:edge:{ip}"
        count = self.redis.incr(key)
        if count == 1:
            self.redis.expire(key, self.window_seconds)
        return count <= self.max_requests

class AppRateLimiter:
    """Fine-grained per-user throttle at the service layer."""
    def __init__(self):
        self.store = {}

    def check(self, user_id: str, quota: int, window: int) -> bool:
        now = time.time()
        if user_id not in self.store:
            self.store[user_id] = []
        timestamps = self.store[user_id]
        while timestamps and timestamps[0] < now - window:
            timestamps.pop(0)
        if len(timestamps) >= quota:
            return False
        timestamps.append(now)
        return True

// io.thecodeforge — system-design tutorial

import redis
import time

class RateLimiterDecisionPoint:
    def __init__(self, redis_host: str):
        self.store = redis.StrictRedis(host=redis_host, decode_responses=True)
        self.rule_cache = {}  # refreshed periodically

    def is_allowed(self, user_id: str, api_key: str) -> bool:
        rule = self.rule_cache.get(api_key)
        if not rule:
            rule = self._fetch_rule(api_key)
        if not rule:
            return True  # fail open if rule missing

        key = f"ratelimit:counter:{user_id}:{api_key}"
        current = self.store.get(key)
        if current and int(current) >= rule['quota']:
            return False

        pipeline = self.store.pipeline()
        pipeline.incr(key)
        pipeline.expire(key, rule['window_seconds'])
        pipeline.execute()
        return True

    def _fetch_rule(self, api_key: str) -> dict:
        # pretend DB call, cache for 60s
        return {'quota': 10, 'window_seconds': 60}

// io.thecodeforge — system-design tutorial

from typing import Protocol, Dict

class RateLimitRule:
    def __init__(self, client_id: str, endpoint: str, max_requests: int, window_seconds: int):
        self.client_id = client_id
        self.endpoint = endpoint
        self.max_requests = max_requests
        self.window_seconds = window_seconds

class RateLimiter(Protocol):
    def is_allowed(self, client_id: str, endpoint: str) -> bool:
        ...
    def remaining(self, client_id: str, endpoint: str) -> int:
        ...
    def reset_time(self, client_id: str, endpoint: str) -> int:
        ...

// io.thecodeforge — system-design tutorial

class NonFunctionalSpec:
    P99_LATENCY_MS = 5
    THROUGHPUT_RPS = 100_000
    AVAILABILITY_SLA = 0.9999
    CONSISTENCY_WINDOW_MS = 100
    MAX_MEMORY_PER_CLIENT = 512  # bytes for sliding window entries