# Check Redis memory breakdown
redis-cli INFO memory
 
# Count rate limit keys
redis-cli SCAN 0 MATCH "rl:swl:*" COUNT 10000 | head -20
 
# Check key count with pattern
redis-cli --scan --pattern "rl:swl:*" | wc -l
 
# Check memory usage of one key
redis-cli MEMORY USAGE "rl:swl:user123"
 
# Find largest keys (use carefully in production - SCAN is O(n))
redis-cli --bigkeys
 
# Check TTL of a key (should be 2x window, not -1 which means no TTL)
redis-cli TTL "rl:swl:user123"

# Set maxmemory and eviction policy
redis-cli CONFIG SET maxmemory 8gb
redis-cli CONFIG SET maxmemory-policy allkeys-lru
# LRU eviction: least recently used keys are evicted first
# Rate limit keys for inactive users will be evicted before application data

# Switch from Sliding Window Log to Sliding Window Counter (O(1) memory per user)
# Old: O(limit) = O(500) per user
# New: O(1) - two integers per user
 
# Lua script: sliding window counter (2 integer keys per user, not 500 sorted set entries)
COUNTER_SCRIPT = """
local base = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local win_id = math.floor(now / window)
local curr_key = base .. ':c:' .. win_id
local prev_key = base .. ':c:' .. (win_id - 1)
local elapsed_frac = (now % window) / window
local curr = tonumber(redis.call('GET', curr_key) or '0')
local prev = tonumber(redis.call('GET', prev_key) or '0')
local estimate = prev * (1 - elapsed_frac) + curr
if estimate < limit then
    redis.call('INCR', curr_key)
    redis.call('EXPIRE', curr_key, window * 2)
    return {1, math.floor(estimate) + 1}
end
return {0, math.floor(estimate)}
"""
# Memory: 500,000 users x 2 keys x 30 bytes = 30 MB (vs 50 GB). 1666x reduction.

# Check Redis operations per second
redis-cli INFO stats | grep instantaneous_ops_per_sec
 
# Check Redis CPU
redis-cli INFO cpu
 
# Check slow commands
redis-cli SLOWLOG GET 25
 
# Check command latency
redis-cli --latency-history -i 1
 
# Find most frequent rate limit operations
redis-cli MONITOR | grep "rl:" | head -100
# (Use MONITOR only briefly - it impacts Redis performance)

# Before: 2 Redis round trips per request (INCR + EXPIRE)
count = r.incr(key)
r.expire(key, 60)
 
# After: 1 Redis round trip (pipeline executes both atomically-ish)
pipe = r.pipeline(transaction=False)  # transaction=False = no MULTI/EXEC overhead
pipe.incr(key)
pipe.expire(key, 60)
count, _ = pipe.execute()
# Reduces network round trips by 50%. Significant improvement.

import time
from cachetools import TTLCache
 
# Cache rate limit decisions for 500ms
# For 10,000 RPS and 5,000 unique users, only 2 Redis calls/user/500ms instead of 10
_local_cache = TTLCache(maxsize=50_000, ttl=0.5)
 
def is_allowed_cached(user_id: str) -> bool:
    cached = _local_cache.get(user_id)
    if cached is not None:
        return cached  # Use cached decision without Redis call
 
    # Cache miss: go to Redis
    result = redis_rate_limiter.is_allowed(user_id)
    _local_cache[user_id] = result["allowed"]
    return result["allowed"]
 
# Trade-off: User who hits limit might get 500ms more requests (cached "allowed")
# Acceptable for general APIs. Not acceptable for payment APIs.

import asyncio
from concurrent.futures import ThreadPoolExecutor
 
# For endpoints where slight over-limit is acceptable:
# Check rate limit asynchronously, don't block the request
async def rate_limit_async(user_id: str, handler):
    # Start handler immediately (no blocking)
    handler_task = asyncio.create_task(handler())
    # Check rate limit concurrently
    rl_task = asyncio.create_task(redis_check_async(user_id))
 
    result = await handler_task
    rl_result = await rl_task
 
    if not rl_result["allowed"]:
        # Request already processed, but log the violation
        # Could add to a "debt" bucket to reduce future allowance
        metrics.increment("rate_limit.post_hoc_denial")
 
    return result

class RateLimiterWithIncidentAwareness:
    def is_allowed(self, identifier: str) -> dict:
        result = self.redis_limiter.is_allowed(identifier)
        result["data_source"] = "primary"
 
        # If we know we are in a post-failover state, note it
        if self._is_post_failover_window():
            result["note"] = "post_failover_approximate"
            metrics.increment("rate_limit.post_failover_check")
 
        return result
 
    def _is_post_failover_window(self) -> bool:
        # Sentinel notifies app of failover via pub/sub
        # App sets a flag for 60 seconds after failover event
        return self._failover_detected_at is not None and \
               time.time() - self._failover_detected_at < 60

# Find top ASNs by request count in nginx logs
awk '{print $1}' access.log | sort | uniq -c | sort -rn | head -20
# If top IPs are from same /24 block, it's likely a rotating attack
 
# Check with whois
whois 203.0.113.0 | grep -E "netname|country|org"
 
# Count unique IPs per 5-minute window
awk '{print $1, $4}' access.log | awk '{print substr($2, 2, 16), $1}' | sort | uniq | wc -l

class BehavioralRateLimiter:
    """
    Detects IP rotation attacks by tracking behavioral signals across requests.
    """
 
    def get_request_fingerprint(self, request) -> str:
        """
        Combine signals that are hard to rotate:
        - TLS fingerprint (JA3 hash): hard to change per request
        - HTTP/2 settings fingerprint: browser-specific, hard to fake
        - Accept-Language, Accept-Encoding: usually consistent per client
        - User-Agent: easy to fake but often consistent for bot libraries
        """
        import hashlib
        ja3 = request.headers.get("X-JA3-Fingerprint", "unknown")
        ua = request.headers.get("User-Agent", "")[:50]
        accept = request.headers.get("Accept", "")[:20]
        lang = request.headers.get("Accept-Language", "")[:10]
 
        fingerprint_raw = f"{ja3}:{ua}:{accept}:{lang}"
        return hashlib.sha256(fingerprint_raw.encode()).hexdigest()[:16]
 
    def is_allowed_with_fingerprint(self, request) -> bool:
        ip = get_real_ip(request)
        fingerprint = self.get_request_fingerprint(request)
 
        # Rate limit by fingerprint regardless of IP
        fp_key = f"rl:fp:{fingerprint}"
        fp_count = redis.incr(fp_key)
        redis.expire(fp_key, 60)
        if fp_count > 100:
            return False  # Same browser/tool, rotating IPs -> blocked
 
        # Also check per-IP
        ip_key = f"rl:ip:{ip}"
        ip_count = redis.incr(ip_key)
        redis.expire(ip_key, 60)
        if ip_count > 20:
            return False
 
        return True

import ipaddress
 
def get_subnet_key(ip: str, prefix_len_v4: int = 24, prefix_len_v6: int = 48) -> str:
    ip_obj = ipaddress.ip_address(ip)
    if ip_obj.version == 4:
        network = ipaddress.ip_network(f"{ip}/{prefix_len_v4}", strict=False)
    else:
        network = ipaddress.ip_network(f"{ip}/{prefix_len_v6}", strict=False)
    return f"rl:subnet:{network.network_address}"
 
# Rate limit by /24 subnet (allows 256 IPv4 addresses to share a limit)
# Even if attacker rotates within /24, they share the counter
subnet_key = get_subnet_key(client_ip)
subnet_count = redis.incr(subnet_key)
redis.expire(subnet_key, 60)

def get_rate_limit_identifier(request) -> tuple[str, str]:
    """
    Returns (identifier, identifier_type).
    Prefer specific identifiers over generic ones.
    """
    # 1. Best: Authenticated user ID (from JWT or session)
    user_id = extract_user_id_from_auth(request)
    if user_id:
        return f"user:{user_id}", "user"
 
    # 2. Good: API key
    api_key = request.headers.get("X-API-Key")
    if api_key:
        return f"apikey:{hash_key(api_key)}", "apikey"
 
    # 3. Fallback: IP (but with higher limit to account for NAT)
    ip = get_real_ip(request)
    return f"ip:{ip}", "ip"
 
IP_BASED_LIMIT = 500     # Higher limit for IP (accounts for NAT sharing)
USER_BASED_LIMIT = 100   # Tighter limit for identified users
 
def get_limit_for_identifier(identifier_type: str) -> int:
    return IP_BASED_LIMIT if identifier_type == "ip" else USER_BASED_LIMIT

# Known CGN address ranges (RFC 6598: 100.64.0.0/10)
CGN_RANGES = ["100.64.0.0/10"]
MOBILE_CARRIER_RANGES = [
    # Load from a database of known shared NAT ranges
    # These change frequently - use a service like MaxMind or IP2Location
]
 
def is_cgn_address(ip: str) -> bool:
    ip_obj = ipaddress.ip_address(ip)
    return any(
        ip_obj in ipaddress.ip_network(r)
        for r in CGN_RANGES + MOBILE_CARRIER_RANGES
    )
 
def get_ip_limit(ip: str) -> int:
    if is_cgn_address(ip):
        return 5_000  # Shared NAT: 50x higher limit
    return 100        # Regular IP

-- BROKEN: Uses client-supplied time (from a potentially drifted server)
local now = tonumber(ARGV[1])  -- application server time (may be drifted)
local window_id = math.floor(now / 60)
 
-- CORRECT: Use Redis server time (single source of truth)
local time_result = redis.call('TIME')
local now = tonumber(time_result[1]) + tonumber(time_result[2]) / 1e6
local window_id = math.floor(now / 60)
-- All rate limit checks on this Redis instance use identical time
-- Clock skew between application servers no longer matters

class RecoveryAwareRateLimiter:
    """
    During recovery from an outage, temporarily reduce effective limits
    to give the system time to warm up.
    """
 
    def __init__(self, normal_limiter, recovery_duration: int = 300):
        self.normal_limiter = normal_limiter
        self.recovery_duration = recovery_duration  # 5-minute ramp
        self._recovery_start = None
 
    def signal_recovery_start(self):
        self._recovery_start = time.time()
 
    def get_recovery_multiplier(self) -> float:
        if self._recovery_start is None:
            return 1.0
 
        elapsed = time.time() - self._recovery_start
        if elapsed >= self.recovery_duration:
            self._recovery_start = None
            return 1.0
 
        # Linear ramp: 10% at recovery start, 100% after 5 minutes
        progress = elapsed / self.recovery_duration
        return 0.1 + (0.9 * progress)
 
    def is_allowed(self, identifier: str, base_limit: int) -> bool:
        multiplier = self.get_recovery_multiplier()
        effective_limit = max(1, int(base_limit * multiplier))
        return self.normal_limiter.is_allowed(identifier, limit=effective_limit)

def maintenance_response(request) -> Response:
    # Give clients DIFFERENT retry times (jittered) so they don't all come back at once
    user_id = extract_user_id(request) or request.remote_addr
    # Deterministic jitter: same user always gets same offset (consistent experience)
    user_hash = int(hashlib.sha256(user_id.encode()).hexdigest(), 16)
    base_retry = 1800  # 30 minutes until maintenance ends
    jitter = user_hash % 600  # Up to 10 minutes of jitter
    retry_after = base_retry + jitter
 
    return Response(
        status=503,
        headers={
            "Retry-After": str(retry_after),
            "X-Maintenance-End": "2025-01-01T02:30:00Z"
        }
    )

# During the 48 hours before a campaign launch, run in shadow mode:
class ShadowRateLimiter:
    def __init__(self, limiter):
        self.limiter = limiter
        self.shadow_log = []
 
    def is_allowed_shadow(self, identifier: str) -> bool:
        result = self.limiter.is_allowed(identifier)
        if not result["allowed"]:
            # Would have been denied - log it, but allow anyway
            self.shadow_log.append({
                "time": time.time(),
                "identifier": identifier,
                "count": result["count"],
                "limit": result["limit"]
            })
            return True  # Allow in shadow mode
        return True

def handle_request_gracefully(request) -> Response:
    identifier, id_type = get_rate_limit_identifier(request)
    result = rate_limiter.check(identifier)
 
    if not result["allowed"] and id_type == "ip":
        # Under heavy load: serve cached/degraded response instead of 429
        cached_response = cache.get_stale(request.path)
        if cached_response:
            return Response(
                body=cached_response,
                status=200,
                headers={
                    "X-Cache": "STALE",
                    "X-RateLimit-Warning": "Near limit, serving cached response"
                }
            )
 
    return normal_handler(request)

# WRONG: In-memory state resets on deployment
class InMemoryRateLimiter:
    def __init__(self):
        self._counters = {}  # LOST on every deployment
 
# CORRECT: All state in Redis (survives any number of deployments)
class RedisRateLimiter:
    def __init__(self, redis_client):
        self.r = redis_client
    # State lives in Redis, not in the application process.
    # Blue-green, rolling, canary deployments all work transparently.
    # Even if ALL instances are restarted simultaneously, state survives.

# Dual-write during migration: write to both, read from Redis
class MigratingRateLimiter:
    def __init__(self, local_limiter, redis_limiter):
        self.local = local_limiter
        self.redis = redis_limiter
 
    def is_allowed(self, identifier: str) -> bool:
        local_result = self.local.is_allowed(identifier)
        redis_result = self.redis.is_allowed(identifier)
        # During migration: use Redis result (more accurate)
        # Log when they disagree
        if local_result != redis_result["allowed"]:
            logger.info("migration_disagreement", identifier=identifier,
                       local=local_result, redis=redis_result["allowed"])
        return redis_result["allowed"]

import redis
import json
from typing import Optional
 
TIER_CACHE_TTL = 300  # Cache tier for 5 minutes
 
def get_user_tier_cached(user_id: str, r: redis.Redis, db) -> str:
    cache_key = f"tier:{user_id}"
 
    # Check Redis cache first (microseconds, not milliseconds)
    cached = r.get(cache_key)
    if cached:
        return cached
 
    # Cache miss: hit DB (happens at most once per 5 minutes per user)
    tier = db.query("SELECT tier FROM users WHERE id = %s", user_id)
    r.setex(cache_key, TIER_CACHE_TTL, tier)
    return tier
 
# At 5,000 RPS with 100,000 unique users and 5-minute TTL:
# Cache hit rate: ~99.9% (5000 req/s x 300s = 1,500,000 requests per TTL period)
# (100,000 users / 1,500,000 requests = 0.07% cache miss rate)
# DB queries for tier lookup: ~3.5/second (not 5,000/second)
# DB connection usage for rate limiting: ~0 (not 30+)

def generate_token(user_id: str) -> str:
    tier = db.get_user_tier(user_id)
    token = create_jwt({"user_id": user_id, "tier": tier})
 
    # Cache tier at login time - invalidate on plan change
    r.setex(f"tier:{user_id}", 3600, tier)
    return token
 
# Now tier is always in cache when user makes API calls (they just logged in)
# DB is only hit when token is generated, not on every API call

@Service
public class EnrichmentClient {
 
    // Outbound rate limiter: limit calls TO Service B
    private final Bucket outboundBucket = Bucket4j.builder()
        .addLimit(Bandwidth.classic(500, Refill.greedy(500, Duration.ofSeconds(1))))
        .build();
 
    public List<EnrichmentData> enrich(List<String> ids) {
        // Batch: instead of N calls, send 1 call with N IDs
        // (Requires Service B to have a bulk endpoint)
        return batchFetch(ids);
    }
 
    public EnrichmentData enrichOne(String id) {
        // Rate-limited outbound call
        if (!outboundBucket.tryConsume(1)) {
            // Outbound limit hit: return cached/default data instead of erroring
            return getDefaultEnrichment(id);
        }
        return serviceB.get(id);
    }
}

# Service B publishes its rate limit contract
# service-b-contract.yaml
rate_limit:
  max_rps: 500
  max_concurrent_requests: 50
  burst_size: 100
 
# Service A MUST configure its outbound limiter based on Service B's contract
# Contract testing: if Service B changes its limit, Service A's tests catch it

class SegregatedQuotaManager:
 
    QUOTA_SEGMENTS = {
        "human_interactive": {"daily": 100_000, "burn_rate_alert": 5_000},   # per hour
        "batch_jobs":        {"daily": 500_000, "burn_rate_alert": 50_000},   # per hour
        "sync_service":      {"daily": 200_000, "burn_rate_alert": 10_000},   # per hour
    }
 
    def check_quota(self, segment: str) -> bool:
        daily_key = f"quota:{segment}:{self._today()}"
        count = self.r.incr(daily_key)
        if count == 1:
            self.r.expire(daily_key, 86400 * 2)
 
        limit = self.QUOTA_SEGMENTS[segment]["daily"]
        if count > limit:
            self.alert(f"Quota exhausted for {segment}")
            return False
 
        # Burn rate alert
        hourly_key = f"quota_hr:{segment}:{self._current_hour()}"
        hourly = int(self.r.get(hourly_key) or 0)
        alert_threshold = self.QUOTA_SEGMENTS[segment]["burn_rate_alert"]
        if hourly > alert_threshold:
            self.alert(f"High burn rate for {segment}: {hourly}/hour (alert at {alert_threshold})")
 
        return True

class JobRateLimiter:
    """
    Detects runaway jobs by tracking per-job-instance request rate.
    """
    def __init__(self, r, max_rpm_per_job: int = 60):
        self.r = r
        self.max_rpm = max_rpm_per_job
 
    def check_job(self, job_id: str) -> bool:
        key = f"job_rl:{job_id}:{int(time.time())//60}"
        count = self.r.incr(key)
        self.r.expire(key, 120)
 
        if count > self.max_rpm:
            # This job instance is making >60 requests/minute - likely a bug
            self.r.set(f"job_killed:{job_id}", 1, ex=3600)
            alert_ops(f"Job {job_id} killed: {count} RPM (max {self.max_rpm})")
            return False
 
        # Check if this job has been killed
        if self.r.exists(f"job_killed:{job_id}"):
            return False
 
        return True

class TenantQuotaManager:
    """
    Each tenant has completely isolated quotas.
    One tenant cannot consume another tenant's resources.
    """
 
    def __init__(self, r, plan_quotas: dict):
        self.r = r
        self.plan_quotas = plan_quotas
 
    def check_tenant_quota(self, tenant_id: str, resource: str) -> dict:
        plan = self.get_tenant_plan(tenant_id)
        quota = self.plan_quotas[plan][resource]
        today = time.strftime("%Y-%m-%d")
        key = f"quota:tenant:{tenant_id}:{resource}:{today}"
 
        count = self.r.incr(key)
        if count == 1:
            self.r.expire(key, 86400 * 2)
 
        return {
            "allowed": count <= quota,
            "used": count,
            "limit": quota,
            "remaining": max(0, quota - count),
            "tenant": tenant_id  # Each tenant has their OWN counter
        }
 
    def get_tenant_quota_usage(self, tenant_id: str) -> dict:
        today = time.strftime("%Y-%m-%d")
        pattern = f"quota:tenant:{tenant_id}:*:{today}"
        usage = {}
        for key in self.r.scan_iter(pattern):
            resource = key.split(":")[3]
            usage[resource] = int(self.r.get(key) or 0)
        return usage

# Redis slow log for Lua
redis-cli CONFIG SET slowlog-log-slower-than 500  # 0.5ms threshold
redis-cli SLOWLOG GET 25
 
# Benchmark your Lua script directly
redis-cli --latency-history -i 1
redis-cli DEBUG SLEEP 0  # Measures baseline latency
 
# Benchmark specific Lua script
redis-cli EVAL "return redis.call('SET', KEYS[1], ARGV[1])" 1 testkey testval

-- BEFORE: Complex GCRA with floating point math (slower)
local tat = tonumber(redis.call('GET', KEYS[1]) or '0')
local now = tonumber(ARGV[1])
local emission_interval = tonumber(ARGV[2])
local burst_offset = tonumber(ARGV[3])
local new_tat = math.max(now, tat) + emission_interval
local allowed_at = new_tat - burst_offset
if allowed_at <= now then
    redis.call('SETEX', KEYS[1], tonumber(ARGV[4]), tostring(new_tat))
    return {1, new_tat - now}
end
return {0, allowed_at - now}
-- Execution time: ~150 microseconds
 
-- AFTER: Simple sliding window counter (faster)
local k1 = KEYS[1] .. ':' .. math.floor(tonumber(ARGV[3]) / tonumber(ARGV[2]))
local k2 = KEYS[1] .. ':' .. (math.floor(tonumber(ARGV[3]) / tonumber(ARGV[2])) - 1)
local c1 = tonumber(redis.call('GET', k1) or '0')
local c2 = tonumber(redis.call('GET', k2) or '0')
local w = (tonumber(ARGV[3]) % tonumber(ARGV[2])) / tonumber(ARGV[2])
if c1 + c2 * (1 - w) < tonumber(ARGV[1]) then
    redis.call('INCR', k1)
    redis.call('EXPIRE', k1, tonumber(ARGV[2]) * 2)
    return 1
end
return 0
-- Execution time: ~50 microseconds. 3x faster.

if response.status_code == 429:
    time.sleep(1)  # Default: retry after 1 second
    return self.retry(request)

def handle_rate_limited_response(identifier: str, reset_at: int) -> Response:
    now = int(time.time())
    retry_after = max(1, reset_at - now)
 
    # Optional: Add jitter to stagger retries
    retry_after_jittered = retry_after + random.randint(0, 10)
 
    return Response(
        status=429,
        headers={
            "Retry-After": str(retry_after_jittered),
            "X-RateLimit-Reset": str(reset_at),
            "X-RateLimit-Remaining": "0"
        },
        body={
            "error": "rate_limit_exceeded",
            "retry_after": retry_after_jittered
        }
    )

def call_with_retry(func, max_retries=5):
    for attempt in range(max_retries):
        response = func()
        if response.status_code == 429:
            # Use server's Retry-After, with fallback to exponential backoff
            retry_after = response.headers.get("Retry-After")
            if retry_after:
                wait = float(retry_after)
            else:
                wait = min(300, (2 ** attempt) + random.uniform(0, 5))  # jitter
            if attempt < max_retries - 1:
                time.sleep(wait)
        else:
            return response
    raise MaxRetriesExceeded()

import hashlib
import random
 
NUM_VIRTUAL_SHARDS = 10
 
def get_sharded_key(base_key: str, shards: int = NUM_VIRTUAL_SHARDS) -> str:
    """
    For hot keys: distribute across N virtual shards to spread load.
    The actual count requires summing all shards (approximate).
    """
    shard_id = random.randint(0, shards - 1)
    return f"{base_key}:shard:{shard_id}"
 
def get_total_count_sharded(base_key: str, r, shards: int = NUM_VIRTUAL_SHARDS) -> int:
    """Get approximate total count across all shards."""
    pipe = r.pipeline()
    for i in range(shards):
        pipe.get(f"{base_key}:shard:{i}")
    results = pipe.execute()
    return sum(int(v or 0) for v in results)
 
# For the hot creator: 10 Redis keys instead of 1
# Spread across 10 shards -> 10x load distribution
# Trade-off: reading total count requires 10 GET operations (batched in pipeline)

def get_limit_for_hot_user(user_id: str) -> int:
    # Detect hot users: if their per-minute count exceeds normal by 100x
    # Automatically route them to a higher limit (so their requests don't
    # clog the rate limit check for other users)
    if is_hot_user(user_id):
        return get_enterprise_limit(user_id)  # higher limit, separate key space
    return get_standard_limit(user_id)

RESOURCE_LIMITS = {
    "/api/users":          100,   # All methods share this limit
    "/api/products":       500,
    "/api/search":         30,
    "/api/auth/login":     5,
}
 
def get_resource_key(request) -> str:
    # Normalize: strip method, normalize path params
    path = normalize_path(request.path)
    # All methods to the same resource share one counter
    return f"rl:resource:{user_id}:{path}"
 
# GET /api/users and POST /api/users both decrement the same "users" counter
# Method-specific limits still possible via a separate higher-cost counter:
WRITE_ADDITIONAL_COST = {
    "POST":   2,   # Writes cost 2x reads
    "PUT":    2,
    "DELETE": 3,
    "PATCH":  1,
    "GET":    1,
    "HEAD":   0,   # No cost
}

class FeatureFlagRateLimitAnalyzer:
    """
    Before enabling a feature flag, estimate its rate limit impact.
    """
    def estimate_impact(
        self,
        feature_name: str,
        api_calls_per_user_per_hour: int,
        rollout_percentage: float,
        total_active_users: int
    ) -> dict:
        affected_users = int(total_active_users * rollout_percentage)
        additional_rps = (affected_users * api_calls_per_user_per_hour) / 3600
        user_usage_pct = api_calls_per_user_per_hour / USER_HOURLY_LIMIT * 100
 
        return {
            "feature": feature_name,
            "affected_users": affected_users,
            "additional_rps": additional_rps,
            "user_rate_limit_usage_pct": user_usage_pct,
            "will_hit_user_limit": user_usage_pct > 80,
            "recommendation": (
                "Increase user limit before rollout"
                if user_usage_pct > 80 else "Safe to roll out"
            )
        }
 
# Before rolling out "smart refresh":
analysis = analyzer.estimate_impact(
    feature_name="smart_refresh",
    api_calls_per_user_per_hour=600,   # 10x polling, 60 polls/hour * 10 = 600
    rollout_percentage=0.01,
    total_active_users=500_000
)
# Would have shown: user_rate_limit_usage_pct = 120% -> will_hit_user_limit = True
# Team would have adjusted limits before rollout

class WebSocketConnectionManager:
    """
    Track and enforce concurrent WebSocket limits with TTL-based cleanup.
    """
 
    MAX_CONNECTIONS_PER_USER = 5
    CONNECTION_TTL = 300  # 5-minute heartbeat window
 
    def add_connection(self, user_id: str, connection_id: str) -> bool:
        key = f"ws:conns:{user_id}"
        conn_ttl_key = f"ws:conn:{connection_id}:ttl"
 
        # Cleanup stale connections first
        self._cleanup_stale(user_id)
 
        # Check current count
        count = int(self.r.scard(key) or 0)
        if count >= self.MAX_CONNECTIONS_PER_USER:
            return False
 
        # Add connection
        self.r.sadd(key, connection_id)
        self.r.expire(key, self.CONNECTION_TTL * 2)
        self.r.setex(conn_ttl_key, self.CONNECTION_TTL, "alive")
        return True
 
    def heartbeat(self, connection_id: str) -> None:
        """Client sends heartbeat every 60 seconds to keep connection alive."""
        conn_ttl_key = f"ws:conn:{connection_id}:ttl"
        self.r.expire(conn_ttl_key, self.CONNECTION_TTL)
 
    def remove_connection(self, user_id: str, connection_id: str) -> None:
        key = f"ws:conns:{user_id}"
        self.r.srem(key, connection_id)
        self.r.delete(f"ws:conn:{connection_id}:ttl")
 
    def _cleanup_stale(self, user_id: str) -> None:
        """Remove connections whose heartbeat TTL has expired."""
        key = f"ws:conns:{user_id}"
        all_connections = self.r.smembers(key)
        for conn_id in all_connections:
            ttl_key = f"ws:conn:{conn_id}:ttl"
            if not self.r.exists(ttl_key):
                # TTL expired = heartbeat not received = stale connection
                self.r.srem(key, conn_id)

class IdempotencyAwareRateLimiter:
    """
    Uses idempotency keys to ensure retries of the same request
    do not consume additional rate limit tokens.
    """
 
    IDEMPOTENCY_TTL = 300  # 5 minutes
 
    def check_and_consume(
        self,
        user_id: str,
        idempotency_key: str = None,
        cost: int = 1
    ) -> dict:
        if idempotency_key:
            # Check if we've already processed this idempotency key
            idem_redis_key = f"idem:{user_id}:{idempotency_key}"
            existing = self.r.get(idem_redis_key)
 
            if existing:
                # Already counted - return same result, no new charge
                result = json.loads(existing)
                result["idempotent_replay"] = True
                return result
 
        # New request: check rate limit
        result = self.rate_limiter.check(user_id, cost=cost)
 
        if idempotency_key and result["allowed"]:
            # Store result for idempotency replay
            self.r.setex(
                f"idem:{user_id}:{idempotency_key}",
                self.IDEMPOTENCY_TTL,
                json.dumps(result)
            )
 
        return result

import uuid
 
class ResilientAPIClient:
    def make_request(self, method: str, path: str, body: dict = None) -> dict:
        # Generate idempotency key ONCE for this logical request
        idempotency_key = str(uuid.uuid4())
 
        for attempt in range(5):
            response = requests.request(
                method, f"{self.base_url}{path}",
                json=body,
                headers={
                    "X-Idempotency-Key": idempotency_key,  # Same key on every retry
                    "Authorization": f"Bearer {self.token}"
                }
            )
            if response.status_code != 429:
                return response.json()
 
            retry_after = float(response.headers.get("Retry-After", 2 ** attempt))
            time.sleep(retry_after + random.uniform(0, 1))
 
        raise MaxRetriesError()