Rate Limiting Explained: Components, Algorithms & Real-World Patterns

import time
import threading
from dataclasses import dataclass, field
from typing import Dict, Tuple

# ─── COMPONENT 1: Rule Store ───────────────────────────────────────────────
# In production this would be loaded from a config file or database.
# Here we define it as a plain dict for clarity.
RATE_LIMIT_RULES: Dict[str, Tuple[int, int]] = {
    # tier_name: (max_requests, window_seconds)
    "free":       (10, 60),    # 10 requests per minute
    "paid":       (100, 60),   # 100 requests per minute
    "enterprise": (1000, 60),  # 1000 requests per minute
}

# ─── COMPONENT 2: Counter Store (in-memory, thread-safe) ───────────────────
# In a distributed system, replace this with a Redis INCR + EXPIRE call.
@dataclass
class WindowCounter:
    count: int = 0
    window_start: float = field(default_factory=time.time)

class InMemoryCounterStore:
    def __init__(self):
        self._store: Dict[str, WindowCounter] = {}
        self._lock = threading.Lock()  # Prevents race conditions in multi-threaded apps

    def increment_and_get(self, client_id: str, window_seconds: int) -> Tuple[int, float]:
        """Atomically increments the counter and returns (current_count, window_start)."""
        with self._lock:
            now = time.time()
            counter = self._store.get(client_id)

            # If no counter exists, or the current window has expired, start fresh
            if counter is None or (now - counter.window_start) >= window_seconds:
                self._store[client_id] = WindowCounter(count=1, window_start=now)
                return 1, now

            counter.count += 1
            return counter.count, counter.window_start

# ─── COMPONENT 3: Decision Engine (Fixed Window algorithm) ─────────────────
class RateLimitDecision:
    ALLOW   = "ALLOW"
    THROTTLE = "THROTTLE"
    REJECT  = "REJECT"

def make_decision(
    client_id: str,
    tier: str,
    counter_store: InMemoryCounterStore
) -> dict:
    """Core algorithm: reads state, applies rule, returns a verdict."""
    max_requests, window_seconds = RATE_LIMIT_RULES[tier]  # Rule Store lookup

    current_count, window_start = counter_store.increment_and_get(
        client_id, window_seconds
    )

    remaining = max(0, max_requests - current_count)
    window_resets_at = window_start + window_seconds
    retry_after = max(0, int(window_resets_at - time.time()))

    if current_count <= max_requests:
        verdict = RateLimitDecision.ALLOW
    else:
        verdict = RateLimitDecision.REJECT

    return {
        "verdict":        verdict,
        "limit":          max_requests,
        "remaining":      remaining,
        "retry_after_s":  retry_after if verdict == RateLimitDecision.REJECT else 0,
    }

# ─── COMPONENT 4: Response Handler ─────────────────────────────────────────
def handle_request(client_id: str, tier: str, counter_store: InMemoryCounterStore):
    """Simulates an HTTP layer: decides, then formats the response."""
    decision = make_decision(client_id, tier, counter_store)

    if decision["verdict"] == RateLimitDecision.ALLOW:
        # Attach rate limit headers to every successful response
        print(f"[200 OK]     client={client_id} "
              f"X-RateLimit-Remaining={decision['remaining']} "
              f"X-RateLimit-Limit={decision['limit']}")
    else:
        # HTTP 429 Too Many Requests — tell the client exactly when to retry
        print(f"[429 REJECT] client={client_id} "
              f"Retry-After={decision['retry_after_s']}s "
              f"X-RateLimit-Limit={decision['limit']}")

# ─── DEMO ───────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    store = InMemoryCounterStore()

    print("=== Free tier client firing 12 requests (limit is 10/min) ===")
    for request_number in range(1, 13):
        print(f"  Request #{request_number}: ", end="")
        handle_request(client_id="user_free_42", tier="free", counter_store=store)

import time
import threading
from dataclasses import dataclass

# Token Bucket Algorithm
# WHY use this? It allows short bursts of traffic while enforcing a
# long-term average rate. Stripe, GitHub, and Twilio all use variants of this.

@dataclass
class TokenBucket:
    capacity: int          # Maximum tokens the bucket can hold (burst limit)
    refill_rate: float     # Tokens added per second (steady-state limit)
    _tokens: float = 0.0
    _last_refill_time: float = 0.0
    _lock: threading.Lock = None

    def __post_init__(self):
        self._tokens = float(self.capacity)  # Start full — first burst is always allowed
        self._last_refill_time = time.monotonic()  # monotonic is safer than time.time() for intervals
        self._lock = threading.Lock()

    def _refill(self):
        """Called before every consume() to add tokens earned since last request."""
        now = time.monotonic()
        elapsed_seconds = now - self._last_refill_time

        # How many tokens have we earned in the time since the last request?
        tokens_earned = elapsed_seconds * self.refill_rate

        # Cap at capacity — you can't save up more than the bucket holds
        self._tokens = min(self.capacity, self._tokens + tokens_earned)
        self._last_refill_time = now

    def consume(self, tokens_needed: int = 1) -> bool:
        """Try to consume tokens. Returns True if allowed, False if rate-limited."""
        with self._lock:  # Thread-safe — critical in async/multi-threaded servers
            self._refill()

            if self._tokens >= tokens_needed:
                self._tokens -= tokens_needed
                return True  # ALLOW
            return False     # REJECT — not enough tokens

    @property
    def tokens_available(self) -> float:
        """Read current token count (approximate — for logging only)."""
        with self._lock:
            self._refill()
            return round(self._tokens, 2)


def simulate_api_traffic():
    """
    Simulates a realistic traffic pattern:
    - An initial burst (legitimate, e.g. app startup)
    - Steady traffic
    - Another burst that partially hits the limit
    """
    # 5 tokens max (burst), refills at 1 token/second (steady-state: 1 req/sec)
    bucket = TokenBucket(capacity=5, refill_rate=1.0)

    print("=== Simulating Token Bucket Rate Limiter ===")
    print(f"Config: capacity=5 tokens, refill=1 token/second\n")

    # Phase 1: Burst of 7 requests (only 5 should pass — bucket starts full at 5)
    print("[Phase 1] Burst of 7 requests fired instantly:")
    for req_num in range(1, 8):
        allowed = bucket.consume()
        status = "ALLOW ✓" if allowed else "REJECT ✗"
        print(f"  Request #{req_num}: {status} | Tokens left: {bucket.tokens_available}")

    # Phase 2: Wait 3 seconds — bucket refills 3 tokens
    print("\n[Phase 2] Waiting 3 seconds for bucket to refill...")
    time.sleep(3)
    print(f"  Tokens after 3s wait: {bucket.tokens_available}")

    # Phase 3: Send 4 more requests — 3 should pass, 1 should be rejected
    print("\n[Phase 3] Sending 4 more requests after refill:")
    for req_num in range(8, 12):
        allowed = bucket.consume()
        status = "ALLOW ✓" if allowed else "REJECT ✗"
        print(f"  Request #{req_num}: {status} | Tokens left: {bucket.tokens_available}")


if __name__ == "__main__":
    simulate_api_traffic()

# Requires: pip install redis
# Assumes Redis running on localhost:6379
# Run: docker run -p 6379:6379 redis

import redis
import time
from typing import Tuple

# ─── Redis connection ────────────────────────────────────────────────────────
# In production, use connection pooling and a Redis Sentinel or Cluster setup
redis_client = redis.Redis(
    host="localhost",
    port=6379,
    db=0,
    decode_responses=True  # Return strings, not bytes
)

class DistributedFixedWindowLimiter:
    """
    Uses Redis INCR + EXPIRE for atomic, distributed-safe counting.
    WHY INCR? It's a single atomic command — no race condition between
    'read current count' and 'write new count' that you'd get in application code.
    """

    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds

    def _make_redis_key(self, client_id: str) -> str:
        """
        Key includes the current window start time so it automatically
        becomes a different key (and thus resets) each new window.
        """
        window_number = int(time.time()) // self.window_seconds
        return f"rate_limit:{client_id}:{window_number}"

    def is_allowed(self, client_id: str) -> Tuple[bool, dict]:
        """
        Returns (is_allowed, metadata_dict).
        Uses a Redis pipeline to batch INCR and EXPIRE into one round-trip.
        """
        key = self._make_redis_key(client_id)

        # Pipeline batches commands — reduces network round-trips from 2 to 1
        pipe = redis_client.pipeline()
        pipe.incr(key)                          # Atomically increment counter
        pipe.expire(key, self.window_seconds)   # Auto-delete key when window expires
        results = pipe.execute()

        current_count = results[0]  # INCR returns the new value after incrementing
        remaining = max(0, self.max_requests - current_count)
        allowed = current_count <= self.max_requests

        metadata = {
            "limit":     self.max_requests,
            "remaining": remaining,
            "count":     current_count,
            "key":       key,
        }
        return allowed, metadata


class DistributedSlidingWindowLimiter:
    """
    Uses a Redis Sorted Set for accurate sliding window counting.
    WHY sorted set? Each member is a request timestamp. ZCOUNT lets us
    query 'how many requests in the last N seconds' with O(log N) complexity.
    """

    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds

    def is_allowed(self, client_id: str) -> Tuple[bool, dict]:
        now = time.time()
        window_start = now - self.window_seconds
        key = f"sliding_rate_limit:{client_id}"

        pipe = redis_client.pipeline()
        # Remove timestamps older than our window (keeps memory bounded)
        pipe.zremrangebyscore(key, 0, window_start)
        # Add current request timestamp as both score and member (unique via timestamp+random suffix)
        pipe.zadd(key, {str(now): now})
        # Count all requests within the window
        pipe.zcount(key, window_start, now)
        # Expire the whole key after the window — cleanup if client goes quiet
        pipe.expire(key, self.window_seconds)
        results = pipe.execute()

        current_count = results[2]  # ZCOUNT result is at index 2
        remaining = max(0, self.max_requests - current_count)
        allowed = current_count <= self.max_requests

        return allowed, {"limit": self.max_requests, "remaining": remaining, "count": current_count}


def demo_distributed_limiters():
    fixed_limiter   = DistributedFixedWindowLimiter(max_requests=5, window_seconds=60)
    sliding_limiter = DistributedSlidingWindowLimiter(max_requests=5, window_seconds=60)

    # Clean up any leftover keys from previous runs
    for k in redis_client.scan_iter("rate_limit:demo_user*"):
        redis_client.delete(k)
    for k in redis_client.scan_iter("sliding_rate_limit:demo_user*"):
        redis_client.delete(k)

    print("=== Fixed Window (Redis INCR) — 7 requests, limit 5 ===")
    for req_num in range(1, 8):
        allowed, meta = fixed_limiter.is_allowed("demo_user_1")
        status = "ALLOW ✓" if allowed else "REJECT ✗"
        print(f"  #{req_num}: {status} | count={meta['count']} remaining={meta['remaining']}")

    print("\n=== Sliding Window (Redis Sorted Set) — 7 requests, limit 5 ===")
    for req_num in range(1, 8):
        allowed, meta = sliding_limiter.is_allowed("demo_user_2")
        status = "ALLOW ✓" if allowed else "REJECT ✗"
        print(f"  #{req_num}: {status} | count={meta['count']} remaining={meta['remaining']}")


if __name__ == "__main__":
    demo_distributed_limiters()