# Good: Namespaced, scannable, evictable
key = f"rl:{algorithm}:{entity_type}:{identifier}:{window}"
 
# Examples:
"rl:fw:user:user_123:1735689600"      # Fixed window, user, window ID
"rl:sw:apikey:abc123"                  # Sliding window, API key
"rl:tb:ip:203.0.113.42"               # Token bucket, IP address
"rl:fw:global:system:1735689600"       # Global limit
 
# Key length matters: binary_remote_addr vs remote_addr in Nginx
# IPv4: 4 bytes vs up to 15 characters
# Use hashing for long identifiers
import hashlib
def make_redis_key(prefix: str, identifier: str) -> str:
    if len(identifier) > 64:
        identifier = hashlib.sha256(identifier.encode()).hexdigest()[:16]
    return f"{prefix}:{identifier}"

@Service
public class LocalRateLimitService {
 
    // ConcurrentHashMap for thread safety
    private final ConcurrentHashMap<String, Bucket> userBuckets = new ConcurrentHashMap<>();
    // Scheduled cleanup to prevent memory leaks
    @Scheduled(fixedDelay = 300_000) // every 5 minutes
    public void cleanupStaleBuckets() {
        // Remove buckets that haven't been used recently
        // In production, use Caffeine cache with expiry instead
        userBuckets.entrySet().removeIf(entry ->
            entry.getValue().getAvailableTokens() == entry.getValue().asVerbose()
                .getConfiguration().getBandwidths()[0].getCapacity()
        );
    }
 
    public boolean isAllowed(String userId) {
        Bucket bucket = userBuckets.computeIfAbsent(userId, key ->
            Bucket4j.builder()
                .addLimit(Bandwidth.classic(100, Refill.greedy(100, Duration.ofMinutes(1))))
                .build()
        );
        return bucket.tryConsume(1);
    }
}

from cachetools import TTLCache
import threading
 
class LocalRateLimiter:
    """
    Local rate limiter with automatic cleanup via TTL cache.
    Each user's counter automatically expires after the window.
    """
 
    def __init__(self, limit: int, window_seconds: int):
        self.limit = limit
        self.window_seconds = window_seconds
        # TTLCache automatically evicts entries after maxsize or ttl
        self.cache = TTLCache(maxsize=100_000, ttl=window_seconds)
        self.lock = threading.Lock()
 
    def is_allowed(self, identifier: str) -> bool:
        with self.lock:
            count = self.cache.get(identifier, 0)
            if count >= self.limit:
                return False
            self.cache[identifier] = count + 1
            return True

# INCORRECT: Non-atomic check-then-increment
count = redis.get(key)
if int(count or 0) < limit:
    redis.incr(key)   # Another thread may increment between these two lines
    return True
return False
 
# BETTER: Atomic increment, then check
count = redis.incr(key)   # Atomic! No race condition in the increment itself
if count == 1:
    redis.expire(key, window_seconds)
return count <= limit
# Still a minor issue: multiple "count == 1" checks can race on expire
# But the count check is now safe

count = redis.incr(key)      # Thread A: gets 1
if count == 1:                # Thread A: true
    redis.expire(key, 60)     # Thread A: sets TTL
                              # CRASH HERE - process dies
                              # Thread A never sets TTL
                              # Thread B: gets 2, count != 1, doesn't set TTL
                              # Key lives FOREVER
 
# Fix: Always set expiry, not just on first request
count = redis.incr(key)
redis.expire(key, 60)  # Set on every request (slightly wasteful but safe)
 
# Better fix: Use SET with NX and EX
# Or use a Lua script

-- atomic_rate_limit.lua
-- Implements sliding window counter with atomic check-and-increment
--
-- KEYS[1]: base key for this rate limit
-- ARGV[1]: limit (max requests per window)
-- ARGV[2]: window_size (in seconds)
-- ARGV[3]: current_time (Unix timestamp, integer seconds)
-- Returns: {allowed (0/1), current_count, remaining, reset_at}
 
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
 
-- Determine window boundaries
local current_window_id = math.floor(now / window)
local prev_window_id = current_window_id - 1
 
local curr_key = key .. ':' .. current_window_id
local prev_key = key .. ':' .. prev_window_id
 
-- Get counts for current and previous windows
local curr_count = tonumber(redis.call('GET', curr_key) or '0')
local prev_count = tonumber(redis.call('GET', prev_key) or '0')
 
-- How far are we into the current window? (0.0 to 1.0)
local window_elapsed = (now % window) / window
 
-- Estimate using sliding window: prev * (1 - elapsed) + curr
local estimated = prev_count * (1 - window_elapsed) + curr_count
 
if estimated < limit then
    -- Allow: increment current window
    local new_count = redis.call('INCR', curr_key)
    if new_count == 1 then
        redis.call('EXPIRE', curr_key, window * 2)
    end
    local reset_at = (current_window_id + 1) * window
    local remaining = limit - math.floor(estimated) - 1
    return {1, math.floor(estimated) + 1, math.max(0, remaining), reset_at}
else
    -- Deny
    local reset_at = (current_window_id + 1) * window
    return {0, math.floor(estimated), 0, reset_at}
end

import redis
import time
 
 
class AtomicSlidingWindowLimiter:
 
    SCRIPT = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local current_window_id = math.floor(now / window)
local prev_window_id = current_window_id - 1
local curr_key = key .. ':' .. current_window_id
local prev_key = key .. ':' .. prev_window_id
local curr_count = tonumber(redis.call('GET', curr_key) or '0')
local prev_count = tonumber(redis.call('GET', prev_key) or '0')
local window_elapsed = (now % window) / window
local estimated = prev_count * (1 - window_elapsed) + curr_count
if estimated < limit then
    local new_count = redis.call('INCR', curr_key)
    if new_count == 1 then redis.call('EXPIRE', curr_key, window * 2) end
    local reset_at = (current_window_id + 1) * window
    return {1, math.floor(estimated) + 1, math.max(0, limit - math.floor(estimated) - 1), reset_at}
else
    return {0, math.floor(estimated), 0, (current_window_id + 1) * window}
end
"""
 
    def __init__(self, r: redis.Redis, limit: int, window_seconds: int):
        self.r = r
        self.limit = limit
        self.window_seconds = window_seconds
        self._script = r.register_script(self.SCRIPT)
 
    def is_allowed(self, identifier: str) -> dict:
        now = int(time.time())
        key = f"rl:sw:{identifier}"
        result = self._script(keys=[key], args=[self.limit, self.window_seconds, now])
 
        return {
            "allowed": bool(int(result[0])),
            "current_count": int(result[1]),
            "remaining": int(result[2]),
            "reset_at": int(result[3]),
            "limit": self.limit
        }

# Redis MULTI/EXEC: optimistic locking with WATCH
def is_allowed_with_transaction(r, key, limit):
    with r.pipeline() as pipe:
        while True:
            try:
                # Watch the key for changes
                pipe.watch(key)
                count = int(pipe.get(key) or 0)
 
                if count >= limit:
                    pipe.reset()
                    return False
 
                # Start transaction
                pipe.multi()
                pipe.incr(key)
                pipe.expire(key, 60)
                pipe.execute()  # Fails if key changed since WATCH
                return True
 
            except redis.WatchError:
                # Another client modified the key - retry
                continue  # This loop is unbounded - dangerous under high contention!

-- This will fail in Redis Cluster if key1 and key2 are on different nodes
local curr = redis.call('GET', KEYS[1])  -- node A
local prev = redis.call('GET', KEYS[2])  -- node B (CROSS-SLOT ERROR!)

# Without hash tags: keys on different slots (BROKEN in cluster)
curr_key = f"rl:sw:{user_id}:curr"   # slot = CRC16("rl:sw:user123:curr") % 16384
prev_key = f"rl:sw:{user_id}:prev"   # slot = CRC16("rl:sw:user123:prev") % 16384
 
# With hash tags: same user, same slot (CORRECT)
curr_key = f"rl:sw:{{{user_id}}}:curr"   # slot = CRC16("user123") % 16384
prev_key = f"rl:sw:{{{user_id}}}:prev"   # slot = CRC16("user123") % 16384
# Both keys go to the same slot because {user_id} determines the hash

SCRIPT = """
local user_id = ARGV[4]
local curr_key = 'rl:sw:{' .. user_id .. '}:curr'
local prev_key  = 'rl:sw:{' .. user_id .. '}:prev'
-- Now both keys are guaranteed to be on the same slot
local curr = tonumber(redis.call('GET', curr_key) or '0')
local prev = tonumber(redis.call('GET', prev_key) or '0')
...
"""

# Nginx sticky session based on IP
upstream backend {
    ip_hash;
    server server1:8080;
    server server2:8080;
    server server3:8080;
}
 
# AWS ELB: enable "Stickiness" in target group settings
# Stickiness type: Application-based / LB-based
# Duration: 1 day

import time
import redis
import threading
from dataclasses import dataclass, field
 
 
@dataclass
class LocalCounter:
    count: int = 0
    last_sync: float = field(default_factory=time.time)
 
 
class HybridRateLimiter:
    """
    Hybrid rate limiter: local approximation + global Redis enforcement.
 
    Strategy:
    1. Each instance maintains a local counter
    2. Requests that pass the local check are then validated globally in Redis
    3. Local sync happens every `sync_interval` seconds
 
    This reduces Redis calls by (sync_interval * local_rps) factor.
    """
 
    def __init__(
        self,
        r: redis.Redis,
        limit: int,
        window_seconds: int,
        # Each server reserves a "chunk" of the global limit
        # For 3 servers with limit=100: reserve_percent=0.4 -> each reserves 40
        reserve_percent: float = 0.4
    ):
        self.r = r
        self.limit = limit
        self.window_seconds = window_seconds
        self.local_limit = int(limit * reserve_percent)
        self.locals: dict[str, LocalCounter] = {}
        self.lock = threading.Lock()
        self._script = r.register_script("""
local key = KEYS[1]
local add = tonumber(ARGV[1])
local limit = tonumber(ARGV[2])
local window = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
local window_id = math.floor(now / window)
local full_key = key .. ':' .. window_id
local count = redis.call('INCRBY', full_key, add)
if count == add then
    redis.call('EXPIRE', full_key, window * 2)
end
if count <= limit then
    return {1, count}
else
    -- Refund: we over-counted, subtract back
    redis.call('DECRBY', full_key, add)
    return {0, count - add}
end
""")
 
    def is_allowed(self, identifier: str, cost: int = 1) -> bool:
        with self.lock:
            if identifier not in self.locals:
                self.locals[identifier] = LocalCounter()
            local = self.locals[identifier]
 
            # Phase 1: Fast local check
            if local.count + cost > self.local_limit:
                # Local quota exhausted, go to Redis
                return self._global_check(identifier, cost)
 
            # Phase 2: Increment local counter
            local.count += cost
 
            # Phase 3: Periodically sync local counts to Redis
            now = time.time()
            if now - local.last_sync >= 1.0:  # sync every second
                self._sync_to_redis(identifier, local)
 
            return True
 
    def _global_check(self, identifier: str, cost: int) -> bool:
        now = int(time.time())
        key = f"rl:hybrid:{identifier}"
        result = self._script(keys=[key], args=[cost, self.limit, self.window_seconds, now])
        return bool(int(result[0]))
 
    def _sync_to_redis(self, identifier: str, local: LocalCounter) -> None:
        """Flush accumulated local counts to Redis."""
        if local.count == 0:
            return
        to_sync = local.count
        local.count = 0
        local.last_sync = time.time()
 
        now = int(time.time())
        key = f"rl:hybrid:{identifier}"
        # Push accumulated local count to Redis
        try:
            self._script(keys=[key], args=[to_sync, self.limit, self.window_seconds, now])
        except redis.RedisError:
            # If Redis is down, restore local count and fail open
            local.count += to_sync

# Envoy local rate limiting filter
- name: envoy.filters.http.local_ratelimit
  typed_config:
    "@type": type.googleapis.com/udpa.type.v1.TypedStruct
    type_url: type.googleapis.com/envoy.extensions.filters.http.local_ratelimit.v3.LocalRateLimit
    value:
      stat_prefix: http_local_rate_limiter
      token_bucket:
        max_tokens: 1000
        tokens_per_fill: 1000
        fill_interval: 1s
      filter_enabled:
        runtime_key: local_rate_limit_enabled
        default_value:
          numerator: 100
          denominator: HUNDRED
      response_headers_to_add:
        - append: false
          header:
            key: x-local-rate-limit
            value: "true"

# Envoy global rate limit config
rate_limits:
  - actions:
      - request_headers:
          header_name: x-user-id
          descriptor_key: user_id
      - request_headers:
          header_name: ":path"
          descriptor_key: path

# Istio EnvoyFilter for rate limiting
apiVersion: networking.istio.io/v1alpha3
kind: EnvoyFilter
metadata:
  name: filter-ratelimit
spec:
  configPatches:
    - applyTo: HTTP_FILTER
      match:
        context: SIDECAR_INBOUND
        listener:
          filterChain:
            filter:
              name: envoy.filters.network.http_connection_manager
      patch:
        operation: INSERT_BEFORE
        value:
          name: envoy.filters.http.ratelimit
          typed_config:
            "@type": type.googleapis.com/envoy.extensions.filters.http.ratelimit.v3.RateLimit
            domain: productpage-ratelimit
            failure_mode_deny: false # fail open
            rate_limit_service:
              grpc_service:
                envoy_grpc:
                  cluster_name: rate_limit_cluster
              transport_api_version: V3

def is_allowed(identifier: str) -> bool:
    try:
        result = redis.execute_rate_limit_check(identifier)
        return result.allowed
    except redis.RedisError:
        logger.error("Redis down! Failing open (allowing all requests)")
        return True  # Allow when Redis is unavailable

def is_allowed(identifier: str) -> bool:
    try:
        result = redis.execute_rate_limit_check(identifier)
        return result.allowed
    except redis.RedisError:
        logger.error("Redis down! Failing closed (denying all requests)")
        return False  # Deny when Redis is unavailable

import time
from enum import Enum
 
 
class CircuitState(Enum):
    CLOSED = "closed"      # Normal: using Redis
    OPEN = "open"          # Failure: using local fallback
    HALF_OPEN = "half_open"  # Testing: trying Redis again
 
 
class ResilientRateLimiter:
    """
    Rate limiter with circuit breaker pattern.
    Falls back to local in-memory limiting when Redis is unavailable.
    """
 
    def __init__(self, redis_limiter, local_limiter, failure_threshold=5, recovery_timeout=60):
        self.redis_limiter = redis_limiter
        self.local_limiter = local_limiter
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitState.CLOSED
        self.last_failure_time = None
 
    def is_allowed(self, identifier: str) -> dict:
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                # Circuit open: use local limiter as fallback
                return self.local_limiter.is_allowed(identifier)
 
        try:
            result = self.redis_limiter.is_allowed(identifier)
            # Success: reset circuit
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
            return result
 
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
 
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
                # Alert! Rate limiter is down
                # alert_system.send("Redis rate limiter DOWN, using local fallback")
 
            # Fallback to local limiting
            return self.local_limiter.is_allowed(identifier)