Rate Limiting Distribuído: token bucket, sliding window, Redis

# fixed_window.py
import time
import redis.asyncio as redis

r = redis.from_url("redis://localhost")

async def allow(client_id: str, limit: int = 100, window_s: int = 60) -> bool:
    now_window = int(time.time()) // window_s
    key = f"rate:{client_id}:{now_window}"

    async with r.pipeline(transaction=True) as pipe:
        pipe.incr(key)
        pipe.expire(key, window_s)        # idempotente
        count, _ = await pipe.execute()

    return count <= limit

# sliding_window_log.py — preciso mas mais memória
import time
import redis.asyncio as redis
import uuid

r = redis.from_url("redis://localhost")

async def allow(client_id: str, limit: int = 100, window_s: int = 60) -> bool:
    now_ms = int(time.time() * 1000)
    window_start = now_ms - (window_s * 1000)
    key = f"rate:log:{client_id}"

    async with r.pipeline(transaction=True) as pipe:
        pipe.zremrangebyscore(key, 0, window_start)     # expire old
        pipe.zcard(key)                                  # count
        pipe.zadd(key, {f"{now_ms}:{uuid.uuid4()}": now_ms})  # add new
        pipe.expire(key, window_s)
        _, count, _, _ = await pipe.execute()

    return count < limit  # < pq o novo já foi adicionado

# sliding_window_counter.py
import time

async def allow(client_id: str, limit: int = 100, window_s: int = 60) -> bool:
    now = time.time()
    cur_win = int(now) // window_s
    prev_win = cur_win - 1
    percent_through = (now % window_s) / window_s

    cur_key = f"rate:sw:{client_id}:{cur_win}"
    prev_key = f"rate:sw:{client_id}:{prev_win}"

    cur_count = int(await r.get(cur_key) or 0)
    prev_count = int(await r.get(prev_key) or 0)

    estimated = prev_count * (1 - percent_through) + cur_count

    if estimated >= limit:
        return False

    await r.incr(cur_key)
    await r.expire(cur_key, window_s * 2)   # manter visível 1 janela extra
    return True

-- token_bucket.lua — atômico via EVAL no Redis
-- KEYS[1] = chave do bucket
-- ARGV[1] = capacidade (C)
-- ARGV[2] = refill rate (R tokens/seg)
-- ARGV[3] = now (unix seconds, com decimais)
-- ARGV[4] = tokens requested (tipicamente 1)

local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])

local data = redis.call('HMGET', key, 'tokens', 'last')
local tokens = tonumber(data[1])
local last = tonumber(data[2])

if tokens == nil then
  tokens = capacity
  last = now
end

-- Refill: adiciona tokens proporcional ao tempo passado, cap em capacity
local elapsed = math.max(0, now - last)
tokens = math.min(capacity, tokens + elapsed * refill_rate)

local allowed = 0
if tokens >= requested then
  tokens = tokens - requested
  allowed = 1
end

redis.call('HMSET', key, 'tokens', tokens, 'last', now)
redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) + 1)

return {allowed, tokens}

# token_bucket_client.py — chamando o script
import time
import redis.asyncio as redis

r = redis.from_url("redis://localhost")

# Carrega uma vez, reusa via EVALSHA
with open("token_bucket.lua") as f:
    TOKEN_BUCKET_SHA = None

async def init_script():
    global TOKEN_BUCKET_SHA
    with open("token_bucket.lua") as f:
        TOKEN_BUCKET_SHA = await r.script_load(f.read())

async def allow(client_id: str, capacity: int = 10, rate: float = 5.0) -> tuple[bool, float]:
    now = time.time()
    key = f"rate:bucket:{client_id}"
    allowed, remaining = await r.evalsha(
        TOKEN_BUCKET_SHA, 1, key, capacity, rate, now, 1
    )
    return bool(allowed), remaining

# leaky_bucket.py — versão educativa em memória
from collections import deque
import time

class LeakyBucket:
    def __init__(self, capacity: int, rate_per_s: float):
        self.capacity = capacity
        self.rate = rate_per_s
        self.queue = deque()
        self.last_leak = time.time()

    def allow(self) -> bool:
        now = time.time()
        # Leak: remove itens proporcional ao tempo passado
        to_leak = int((now - self.last_leak) * self.rate)
        for _ in range(min(to_leak, len(self.queue))):
            self.queue.popleft()
        self.last_leak = now

        if len(self.queue) >= self.capacity:
            return False
        self.queue.append(now)
        return True

HTTP/1.1 429 Too Many Requests
Content-Type: application/json
Retry-After: 30
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1744803660

{
  "error": "rate_limit_exceeded",
  "message": "Too many requests. Try again in 30 seconds.",
  "retry_after": 30
}