Bài 52: Rate Limiting — bảo vệ API khỏi abuse

Sau bài này bạn sẽ:

• ✅ Phân biệt được 4 thuật toán rate limiting và khi nào dùng cái nào
• ✅ Cài slowapi vào FastAPI, dùng Redis backend cho multi-instance
• ✅ Tự implement token bucket với Redis + asyncio
• ✅ Thiết kế tier-based limit và token-based limit cho LLM
• ✅ Trả HTTP 429 response đúng headers chuẩn
• ✅ Biết giới hạn của từng approach và common pitfalls

API thông thường không rate limit vẫn sống được (vì request rẻ). AI API thì khác — mỗi LLM call tốn tiền thật. Các lý do cụ thể:

Abuse prevention

Bot scrape API, script spam request, DDoS. Không có limit thì toàn bộ budget đi hết trong vài phút. Với AI inference endpoint, 1 bad actor có thể tiêu $500 OpenAI credit trước khi bạn phát hiện.

Cost control

User hợp lệ nhưng vô tình spam — ví dụ frontend gọi lại liên tục do bug. Mỗi LLM request gpt-4o tốn ~$0.005-0.015 tùy token count. 10,000 request/giờ = $50-150 chỉ từ 1 user. Rate limit cắt điều này sớm.

Fair use giữa nhiều user

Khi nhiều user share cùng resource (cùng API key OpenAI, cùng GPU), không có limit thì 1 user heavy có thể chiếm toàn bộ capacity, làm user khác timeout.

SLA enforcement theo tier

Free tier: 10 req/phút. Pro tier: 100 req/phút. Enterprise: 1000 req/phút. Đây là business logic cần enforce programmatically, không thể dựa vào user tự giác.

Bảo vệ trước provider-side limit

OpenAI, Anthropic đều có rate limit downstream (ví dụ 10,000 RPM với tier 1). Nếu để user gọi thẳng, bạn có thể vượt limit của provider và toàn bộ service bị 429 từ OpenAI. Limit user trước để tổng không bao giờ vượt quota của provider.

Fixed Window

Đếm số request trong cửa sổ cố định. Mỗi đầu cửa sổ reset counter về 0.

Window 60s: [request 1..100] → reset
                                              [request 1..100] → reset

Vấn đề: Burst tại biên cửa sổ. User gửi 100 request cuối giây 59, rồi 100 request đầu giây 60 = 200 request trong vòng 2 giây, nhưng cả hai window đều hợp lệ.

Ưu điểm: Đơn giản nhất, implement bằng INCR + EXPIRE trong Redis.

Sliding Window

Cửa sổ trượt theo thời gian thực, không reset tại mốc cố định. Mỗi lúc đếm, nhìn ngược lại đúng N giây.

Thời điểm T: đếm request từ [T-60s, T]
T+1s:        đếm request từ [T-59s, T+1s]  ← cửa sổ trượt

Ưu điểm: Không có burst tại biên.

Implementation: Lưu timestamp của từng request trong Redis sorted set, query theo range. Tốn memory hơn fixed window.

Token Bucket

Bucket có sức chứa tối đa N token. Token được nạp với tốc độ R token/giây. Mỗi request tiêu thụ 1 (hoặc nhiều hơn) token. Khi bucket rỗng → từ chối request.

Capacity: 50 token
Refill:   10 token/giây
Trạng thái: [■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■] 50
Burst 30 req: [■■■■■■■■■■■■■■■■■■■■] 20 còn lại
Chờ 3s → nạp thêm 30 → [■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■] 50

Đặc tính: Cho phép burst tối đa N request sau đó throttle về tốc độ R. Phổ biến nhất — AWS API Gateway, Stripe, Cloudflare đều dùng.

Leaky Bucket

Request đến → vào queue (capacity N). Queue "rò rỉ" (leak) ra ngoài với rate R cố định. Queue đầy → từ chối.

Request vào →  [■■■■■■■■■■] queue capacity 10
                              ↓ R request/giây (rate cố định)
                           processed

Đặc tính: Output rate hoàn toàn smooth, không có burst. Phù hợp khi downstream không chịu được spike (ví dụ limit chặt từ LLM provider). Nhược điểm: Request trong queue có thể phải đợi lâu.

So sánh nhanh

slowapi (v0.1.9+) là wrapper của limits library cho FastAPI, port từ Flask-Limiter. Dùng sliding window hoặc fixed window tùy backend.

Cài đặt

pip install slowapi
# Redis backend (cần cho multi-instance):
pip install redis

Setup cơ bản — limit theo IP

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

app = FastAPI()

# key_func xác định "ai" đang bị limit — mặc định là IP
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


@app.post("/predict")
@limiter.limit("10/minute")
async def predict(request: Request, body: dict):
    # request: Request phải có trong signature dù không dùng trực tiếp
    return {"result": "..."}

Quan trọng: Tham số request: Request phải có trong signature của endpoint được decorate. slowapi đọc metadata từ request object để đếm.

Các format limit string hợp lệ

"10/second"     # 10 request mỗi giây
"100/minute"    # 100 request mỗi phút
"1000/hour"     # 1000 request mỗi giờ
"5000/day"      # 5000 request mỗi ngày
"10 per second" # format thay thế, cùng kết quả

Áp dụng global limit

# Limit áp dụng cho toàn bộ app, không cần decorator từng route
limiter = Limiter(
    key_func=get_remote_address,
    default_limits=["200/minute"],  # mặc định cho mọi route
)

# Route muốn override:
@app.post("/chat")
@limiter.limit("20/minute")  # ghi đè default
async def chat(request: Request, body: dict):
    ...

# Route muốn exempt khỏi limit:
@app.get("/health")
@limiter.exempt
async def health():
    return {"status": "ok"}

Behavior khi vượt limit

_rate_limit_exceeded_handler mặc định trả HTTP 429 với body:

{"error": "Rate limit exceeded: 10 per 1 minute"}

Để tùy chỉnh response format:

from fastapi import Request
from fastapi.responses import JSONResponse
from slowapi.errors import RateLimitExceeded

async def custom_rate_limit_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
        status_code=429,
        content={
            "error": "rate_limit_exceeded",
            "message": f"Quota exceeded. Retry after {exc.retry_after} seconds.",
            "retry_after": exc.retry_after,
        },
        headers={"Retry-After": str(exc.retry_after)},
    )

app.add_exception_handler(RateLimitExceeded, custom_rate_limit_handler)

Mặc định slowapi dùng in-memory storage — chỉ hoạt động đúng với 1 process. Khi deploy nhiều instance (Kubernetes, multiple workers), mỗi instance có counter riêng. User bị limit N req/phút nhân số instance.

Cấu hình Redis backend

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis://localhost:6379",
    # Production có auth:
    # storage_uri="redis://:password@redis-host:6379/0",
)

Redis Sentinel / Cluster

# Redis Sentinel (HA)
limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis+sentinel://sentinel1:26379,sentinel2:26379/mymaster",
)

Kiểm tra Redis connection khi startup

from contextlib import asynccontextmanager
import redis.asyncio as aioredis

redis_client: aioredis.Redis = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_client
    redis_client = aioredis.from_url("redis://localhost:6379")
    try:
        await redis_client.ping()
    except Exception as e:
        raise RuntimeError(f"Cannot connect to Redis: {e}")
    yield
    await redis_client.aclose()

app = FastAPI(lifespan=lifespan)

Lưu ý: Khi Redis unavailable, slowapi mặc định fail open (cho phép request) để không block toàn bộ service. Behavior này configurable nhưng fail open thường đúng hơn fail close cho production.

Limit theo IP không đủ khi user dùng VPN hoặc nhiều user share 1 IP (công ty, university). Giải pháp đúng là limit theo user ID hoặc API key.

key_func đọc API key từ header

def get_api_key(request: Request) -> str:
    api_key = request.headers.get("X-API-Key")
    if not api_key:
        # Fallback về IP nếu không có API key
        # Hoặc raise để force authentication
        return get_remote_address(request)
    return api_key


limiter = Limiter(
    key_func=get_api_key,
    storage_uri="redis://localhost:6379",
)


@app.post("/chat")
@limiter.limit("100/hour")
async def chat(request: Request, body: dict):
    ...

key_func đọc user_id từ JWT

from jose import jwt, JWTError

SECRET_KEY = "..."  # Trong thực tế đọc từ env

def get_user_id_from_token(request: Request) -> str:
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        return get_remote_address(request)
    token = auth.removeprefix("Bearer ")
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload.get("sub", get_remote_address(request))
    except JWTError:
        return get_remote_address(request)


limiter = Limiter(
    key_func=get_user_id_from_token,
    storage_uri="redis://localhost:6379",
)

slowapi hỗ trợ dynamic limit string — limit có thể thay đổi theo user thay vì hardcode một giá trị.

from fastapi import Request
from slowapi import Limiter
from slowapi.util import get_remote_address

# Giả sử có hàm lookup user từ DB/cache
async def get_user_tier(user_id: str) -> str:
    # Thực tế: query DB hoặc Redis cache
    user = await db.users.find_one({"_id": user_id})
    return user["tier"] if user else "free"


def get_user_id(request: Request) -> str:
    return request.headers.get("X-API-Key", get_remote_address(request))


def dynamic_limit(key: str) -> str:
    """
    dynamic_limit nhận key (kết quả của key_func) không phải Request.
    Phải sync nếu dùng trực tiếp với slowapi.
    Cách an toàn: cache tier trong Redis, lookup sync.
    """
    import redis as sync_redis
    r = sync_redis.from_url("redis://localhost:6379")
    tier = r.get(f"user_tier:{key}")
    tier = tier.decode() if tier else "free"

    limits = {
        "free": "10/minute",
        "pro": "100/minute",
        "enterprise": "1000/minute",
    }
    return limits.get(tier, "10/minute")


limiter = Limiter(
    key_func=get_user_id,
    storage_uri="redis://localhost:6379",
)


@app.post("/chat")
@limiter.limit(dynamic_limit)
async def chat(request: Request, body: dict):
    ...

Cập nhật tier trong Redis khi user upgrade

@app.post("/admin/users/{user_id}/upgrade")
async def upgrade_user(user_id: str, new_tier: str):
    await redis_client.set(
        f"user_tier:{user_id}",
        new_tier,
        ex=86400,  # 24h TTL, refresh từ DB định kỳ
    )
    return {"upgraded": True}

Cách khác: dùng limits.storage.RedisStorage trực tiếp và viết middleware custom thay vì slowapi decorator — linh hoạt hơn nhưng code nhiều hơn.

Khi cần control chi tiết hơn slowapi cho phép, tự implement token bucket. Ví dụ dưới đây dùng redis.asyncio, phù hợp với FastAPI async.

Implementation đơn giản (có race condition)

import time
import redis.asyncio as aioredis

redis_client: aioredis.Redis = None  # khởi tạo trong lifespan


async def check_token_bucket(
    key: str,
    capacity: int,        # tối đa N token
    refill_per_sec: float # nạp R token mỗi giây
) -> bool:
    """
    Trả True nếu request được phép, False nếu bị rate limit.
    """
    redis_key = f"ratelimit:bucket:{key}"
    now = time.time()

    # Đọc trạng thái hiện tại
    async with redis_client.pipeline() as pipe:
        pipe.hget(redis_key, "tokens")
        pipe.hget(redis_key, "last_refill")
        tokens_str, last_refill_str = await pipe.execute()

    tokens = float(tokens_str) if tokens_str else float(capacity)
    last_refill = float(last_refill_str) if last_refill_str else now

    # Tính token được nạp thêm từ lần cuối
    elapsed = now - last_refill
    tokens = min(float(capacity), tokens + elapsed * refill_per_sec)

    if tokens < 1.0:
        return False  # bucket rỗng, từ chối

    tokens -= 1.0

    # Ghi lại trạng thái
    async with redis_client.pipeline() as pipe:
        pipe.hset(redis_key, mapping={"tokens": tokens, "last_refill": now})
        pipe.expire(redis_key, 3600)  # tự xóa sau 1h không dùng
        await pipe.execute()

    return True

Dùng trong endpoint

from fastapi import HTTPException

@app.post("/generate")
async def generate(request: Request, body: dict):
    api_key = request.headers.get("X-API-Key", request.client.host)

    allowed = await check_token_bucket(
        key=api_key,
        capacity=50,          # burst tối đa 50
        refill_per_sec=10.0,  # 10 req/giây trung bình
    )
    if not allowed:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded",
            headers={"Retry-After": "1"},
        )

    # ... xử lý request

Vấn đề race condition và Lua script

Implementation trên có race condition: hai request đồng thời có thể đọc cùng một giá trị tokens trước khi cái nào ghi. Trong production, dùng Lua script để đảm bảo atomicity:

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

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

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

if tokens < 1 then
    return 0
end

tokens = tokens - 1
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
redis.call('EXPIRE', key, 3600)
return 1
"""

# Đăng ký script khi startup
token_bucket_sha = None

async def init_lua_script():
    global token_bucket_sha
    token_bucket_sha = await redis_client.script_load(TOKEN_BUCKET_SCRIPT)


async def check_token_bucket_atomic(key: str, capacity: int, refill_per_sec: float) -> bool:
    now = time.time()
    result = await redis_client.evalsha(
        token_bucket_sha,
        1,  # số KEYS
        f"ratelimit:bucket:{key}",
        capacity,
        refill_per_sec,
        now,
    )
    return bool(result)

Lua script chạy atomic trong Redis — không có race condition dù nhiều process gọi đồng thời.

RFC 6585 định nghĩa status code 429 và các headers khuyến nghị. Client cần headers này để retry đúng cách.

Headers chuẩn

Trả headers trên mọi response (không chỉ 429)

Client cần biết remaining quota trước khi bị limit để điều chỉnh tốc độ gửi. Thêm headers vào mọi response:

from fastapi import Response

@app.post("/chat")
async def chat(request: Request, response: Response, body: dict):
    api_key = request.headers.get("X-API-Key", request.client.host)

    # Lấy trạng thái bucket hiện tại
    bucket_info = await get_bucket_info(api_key)

    response.headers["X-RateLimit-Limit"] = str(bucket_info["capacity"])
    response.headers["X-RateLimit-Remaining"] = str(int(bucket_info["tokens"]))
    response.headers["X-RateLimit-Reset"] = str(bucket_info["reset_at"])

    if bucket_info["tokens"] < 1:
        response.headers["Retry-After"] = str(bucket_info["retry_after"])
        raise HTTPException(status_code=429, detail="Rate limit exceeded")

    # ... xử lý

Response body chuẩn khi 429

{
  "error": "rate_limit_exceeded",
  "message": "You have exceeded the rate limit of 100 requests per minute.",
  "limit": 100,
  "window": "60s",
  "retry_after": 30
}

Rate limit ở tầng application (FastAPI) xử lý sau khi request đã vào process. Limit ở nginx xử lý trước — rẻ hơn nhiều, không tốn CPU của Python process.

Cấu hình nginx

http {
    # Định nghĩa zone: lưu state cho 10MB, rate 10 req/giây theo IP
    limit_req_zone $binary_remote_addr zone=api_zone:10m rate=10r/s;

    server {
        listen 80;

        location /api/ {
            # burst=20: cho phép burst 20 req, nodelay: không delay burst
            limit_req zone=api_zone burst=20 nodelay;

            # Trả 429 thay vì default 503
            limit_req_status 429;

            proxy_pass http://backend:8000;
        }

        # Endpoint không cần limit:
        location /health {
            proxy_pass http://backend:8000;
        }
    }
}

Giải thích tham số

• rate=10r/s: tốc độ tối đa 10 request/giây (leaky bucket algorithm bên trong nginx).
• burst=20: cho phép queue 20 request vượt rate. Khi queue đầy → từ chối.
• nodelay: xử lý burst ngay lập tức thay vì delay từng request xuống 10r/s. Không có nodelay thì burst sẽ được delay dần → user thấy latency cao.
• $binary_remote_addr: key theo IP dạng binary (compact hơn string).

Trade-off của nginx rate limit

• Ưu điểm: Không cần Redis, không tốn app process, bảo vệ trước cả khi app overload.
• Nhược điểm: Chỉ limit theo IP, không biết user_id hay API key. Nhiều user sau 1 reverse proxy hay load balancer có thể bị limit oan. VPN users thường share IP.

Trong practice, kết hợp cả hai: nginx limit theo IP để chặn bot/DDoS, app limit theo user_id để enforce business rule.

Khi không muốn tự quản lý Redis và nginx config, có thể dùng managed solutions:

AWS API Gateway + usage plan là lựa chọn phổ biến khi đã deploy trên AWS. Không cần thêm infrastructure, rate limit + quota tự động trả 429 với Retry-After.

Request count không đủ cho AI API. 1 request 10 token và 1 request 100,000 token có chi phí chênh nhau 10,000 lần. Giải pháp: limit theo token count hoặc dollar cost.

Token-based limit

import datetime
import tiktoken  # pip install tiktoken

encoder = tiktoken.encoding_for_model("gpt-4o")

TOKEN_LIMIT_PER_HOUR = {
    "free": 50_000,
    "pro": 500_000,
    "enterprise": 5_000_000,
}


async def check_token_limit(user_id: str, tier: str, estimated_tokens: int) -> bool:
    now = datetime.datetime.utcnow()
    hour_key = now.strftime("%Y%m%d%H")
    redis_key = f"token_usage:{user_id}:{hour_key}"

    current_usage = await redis_client.get(redis_key)
    current_usage = int(current_usage) if current_usage else 0

    limit = TOKEN_LIMIT_PER_HOUR.get(tier, TOKEN_LIMIT_PER_HOUR["free"])

    if current_usage + estimated_tokens > limit:
        return False
    return True


async def record_token_usage(user_id: str, actual_tokens: int) -> None:
    now = datetime.datetime.utcnow()
    hour_key = now.strftime("%Y%m%d%H")
    redis_key = f"token_usage:{user_id}:{hour_key}"

    await redis_client.incrby(redis_key, actual_tokens)
    await redis_client.expire(redis_key, 7200)  # TTL 2h


@app.post("/chat")
async def chat(request: Request, body: dict):
    user_id = request.headers.get("X-API-Key", request.client.host)
    tier = await get_user_tier(user_id)

    # Estimate token trước khi gọi LLM
    messages = body.get("messages", [])
    prompt_text = " ".join(m["content"] for m in messages)
    estimated = len(encoder.encode(prompt_text)) + 500  # buffer cho completion

    if not await check_token_limit(user_id, tier, estimated):
        raise HTTPException(
            status_code=429,
            detail="Hourly token limit exceeded",
            headers={"Retry-After": str(3600 - datetime.datetime.utcnow().minute * 60)},
        )

    # Gọi LLM
    response = await call_llm(messages)
    actual_tokens = response.usage.total_tokens

    # Ghi actual usage sau khi có kết quả
    await record_token_usage(user_id, actual_tokens)

    return {"content": response.choices[0].message.content}

Cost-based limit

Tương tự nhưng track bằng đơn vị dollar thay vì token. Hữu ích khi mix nhiều model có giá khác nhau:

MODEL_COST_PER_1K_TOKENS = {
    "gpt-4o": 0.005,          # $5/1M input tokens
    "gpt-4o-mini": 0.00015,   # $0.15/1M input tokens
    "claude-3-5-sonnet": 0.003,
}

async def record_cost(user_id: str, model: str, tokens: int) -> float:
    cost = (tokens / 1000) * MODEL_COST_PER_1K_TOKENS.get(model, 0.005)
    today = datetime.datetime.utcnow().strftime("%Y%m%d")
    redis_key = f"cost_usage:{user_id}:{today}"

    # Lưu dạng float nguyên cent để tránh float precision issue
    cost_cents = int(cost * 100)
    await redis_client.incrby(redis_key, cost_cents)
    await redis_client.expire(redis_key, 172800)  # 2 ngày

    return cost

Rate limit đếm request theo thời gian (req/phút). Concurrent limit đếm số request đang xử lý cùng lúc. Hai loại này bổ sung nhau.

Use case: LLM streaming response mất 10-30 giây. Một user mở 100 connection đồng thời → 100 connection giữ 100 slot thread/goroutine, block user khác. Rate limit không bắt được điều này vì đó là 100 request khác nhau.

Semaphore per user với Redis

MAX_CONCURRENT = {"free": 2, "pro": 10, "enterprise": 50}


async def acquire_concurrent_slot(user_id: str, tier: str) -> bool:
    """Trả True nếu còn slot, False nếu đã đủ concurrent."""
    max_slots = MAX_CONCURRENT.get(tier, 2)
    redis_key = f"concurrent:{user_id}"

    # Atomic increment + check
    current = await redis_client.incr(redis_key)
    if current == 1:
        # Key mới tạo, set TTL phòng crash không release
        await redis_client.expire(redis_key, 120)

    if current > max_slots:
        await redis_client.decr(redis_key)
        return False
    return True


async def release_concurrent_slot(user_id: str) -> None:
    redis_key = f"concurrent:{user_id}"
    count = await redis_client.decr(redis_key)
    if count <= 0:
        await redis_client.delete(redis_key)


@app.post("/stream-chat")
async def stream_chat(request: Request, body: dict):
    user_id = request.headers.get("X-API-Key", request.client.host)
    tier = await get_user_tier(user_id)

    if not await acquire_concurrent_slot(user_id, tier):
        raise HTTPException(
            status_code=429,
            detail=f"Too many concurrent requests. Max {MAX_CONCURRENT[tier]}.",
        )

    try:
        # ... streaming LLM call
        pass
    finally:
        # PHẢI release trong finally để không bị leak khi exception
        await release_concurrent_slot(user_id)

Dù đã limit phía user, vẫn có thể nhận 429 từ OpenAI/Anthropic do nhiều nguyên nhân: traffic spike đồng thời, tier RPM thấp, hoặc provider tạm thời giảm limit. Cần retry với exponential backoff.

from tenacity import (
    retry,
    retry_if_exception_type,
    wait_exponential,
    stop_after_attempt,
    before_sleep_log,
)
import openai
import logging

logger = logging.getLogger(__name__)


@retry(
    retry=retry_if_exception_type(openai.RateLimitError),
    wait=wait_exponential(multiplier=1, min=4, max=60),
    stop=stop_after_attempt(5),
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
async def call_openai_with_retry(messages: list) -> openai.types.chat.ChatCompletion:
    client = openai.AsyncOpenAI()
    return await client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
    )

Giải thích tham số wait_exponential:

• multiplier=1: hệ số nhân.
• min=4: chờ tối thiểu 4 giây lần đầu.
• max=60: chờ tối đa 60 giây.
• Sequence: 4s → 8s → 16s → 32s → 60s.

Đọc Retry-After từ response header của provider

OpenAI trả Retry-After trong header khi 429. tenacity không tự đọc header này. Nếu cần chính xác hơn:

import asyncio

async def call_openai_respecting_header(messages: list):
    client = openai.AsyncOpenAI()
    for attempt in range(5):
        try:
            return await client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
            )
        except openai.RateLimitError as e:
            if attempt == 4:
                raise
            retry_after = int(e.response.headers.get("Retry-After", 4))
            await asyncio.sleep(retry_after)

Rate limit event cần được track để phân biệt: user bị limit hợp lệ (heavy use) hay đang bị tấn công, hay limit của bạn quá chặt.

Prometheus metrics

from prometheus_client import Counter, Histogram

rate_limit_hits = Counter(
    "ratelimit_hits_total",
    "Number of requests rejected by rate limiter",
    labelnames=["user_tier", "endpoint", "limit_type"],
)

rate_limit_remaining = Histogram(
    "ratelimit_remaining_tokens",
    "Remaining tokens in rate limit bucket at request time",
    labelnames=["user_tier"],
    buckets=[0, 1, 5, 10, 25, 50, 100, 250, 500],
)


# Trong exception handler hoặc middleware:
async def custom_rate_limit_handler(request: Request, exc: RateLimitExceeded):
    tier = request.state.user_tier if hasattr(request.state, "user_tier") else "unknown"
    rate_limit_hits.labels(
        user_tier=tier,
        endpoint=request.url.path,
        limit_type="request_count",
    ).inc()
    return JSONResponse(status_code=429, content={"error": "rate_limit_exceeded"})

Alerts cần thiết

• High 429 rate: >10% request trả 429 trong 5 phút → có thể đang bị attack hoặc limit quá chặt.
• Spike từ 1 user_id: 1 user tạo ra >50% tổng 429 → suspicious, có thể là bot.
• Provider 429: Nhận 429 từ OpenAI/Anthropic → cần review tier hoặc reduce traffic.

Grafana dashboard

Metrics hữu ích để visualize:

• rate(ratelimit_hits_total[5m]) — rate limit hit per second.
• ratelimit_hits_total group by user_tier — xem tier nào bị nhiều nhất.
• histogram_quantile(0.95, ratelimit_remaining_tokens) — P95 remaining tokens cho thấy buffer còn bao nhiêu.

1. In-memory limit khi chạy nhiều instance

Triệu chứng: user gửi 100 req/phút nhưng bị limit ở 10 req/phút trên mỗi instance. Với 5 instance, limit thực tế là 50 req/phút. Giải pháp: luôn dùng Redis backend khi deploy hơn 1 process.

2. Rate limit chỉ ở app, không ở proxy

Traffic đến load balancer, được phân phối đều. Nếu 1 instance crash, traffic dồn sang instance còn lại — instance đó chưa biết limit cũ. Redis backend giải quyết vì state tập trung.

3. Quên handle 429 ở client

Frontend nhận 429 nhưng không đọc Retry-After, retry ngay lập tức → vòng lặp 429. Cần implement exponential backoff ở client khi nhận 429.

4. Limit theo IP với nhiều user share 1 IP

Office, university, VPN — nhiều user share 1 IP. Limit theo IP sẽ limit tất cả cùng lúc. Giải pháp: luôn limit theo user_id/API key ở application layer; IP limit chỉ dùng ở nginx layer để chặn bot.

5. Token estimate không chính xác

tiktoken estimate tốt với OpenAI models nhưng không chính xác 100% với special tokens, formatting, v.v. Nên add buffer 10-20% và reconcile sau khi có actual usage từ response. Có thể dẫn đến over-limit (cho phép request nhưng thực tế vượt quota) hoặc under-limit (reject request hợp lệ).

6. Concurrent slot không được release khi exception

Nếu không dùng try/finally trong concurrent limit implementation, exception làm slot bị chiếm mãi mãi (cho đến khi TTL expire). Luôn release trong finally.

7. Limit tập trung ở tầng quá thấp hoặc quá cao

Chỉ nginx limit → không phân biệt được user, không enforce business tier. Chỉ app limit → tốn process, không chặn được DDoS. Tốt nhất là cả hai: nginx chặn IP abuse ở edge, app enforce business rule theo user_id.

Bài 53: API Key management — không hard-code, dùng env / secret manager