# 600 Reference 68991c98

> API Rate Limiting Reference

- Skill: `tools-only/600-reference-68991c98` (Agent Skill, multi-file: 3 files)
- Install (CLI): `npx skillmds@latest add tools-only/600-reference-68991c98`
- Raw SKILL.md: https://api.skillmd.com/api/skills/tools-only/600-reference-68991c98/raw
- Safety review: pending (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: tools-only (https://skillmd.com/u/tools-only)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/tools-only/600-reference-68991c98

---

# API Rate Limiting Reference

> **Comprehensive technical reference for rate limiting algorithms, implementations, and best practices**

## Table of Contents

1. [Fundamentals](#fundamentals)
2. [Rate Limiting Algorithms](#rate-limiting-algorithms)
3. [Distributed Rate Limiting](#distributed-rate-limiting)
4. [HTTP Headers and Standards](#http-headers-and-standards)
5. [Storage Backends](#storage-backends)
6. [Implementation Patterns](#implementation-patterns)
7. [Limiting Strategies](#limiting-strategies)
8. [Error Handling and Responses](#error-handling-and-responses)
9. [Performance and Scalability](#performance-and-scalability)
10. [Security Considerations](#security-considerations)
11. [Testing and Monitoring](#testing-and-monitoring)
12. [Common Anti-Patterns](#common-anti-patterns)
13. [Language-Specific Implementations](#language-specific-implementations)
14. [References and Standards](#references-and-standards)

---

## Fundamentals

### What is Rate Limiting?

**Rate limiting** is a technique for controlling the rate at which clients can make requests to an API or service. It prevents abuse, ensures fair resource allocation, and protects services from being overwhelmed.

### Why Rate Limit?

**Primary objectives**:
1. **Prevent abuse**: Stop malicious actors from overwhelming your service
2. **Ensure fair usage**: Distribute resources equitably among users
3. **Control costs**: Limit expensive operations (database queries, external API calls)
4. **Maintain stability**: Prevent cascading failures under high load
5. **Business monetization**: Differentiate service tiers

### Rate Limiting vs Throttling

```
RATE LIMITING
├─ Hard limits enforced
├─ Requests rejected when exceeded
├─ Clear quotas communicated
└─ 429 Too Many Requests response

THROTTLING
├─ Soft limits enforced
├─ Requests slowed/delayed
├─ Queuing mechanisms
└─ May not return errors
```

### Key Metrics

**Requests Per Second (RPS)**:
```
RPS = Total Requests / Time Period (seconds)

Example: 100 requests in 10 seconds = 10 RPS
```

**Burst Capacity**:
- Maximum requests allowed in short time window
- Important for handling traffic spikes
- Typically higher than sustained rate

**Window Duration**:
- Time period for counting requests
- Common: 1 second, 1 minute, 1 hour, 1 day
- Trade-off: Shorter = smoother, Longer = simpler

---

## Rate Limiting Algorithms

### Token Bucket Algorithm

**Description**: Tokens added to bucket at fixed rate. Each request consumes tokens. Request rejected if insufficient tokens available.

**Characteristics**:
- Allows bursts (up to bucket capacity)
- Smooth average rate over time
- Good for credit-based systems

**Parameters**:
- `capacity`: Maximum tokens in bucket
- `refill_rate`: Tokens added per second
- `tokens_per_request`: Tokens consumed per request (usually 1)

**Mathematical Model**:
```
tokens(t) = min(capacity, tokens(t-1) + refill_rate * Δt)

allowed = tokens(t) >= tokens_needed
if allowed:
    tokens(t) = tokens(t) - tokens_needed
```

**Python Implementation**:
```python
import time
from typing import Optional

class TokenBucket:
    """Token bucket rate limiter

    Allows bursts while maintaining average rate.
    """

    def __init__(self, capacity: int, refill_rate: float):
        """
        Args:
            capacity: Maximum tokens in bucket
            refill_rate: Tokens added per second
        """
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = capacity
        self.last_refill = time.time()
        self._lock = threading.Lock()

    def _refill(self) -> None:
        """Refill tokens based on elapsed time"""
        now = time.time()
        elapsed = now - self.last_refill

        # Add tokens for time elapsed
        tokens_to_add = elapsed * self.refill_rate
        self.tokens = min(self.capacity, self.tokens + tokens_to_add)
        self.last_refill = now

    def consume(self, tokens: int = 1) -> bool:
        """Attempt to consume tokens

        Args:
            tokens: Number of tokens to consume

        Returns:
            True if tokens consumed, False if insufficient
        """
        with self._lock:
            self._refill()

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True

            return False

    def get_tokens(self) -> float:
        """Get current token count"""
        with self._lock:
            self._refill()
            return self.tokens

    def wait_time(self, tokens: int = 1) -> float:
        """Calculate wait time for tokens

        Returns:
            Seconds to wait until tokens available
        """
        with self._lock:
            self._refill()

            if self.tokens >= tokens:
                return 0.0

            deficit = tokens - self.tokens
            return deficit / self.refill_rate

# Usage example
limiter = TokenBucket(capacity=100, refill_rate=10)  # 10 tokens/sec

if limiter.consume():
    # Process request
    response = handle_request()
else:
    # Rate limited
    wait = limiter.wait_time()
    return error_response(f"Rate limited. Try again in {wait:.1f} seconds", 429)
```

**Distributed Implementation (Redis + Lua)**:
```python
import redis
import time

class DistributedTokenBucket:
    """Distributed token bucket using Redis"""

    def __init__(self, redis_client: redis.Redis, key: str,
                 capacity: int, refill_rate: float):
        self.redis = redis_client
        self.key = f"rate_limit:token_bucket:{key}"
        self.capacity = capacity
        self.refill_rate = refill_rate

        # Lua script for atomic token bucket operations
        self.script = """
        local key = KEYS[1]
        local capacity = tonumber(ARGV[1])
        local refill_rate = tonumber(ARGV[2])
        local tokens_needed = tonumber(ARGV[3])
        local now = tonumber(ARGV[4])

        -- Get current state
        local state = redis.call('HMGET', key, 'tokens', 'last_refill')
        local tokens = tonumber(state[1])
        local last_refill = tonumber(state[2])

        -- Initialize if needed
        if not tokens then
            tokens = capacity
            last_refill = now
        end

        -- Refill tokens
        local elapsed = now - last_refill
        tokens = math.min(capacity, tokens + (elapsed * refill_rate))

        -- Check if enough tokens
        if tokens >= tokens_needed then
            tokens = tokens - tokens_needed
            redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
            redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) * 2)
            return {1, tokens}  -- allowed, remaining
        else
            return {0, tokens}  -- denied, remaining
        end
        """

        self.script_sha = self.redis.script_load(self.script)

    def consume(self, tokens: int = 1) -> tuple[bool, float]:
        """Attempt to consume tokens

        Returns:
            (allowed, remaining_tokens)
        """
        now = time.time()

        try:
            result = self.redis.evalsha(
                self.script_sha,
                1,  # number of keys
                self.key,
                self.capacity,
                self.refill_rate,
                tokens,
                now
            )

            allowed = bool(result[0])
            remaining = float(result[1])

            return allowed, remaining

        except redis.exceptions.NoScriptError:
            # Script not cached, reload
            self.script_sha = self.redis.script_load(self.script)
            return self.consume(tokens)

# Usage
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
limiter = DistributedTokenBucket(redis_client, "user:123", capacity=100, refill_rate=10)

allowed, remaining = limiter.consume()
if allowed:
    print(f"Request allowed. {remaining:.1f} tokens remaining")
else:
    print(f"Rate limited. {remaining:.1f} tokens available")
```

**Best For**:
- APIs with bursty traffic patterns
- Credit-based systems
- Flexible rate limiting with burst tolerance

**Pros**:
- Simple to understand and implement
- Allows bursts while maintaining average rate
- Low memory footprint
- Fast operations

**Cons**:
- Allows bursts (may not be desired)
- Slightly more complex than fixed window

---

### Leaky Bucket Algorithm

**Description**: Requests enter a queue (bucket). Processed at fixed rate (leak). Requests rejected if queue full.

**Characteristics**:
- Smooths traffic to constant rate
- No bursts allowed
- Queue-based approach

**Parameters**:
- `capacity`: Maximum queue size
- `leak_rate`: Requests processed per second

**Mathematical Model**:
```
queue_size(t) = max(0, queue_size(t-1) - leak_rate * Δt)

if queue_size(t) < capacity:
    queue_size(t) = queue_size(t) + 1
    allowed = true
else:
    allowed = false
```

**Python Implementation**:
```python
import time
import threading
from collections import deque

class LeakyBucket:
    """Leaky bucket rate limiter

    Processes requests at constant rate.
    """

    def __init__(self, capacity: int, leak_rate: float):
        """
        Args:
            capacity: Maximum queue size
            leak_rate: Requests processed per second
        """
        self.capacity = capacity
        self.leak_rate = leak_rate
        self.queue_size = 0.0
        self.last_leak = time.time()
        self._lock = threading.Lock()

    def _leak(self) -> None:
        """Process (leak) requests based on elapsed time"""
        now = time.time()
        elapsed = now - self.last_leak

        # Leak (process) requests
        leaked = elapsed * self.leak_rate
        self.queue_size = max(0.0, self.queue_size - leaked)
        self.last_leak = now

    def allow_request(self) -> bool:
        """Check if request can be queued

        Returns:
            True if space available in queue
        """
        with self._lock:
            self._leak()

            if self.queue_size < self.capacity:
                self.queue_size += 1.0
                return True

            return False

    def get_queue_size(self) -> float:
        """Get current queue size"""
        with self._lock:
            self._leak()
            return self.queue_size

# Usage
limiter = LeakyBucket(capacity=50, leak_rate=5)  # 5 requests/sec

if limiter.allow_request():
    # Queue request
    response = handle_request()
else:
    # Queue full
    return error_response("Rate limited. Queue full.", 429)
```

**Distributed Implementation (Redis)**:
```python
import redis
import time

class DistributedLeakyBucket:
    """Distributed leaky bucket using Redis"""

    def __init__(self, redis_client: redis.Redis, key: str,
                 capacity: int, leak_rate: float):
        self.redis = redis_client
        self.key = f"rate_limit:leaky_bucket:{key}"
        self.capacity = capacity
        self.leak_rate = leak_rate

        self.script = """
        local key = KEYS[1]
        local capacity = tonumber(ARGV[1])
        local leak_rate = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])

        local state = redis.call('HMGET', key, 'size', 'last_leak')
        local size = tonumber(state[1]) or 0
        local last_leak = tonumber(state[2]) or now

        -- Leak based on time elapsed
        local elapsed = now - last_leak
        local leaked = elapsed * leak_rate
        size = math.max(0, size - leaked)

        -- Check if space available
        if size < capacity then
            size = size + 1
            redis.call('HMSET', key, 'size', size, 'last_leak', now)
            redis.call('EXPIRE', key, math.ceil(capacity / leak_rate) * 2)
            return {1, capacity - size}  -- allowed, remaining
        else
            return {0, 0}  -- denied
        end
        """

        self.script_sha = self.redis.script_load(self.script)

    def allow_request(self) -> tuple[bool, float]:
        """Check if request can be queued

        Returns:
            (allowed, remaining_capacity)
        """
        now = time.time()

        try:
            result = self.redis.evalsha(
                self.script_sha,
                1,
                self.key,
                self.capacity,
                self.leak_rate,
                now
            )

            return bool(result[0]), float(result[1])

        except redis.exceptions.NoScriptError:
            self.script_sha = self.redis.script_load(self.script)
            return self.allow_request()
```

**Best For**:
- Protecting downstream services
- Guaranteed constant load
- Queue-based processing

**Pros**:
- Smooth, predictable rate
- No bursts
- Protects downstream services

**Cons**:
- No burst tolerance
- May delay requests
- More complex than token bucket

---

### Fixed Window Algorithm

**Description**: Time divided into fixed windows. Counter increments per request. Counter resets at window boundary.

**Characteristics**:
- Simple to implement
- Easy to understand
- Allows bursts at boundaries

**Parameters**:
- `limit`: Maximum requests per window
- `window_duration`: Length of time window (seconds)

**Mathematical Model**:
```
window_id = floor(timestamp / window_duration)
count[window_id] = count[window_id] + 1

allowed = count[window_id] <= limit
```

**Problem: Boundary Burst**:
```
Window 1 (11:59:00 - 11:59:59): 100 requests at 11:59:59
Window 2 (12:00:00 - 12:00:59): 100 requests at 12:00:00

Result: 200 requests in 1 second (burst at boundary)
```

**Python Implementation**:
```python
import time
import threading
from collections import defaultdict

class FixedWindow:
    """Fixed window rate limiter

    Simple per-period limits with boundary bursts.
    """

    def __init__(self, limit: int, window_seconds: int):
        """
        Args:
            limit: Maximum requests per window
            window_seconds: Duration of window in seconds
        """
        self.limit = limit
        self.window_seconds = window_seconds
        self.counters = defaultdict(int)
        self._lock = threading.Lock()

    def _get_window_id(self) -> int:
        """Get current window identifier"""
        return int(time.time() // self.window_seconds)

    def allow_request(self) -> bool:
        """Check if request is allowed

        Returns:
            True if under limit, False otherwise
        """
        with self._lock:
            window_id = self._get_window_id()

            # Clean old windows (optional optimization)
            old_windows = [wid for wid in self.counters if wid < window_id - 1]
            for wid in old_windows:
                del self.counters[wid]

            # Check limit
            if self.counters[window_id] < self.limit:
                self.counters[window_id] += 1
                return True

            return False

    def get_remaining(self) -> int:
        """Get remaining requests in current window"""
        with self._lock:
            window_id = self._get_window_id()
            count = self.counters[window_id]
            return max(0, self.limit - count)

    def get_reset_time(self) -> int:
        """Get timestamp when window resets"""
        window_id = self._get_window_id()
        return int((window_id + 1) * self.window_seconds)

# Usage
limiter = FixedWindow(limit=100, window_seconds=60)  # 100 req/min

if limiter.allow_request():
    response = handle_request()
else:
    reset_time = limiter.get_reset_time()
    retry_after = reset_time - int(time.time())
    return error_response(f"Rate limited. Try in {retry_after}s", 429)
```

**Distributed Implementation (Redis)**:
```python
import redis
import time

class DistributedFixedWindow:
    """Distributed fixed window using Redis"""

    def __init__(self, redis_client: redis.Redis, key: str,
                 limit: int, window_seconds: int):
        self.redis = redis_client
        self.key_prefix = f"rate_limit:fixed_window:{key}"
        self.limit = limit
        self.window_seconds = window_seconds

    def _get_window_key(self) -> str:
        """Get Redis key for current window"""
        window_id = int(time.time() // self.window_seconds)
        return f"{self.key_prefix}:{window_id}"

    def allow_request(self) -> tuple[bool, int, int]:
        """Check if request is allowed

        Returns:
            (allowed, remaining, reset_time)
        """
        key = self._get_window_key()

        # Increment counter atomically
        pipe = self.redis.pipeline()
        pipe.incr(key)
        pipe.expire(key, self.window_seconds * 2)  # TTL safety margin
        results = pipe.execute()

        count = results[0]
        allowed = count <= self.limit
        remaining = max(0, self.limit - count)

        # Calculate reset time
        window_id = int(time.time() // self.window_seconds)
        reset_time = int((window_id + 1) * self.window_seconds)

        return allowed, remaining, reset_time

    def get_remaining(self) -> int:
        """Get remaining requests in current window"""
        key = self._get_window_key()
        count = int(self.redis.get(key) or 0)
        return max(0, self.limit - count)

# Usage
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
limiter = DistributedFixedWindow(redis_client, "user:123", limit=100, window_seconds=60)

allowed, remaining, reset_time = limiter.allow_request()
if allowed:
    # Add headers
    response.headers['X-RateLimit-Limit'] = str(limiter.limit)
    response.headers['X-RateLimit-Remaining'] = str(remaining)
    response.headers['X-RateLimit-Reset'] = str(reset_time)
else:
    retry_after = reset_time - int(time.time())
    return error_response(f"Rate limited. Reset in {retry_after}s", 429)
```

**Best For**:
- Simple quota enforcement
- Easy user communication
- Low complexity requirements

**Pros**:
- Extremely simple
- Low memory usage
- Fast operations
- Easy to explain to users

**Cons**:
- Allows boundary bursts (2x limit in 1 second)
- Not smooth rate limiting

---

### Sliding Window Algorithm

**Description**: Track request timestamps. Count requests in rolling time window. Remove old requests outside window.

**Characteristics**:
- Smooth rate limiting
- No boundary bursts
- Higher memory usage

**Parameters**:
- `limit`: Maximum requests in window
- `window_duration`: Rolling window size (seconds)

**Mathematical Model**:
```
window_start = current_time - window_duration
requests_in_window = count(requests where timestamp >= window_start)

allowed = requests_in_window < limit
```

**Python Implementation**:
```python
import time
import threading
from collections import deque

class SlidingWindow:
    """Sliding window rate limiter

    Smooth rate limiting without boundary bursts.
    """

    def __init__(self, limit: int, window_seconds: int):
        """
        Args:
            limit: Maximum requests in window
            window_seconds: Rolling window duration
        """
        self.limit = limit
        self.window_seconds = window_seconds
        self.requests = deque()
        self._lock = threading.Lock()

    def _clean_old_requests(self, now: float) -> None:
        """Remove requests outside window"""
        window_start = now - self.window_seconds

        while self.requests and self.requests[0] < window_start:
            self.requests.popleft()

    def allow_request(self) -> bool:
        """Check if request is allowed

        Returns:
            True if under limit, False otherwise
        """
        with self._lock:
            now = time.time()
            self._clean_old_requests(now)

            if len(self.requests) < self.limit:
                self.requests.append(now)
                return True

            return False

    def get_remaining(self) -> int:
        """Get remaining requests in window"""
        with self._lock:
            now = time.time()
            self._clean_old_requests(now)
            return max(0, self.limit - len(self.requests))

    def get_oldest_request_time(self) -> float:
        """Get timestamp of oldest request in window"""
        with self._lock:
            now = time.time()
            self._clean_old_requests(now)

            if self.requests:
                return self.requests[0]
            return now

# Usage
limiter = SlidingWindow(limit=100, window_seconds=60)

if limiter.allow_request():
    response = handle_request()
else:
    # Calculate when next request allowed
    oldest = limiter.get_oldest_request_time()
    retry_after = int(oldest + limiter.window_seconds - time.time())
    return error_response(f"Rate limited. Try in {retry_after}s", 429)
```

**Distributed Implementation (Redis Sorted Sets)**:
```python
import redis
import time
import uuid

class DistributedSlidingWindow:
    """Distributed sliding window using Redis sorted sets"""

    def __init__(self, redis_client: redis.Redis, key: str,
                 limit: int, window_seconds: int):
        self.redis = redis_client
        self.key = f"rate_limit:sliding_window:{key}"
        self.limit = limit
        self.window_seconds = window_seconds

        # Lua script for atomic operations
        self.script = """
        local key = KEYS[1]
        local limit = tonumber(ARGV[1])
        local window = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])
        local request_id = ARGV[4]

        local window_start = now - window

        -- Remove old requests
        redis.call('ZREMRANGEBYSCORE', key, 0, window_start)

        -- Count requests in window
        local count = redis.call('ZCARD', key)

        if count < limit then
            -- Add current request
            redis.call('ZADD', key, now, request_id)
            redis.call('EXPIRE', key, window * 2)
            return {1, limit - count - 1}  -- allowed, remaining
        else
            return {0, 0}  -- denied
        end
        """

        self.script_sha = self.redis.script_load(self.script)

    def allow_request(self) -> tuple[bool, int]:
        """Check if request is allowed

        Returns:
            (allowed, remaining)
        """
        now = time.time()
        request_id = str(uuid.uuid4())

        try:
            result = self.redis.evalsha(
                self.script_sha,
                1,
                self.key,
                self.limit,
                self.window_seconds,
                now,
                request_id
            )

            return bool(result[0]), int(result[1])

        except redis.exceptions.NoScriptError:
            self.script_sha = self.redis.script_load(self.script)
            return self.allow_request()

    def get_remaining(self) -> int:
        """Get remaining requests in window"""
        now = time.time()
        window_start = now - self.window_seconds

        # Clean old requests
        self.redis.zremrangebyscore(self.key, 0, window_start)

        # Count remaining
        count = self.redis.zcard(self.key)
        return max(0, self.limit - count)

    def get_oldest_request(self) -> float:
        """Get timestamp of oldest request in window"""
        now = time.time()
        window_start = now - self.window_seconds

        # Clean old requests
        self.redis.zremrangebyscore(self.key, 0, window_start)

        # Get oldest
        oldest = self.redis.zrange(self.key, 0, 0, withscores=True)
        if oldest:
            return float(oldest[0][1])
        return now

# Usage
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
limiter = DistributedSlidingWindow(redis_client, "user:123", limit=100, window_seconds=60)

allowed, remaining = limiter.allow_request()
if allowed:
    print(f"Request allowed. {remaining} remaining")
else:
    oldest = limiter.get_oldest_request()
    retry_after = int(oldest + limiter.window_seconds - time.time())
    print(f"Rate limited. Try in {retry_after} seconds")
```

**Best For**:
- Fair rate limiting
- Preventing boundary bursts
- High-accuracy requirements

**Pros**:
- No boundary bursts
- Smooth rate limiting
- Fair to all users
- Accurate limiting

**Cons**:
- Higher memory usage (stores timestamps)
- More complex implementation
- Slower than fixed window

---

### Sliding Window Counter (Hybrid)

**Description**: Combines fixed window simplicity with sliding window accuracy. Uses weighted counts from current and previous windows.

**Mathematical Model**:
```
current_window_count = requests in current window
previous_window_count = requests in previous window

window_progress = (current_time % window_duration) / window_duration
estimated_count = (previous_window_count * (1 - window_progress)) + current_window_count

allowed = estimated_count < limit
```

**Python Implementation**:
```python
import time
import threading
from collections import defaultdict

class SlidingWindowCounter:
    """Sliding window counter (hybrid approach)

    Approximates sliding window with fixed window simplicity.
    """

    def __init__(self, limit: int, window_seconds: int):
        """
        Args:
            limit: Maximum requests in window
            window_seconds: Window duration
        """
        self.limit = limit
        self.window_seconds = window_seconds
        self.counters = defaultdict(int)
        self._lock = threading.Lock()

    def _get_window_id(self, timestamp: float) -> int:
        """Get window ID for timestamp"""
        return int(timestamp // self.window_seconds)

    def _estimate_count(self, now: float) -> float:
        """Estimate request count in sliding window"""
        current_window = self._get_window_id(now)
        previous_window = current_window - 1

        # Get counts
        current_count = self.counters[current_window]
        previous_count = self.counters[previous_window]

        # Calculate weight for previous window
        window_progress = (now % self.window_seconds) / self.window_seconds
        previous_weight = 1.0 - window_progress

        # Weighted sum
        return (previous_count * previous_weight) + current_count

    def allow_request(self) -> bool:
        """Check if request is allowed

        Returns:
            True if under limit, False otherwise
        """
        with self._lock:
            now = time.time()
            estimated = self._estimate_count(now)

            if estimated < self.limit:
                current_window = self._get_window_id(now)
                self.counters[current_window] += 1

                # Clean old windows
                old_window = current_window - 2
                if old_window in self.counters:
                    del self.counters[old_window]

                return True

            return False

    def get_remaining(self) -> int:
        """Get remaining requests (approximate)"""
        with self._lock:
            now = time.time()
            estimated = self._estimate_count(now)
            return max(0, int(self.limit - estimated))

# Usage
limiter = SlidingWindowCounter(limit=100, window_seconds=60)

if limiter.allow_request():
    response = handle_request()
else:
    return error_response("Rate limited", 429)
```

**Best For**:
- Balance between accuracy and simplicity
- Lower memory than pure sliding window
- Better than fixed window for boundary cases

**Pros**:
- More accurate than fixed window
- Less memory than pure sliding window
- Good balance of trade-offs

**Cons**:
- Approximate (not exact)
- Still allows some boundary bursts (reduced)

---

## Distributed Rate Limiting

### Redis-Based Rate Limiting

**Why Redis?**
- Atomic operations (INCR, ZADD, Lua scripts)
- Fast in-memory storage
- Built-in expiration (TTL)
- Distributed consistency
- High availability (Redis Sentinel, Cluster)

**Connection Patterns**:
```python
import redis
from redis.cluster import RedisCluster
from redis.sentinel import Sentinel

# Single Redis instance
redis_client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,
    socket_timeout=1.0,
    socket_connect_timeout=1.0,
    retry_on_timeout=True
)

# Redis Sentinel (high availability)
sentinel = Sentinel(
    [('sentinel1', 26379), ('sentinel2', 26379)],
    socket_timeout=1.0
)
redis_client = sentinel.master_for('mymaster', socket_timeout=1.0)

# Redis Cluster (sharding)
redis_client = RedisCluster(
    host='redis-cluster',
    port=6379,
    decode_responses=True
)

# Connection pooling
pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    max_connections=50,
    decode_responses=True
)
redis_client = redis.Redis(connection_pool=pool)
```

**Lua Scripts for Atomicity**:
```python
import redis

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Token bucket Lua script
token_bucket_script = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local tokens_needed = tonumber(ARGV[3])
local now = tonumber(ARGV[4])

local state = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(state[1]) or capacity
local last_refill = tonumber(state[2]) or now

-- Refill
local elapsed = now - last_refill
tokens = math.min(capacity, tokens + (elapsed * refill_rate))

-- Consume
if tokens >= tokens_needed then
    tokens = tokens - tokens_needed
    redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
    redis.call('EXPIRE', key, 3600)
    return {1, tokens}
else
    return {0, tokens}
end
"""

# Load script
script_sha = redis_client.script_load(token_bucket_script)

# Execute atomically
def check_rate_limit(user_id: str, capacity: int, refill_rate: float, tokens: int = 1):
    import time
    now = time.time()

    result = redis_client.evalsha(
        script_sha,
        1,  # number of keys
        f"rate_limit:{user_id}",  # KEYS[1]
        capacity,  # ARGV[1]
        refill_rate,  # ARGV[2]
        tokens,  # ARGV[3]
        now  # ARGV[4]
    )

    allowed = bool(result[0])
    remaining = float(result[1])

    return allowed, remaining

# Usage
allowed, remaining = check_rate_limit("user:123", capacity=100, refill_rate=10)
```

**Error Handling**:
```python
import redis
import logging

logger = logging.getLogger(__name__)

def rate_limit_with_fallback(limiter_func, *args, **kwargs):
    """Rate limit with fallback on Redis failure

    Fail-open strategy: Allow requests if Redis unavailable
    """
    try:
        return limiter_func(*args, **kwargs)

    except redis.exceptions.ConnectionError as e:
        logger.error(f"Redis connection failed: {e}")
        return True  # Fail open (allow request)

    except redis.exceptions.TimeoutError as e:
        logger.error(f"Redis timeout: {e}")
        return True  # Fail open

    except redis.exceptions.RedisError as e:
        logger.error(f"Redis error: {e}")
        return True  # Fail open

    except Exception as e:
        logger.exception(f"Unexpected error in rate limiter: {e}")
        return True  # Fail open

# Usage
allowed = rate_limit_with_fallback(limiter.allow_request)
if allowed:
    response = handle_request()
else:
    return error_response("Rate limited", 429)
```

---

### Multi-Tier Rate Limiting

**Hierarchical Limits**:
```python
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class RateLimit:
    """Rate limit configuration"""
    limit: int
    window_seconds: int
    name: str

class MultiTierRateLimiter:
    """Multi-tier rate limiter with multiple time windows"""

    def __init__(self, redis_client: redis.Redis, key: str,
                 limits: List[RateLimit]):
        """
        Args:
            redis_client: Redis connection
            key: Base key for rate limiting
            limits: List of rate limits to enforce
        """
        self.redis = redis_client
        self.key = key
        self.limits = limits
        self.limiters = []

        # Create limiter for each tier
        for limit_config in limits:
            limiter = DistributedFixedWindow(
                redis_client,
                f"{key}:{limit_config.name}",
                limit_config.limit,
                limit_config.window_seconds
            )
            self.limiters.append((limit_config, limiter))

    def allow_request(self) -> tuple[bool, Dict]:
        """Check all rate limit tiers

        Returns:
            (allowed, details)
        """
        details = {}

        for limit_config, limiter in self.limiters:
            allowed, remaining, reset_time = limiter.allow_request()

            details[limit_config.name] = {
                'limit': limit_config.limit,
                'remaining': remaining,
                'reset': reset_time,
                'window': limit_config.window_seconds
            }

            if not allowed:
                # First tier that fails
                return False, details

        return True, details

# Usage
limits = [
    RateLimit(limit=10, window_seconds=1, name='per_second'),
    RateLimit(limit=100, window_seconds=60, name='per_minute'),
    RateLimit(limit=1000, window_seconds=3600, name='per_hour'),
    RateLimit(limit=10000, window_seconds=86400, name='per_day'),
]

limiter = MultiTierRateLimiter(redis_client, "user:123", limits)

allowed, details = limiter.allow_request()
if allowed:
    # Add all tier details to headers
    for tier_name, tier_info in details.items():
        response.headers[f'X-RateLimit-{tier_name}-Limit'] = str(tier_info['limit'])
        response.headers[f'X-RateLimit-{tier_name}-Remaining'] = str(tier_info['remaining'])
else:
    # Find most restrictive tier
    return error_response("Rate limit exceeded", 429, details)
```

---

## HTTP Headers and Standards

### RFC 6585: Additional HTTP Status Codes

**429 Too Many Requests** (RFC 6585, Section 4):
```
HTTP/1.1 429 Too Many Requests
Content-Type: application/json
Retry-After: 60

{
  "error": "rate_limit_exceeded",
  "message": "Too many requests. Please retry after 60 seconds."
}
```

**Specification**:
- Status code `429` indicates client has sent too many requests
- SHOULD include `Retry-After` header indicating when to retry
- MAY include details about rate limiting in response body

**Reference**: [RFC 6585](https://tools.ietf.org/html/rfc6585#section-4)

---

### Rate Limit Headers

**De facto standard headers** (based on GitHub, Twitter, Stripe):
```
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 45
X-RateLimit-Reset: 1234567890
```

**Header Definitions**:
- `X-RateLimit-Limit`: Maximum requests allowed in time window
- `X-RateLimit-Remaining`: Requests remaining in current window
- `X-RateLimit-Reset`: Unix timestamp when limit resets

**Retry-After Header** (RFC 7231):
```
Retry-After: 60          # Seconds to wait
Retry-After: Wed, 21 Oct 2025 07:28:00 GMT  # Absolute time
```

**Implementation**:
```python
from flask import Flask, jsonify, make_response
import time

app = Flask(__name__)

def add_rate_limit_headers(response, limit: int, remaining: int, reset: int):
    """Add standard rate limit headers"""
    response.headers['X-RateLimit-Limit'] = str(limit)
    response.headers['X-RateLimit-Remaining'] = str(remaining)
    response.headers['X-RateLimit-Reset'] = str(reset)
    return response

def rate_limit_exceeded_response(limit: int, reset: int):
    """Create 429 response with proper headers"""
    retry_after = reset - int(time.time())

    response = make_response(jsonify({
        'error': 'rate_limit_exceeded',
        'message': f'Too many requests. Please retry after {retry_after} seconds.',
        'retry_after': retry_after,
        'limit': limit,
        'reset': reset
    }), 429)

    response.headers['Retry-After'] = str(max(0, retry_after))
    response.headers['X-RateLimit-Limit'] = str(limit)
    response.headers['X-RateLimit-Remaining'] = '0'
    response.headers['X-RateLimit-Reset'] = str(reset)

    return response

@app.route('/api/resource')
def get_resource():
    user_id = get_user_id()
    limiter = DistributedFixedWindow(redis_client, user_id, limit=100, window_seconds=60)

    allowed, remaining, reset = limiter.allow_request()

    if not allowed:
        return rate_limit_exceeded_response(limiter.limit, reset)

    # Process request
    data = {'resource': 'data'}
    response = make_response(jsonify(data), 200)
    return add_rate_limit_headers(response, limiter.limit, remaining, reset)
```

**IETF Draft: RateLimit Headers**:

A more recent standard is being developed:
```
RateLimit-Limit: 100
RateLimit-Remaining: 45
RateLimit-Reset: 60
```

**Reference**: [draft-ietf-httpapi-ratelimit-headers](https://datatracker.ietf.org/doc/html/draft-ietf-httpapi-ratelimit-headers)

---

## Storage Backends

### Redis

**Best For**: Distributed systems, high throughput, atomic operations

**Setup**:
```bash
# Install Redis
brew install redis  # macOS
apt-get install redis-server  # Ubuntu

# Start Redis
redis-server

# Connect
redis-cli
```

**Python Client**:
```python
import redis

# Basic connection
client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True
)

# Test connection
client.ping()  # Returns True if connected
```

**Key Patterns**:
```python
# Fixed window
key = f"rate_limit:fixed:{user_id}:{window_id}"
redis.incr(key)
redis.expire(key, window_seconds * 2)

# Sliding window (sorted set)
key = f"rate_limit:sliding:{user_id}"
redis.zadd(key, {request_id: timestamp})
redis.zremrangebyscore(key, 0, window_start)
redis.expire(key, window_seconds)

# Token bucket (hash)
key = f"rate_limit:token:{user_id}"
redis.hmset(key, {'tokens': tokens, 'last_refill': timestamp})
redis.expire(key, 3600)
```

---

### Memcached

**Best For**: Simple counters, low memory overhead

**Setup**:
```bash
# Install
brew install memcached  # macOS
apt-get install memcached  # Ubuntu

# Start
memcached -d -m 64 -p 11211
```

**Python Client**:
```python
import pymemcache.client.base

client = pymemcache.client.base.Client(('localhost', 11211))

def fixed_window_memcached(user_id: str, limit: int, window_seconds: int) -> bool:
    """Fixed window rate limiting with Memcached"""
    now = int(time.time())
    window_id = now // window_seconds
    key = f"rate_limit:{user_id}:{window_id}"

    # Increment counter
    try:
        count = client.incr(key, 1)
    except pymemcache.exceptions.MemcacheError:
        # Key doesn't exist, create it
        client.set(key, 1, expire=window_seconds * 2)
        count = 1

    return count <= limit
```

---

### In-Memory (Local)

**Best For**: Single-instance applications, no external dependencies

**Implementation**:
```python
import time
import threading
from collections import defaultdict
from typing import Dict

class InMemoryRateLimiter:
    """Thread-safe in-memory rate limiter"""

    def __init__(self):
        self.limits: Dict[str, Dict] = defaultdict(dict)
        self._lock = threading.Lock()

    def check_fixed_window(self, key: str, limit: int, window_seconds: int) -> bool:
        """Fixed window rate limiting"""
        now = time.time()
        window_id = int(now // window_seconds)

        with self._lock:
            # Initialize if needed
            if window_id not in self.limits[key]:
                self.limits[key][window_id] = 0

            # Clean old windows
            old_windows = [wid for wid in self.limits[key] if wid < window_id]
            for wid in old_windows:
                del self.limits[key][wid]

            # Check limit
            if self.limits[key][window_id] < limit:
                self.limits[key][window_id] += 1
                return True

            return False

    def check_token_bucket(self, key: str, capacity: int, refill_rate: float) -> bool:
        """Token bucket rate limiting"""
        now = time.time()

        with self._lock:
            # Initialize if needed
            if key not in self.limits:
                self.limits[key] = {
                    'tokens': capacity,
                    'last_refill': now
                }

            # Refi

…(truncated)