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