# Redis distributed lock with Lua scripts
async with RedisLock(redis_client, "payment:order-123"):
    await process_payment(order_id)

# Circuit breaker for external APIs
@circuit_breaker(failure_threshold=5, recovery_timeout=30)
@retry(max_attempts=3, base_delay=1.0)
async def call_external_api():
    ...

# Idempotent API endpoint
@router.post("/payments")
async def create_payment(
    data: PaymentCreate,
    idempotency_key: str = Header(..., alias="Idempotency-Key"),
):
    return await idempotent_execute(db, idempotency_key, "/payments", process)

# Token bucket rate limiting
limiter = TokenBucketLimiter(redis_client, capacity=100, refill_rate=10)
if await limiter.is_allowed(f"user:{user_id}"):
    await handle_request()

# LOCKS: Never forget TTL (causes deadlocks)
await redis.set(f"lock:{name}", "1")  # WRONG - no expiry!

# LOCKS: Never release without owner check
await redis.delete(f"lock:{name}")  # WRONG - might release others' lock

# RESILIENCE: Never retry non-retryable errors
@retry(max_attempts=5, retryable_exceptions={Exception})  # Retries 401!

# RESILIENCE: Never put retry outside circuit breaker
@retry  # Would retry when circuit is open!
@circuit_breaker
async def call(): ...

# IDEMPOTENCY: Never use non-deterministic keys
key = str(uuid.uuid4())  # Different every time!

# IDEMPOTENCY: Never cache error responses
if response.status_code >= 400:
    await cache_response(key, response)  # Errors should retry!

# RATE LIMITING: Never use in-memory counters in distributed systems
request_counts = {}  # Lost on restart, not shared across instances

// WRONG: Cache forever, no way to update
export default {
  async fetch(request: Request) {
    const cache = caches.default;
    const cached = await cache.match(request);
    if (cached) return cached; // Stale forever!

    const response = await fetch(request);
    await cache.put(request, response.clone()); // No expiry!
    return response;
  }
};

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    const cacheKey = new Request(url.toString(), request);
    const cache = caches.default;

    // Check edge cache
    let response = await cache.match(cacheKey);
    if (response) return response;

    // Check KV (global, eventually consistent)
    const kvData = await env.CACHE_KV.get(url.pathname, 'text');
    if (kvData) {
      response = new Response(kvData, {
        headers: {
          'Content-Type': 'application/json',
          'Cache-Control': 'public, max-age=60, stale-while-revalidate=300',
          'CDN-Cache-Control': 'max-age=3600', // CDN caches longer
        },
      });
      // Populate edge cache
      await cache.put(cacheKey, response.clone());
      return response;
    }

    // Fetch from origin
    response = await fetch(request);
    const body = await response.text();

    // Store in KV with TTL
    await env.CACHE_KV.put(url.pathname, body, { expirationTtl: 3600 });

    const cachedResponse = new Response(body, {
      headers: {
        'Content-Type': 'application/json',
        'Cache-Control': 'public, max-age=60, stale-while-revalidate=300',
      },
    });
    await cache.put(cacheKey, cachedResponse.clone());
    return cachedResponse;
  }
};

# Static assets (hashed filenames)
Cache-Control: public, max-age=31536000, immutable

# API responses (frequently changing)
Cache-Control: public, max-age=60, stale-while-revalidate=300

# Personalized content
Cache-Control: private, no-cache
Vary: Cookie, Authorization

// WRONG: Node.js APIs not available in edge runtime
import fs from 'fs';
import { createHash } from 'crypto';

export default async function handler(req: Request) {
  const data = fs.readFileSync('./config.json'); // FAILS at edge
  const hash = createHash('sha256').update(data); // FAILS at edge
  return new Response(hash.digest('hex'));
}

// Cloudflare Worker with Web Crypto API
export default {
  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
    const country = (request as any).cf?.country || 'US';

    // Web Crypto API (available at edge)
    const token = request.headers.get('Authorization')?.replace('Bearer ', '');
    if (token) {
      const key = await crypto.subtle.importKey(
        'raw',
        new TextEncoder().encode(SECRET),
        { name: 'HMAC', hash: 'SHA-256' },
        false,
        ['verify']
      );
    }

    // Geo-based routing
    if (country === 'EU') {
      return Response.redirect(`https://eu.example.com${url.pathname}`);
    }

    return fetch(request);
  }
};

// middleware.ts (Next.js Edge Middleware)
import { NextResponse } from 'next/server';
import type { NextRequest } from 'next/server';

export const config = { matcher: ['/dashboard/:path*'] };

export function middleware(request: NextRequest) {
  const token = request.cookies.get('session');
  if (!token) {
    return NextResponse.redirect(new URL('/login', request.url));
  }
  // A/B testing via cookie
  const bucket = request.cookies.get('ab-bucket')?.value || 'control';
  return NextResponse.rewrite(new URL(`/${bucket}${request.nextUrl.pathname}`, request.url));
}

# WRONG: If publish fails, DB has data but event is lost
async def create_order(order: Order):
    await db.insert(order)              # Step 1: succeeds
    await message_broker.publish(        # Step 2: might fail!
        "order.created", order.dict()
    )
    # If publish fails: order exists but no event was sent
    # Downstream services never know about the order

# Outbox table: events written atomically with business data
async def create_order(order: Order, db: AsyncSession):
    async with db.begin():
        # Both in same transaction — atomic!
        db.add(order)
        db.add(OutboxEvent(
            aggregate_id=order.id,
            event_type="order.created",
            payload=order.dict(),
            created_at=datetime.utcnow(),
        ))

# Outbox publisher (separate process, polls for unsent events)
async def publish_outbox_events(db: AsyncSession, broker: MessageBroker):
    while True:
        async with db.begin():
            events = await db.execute(
                select(OutboxEvent)
                .where(OutboxEvent.published_at.is_(None))
                .order_by(OutboxEvent.created_at)
                .limit(100)
                .with_for_update(skip_locked=True)  # Concurrent workers safe
            )
            for event in events.scalars():
                await broker.publish(event.event_type, event.payload)
                event.published_at = datetime.utcnow()
        await asyncio.sleep(1)

class OrderSaga:
    steps = [
        SagaStep("reserve_inventory", compensate="release_inventory"),
        SagaStep("charge_payment", compensate="refund_payment"),
        SagaStep("ship_order", compensate="cancel_shipment"),
    ]

    async def execute(self, order_id: str):
        completed = []
        for step in self.steps:
            try:
                await step.execute(order_id)
                completed.append(step)
            except Exception:
                # Compensate in reverse order
                for s in reversed(completed):
                    await s.compensate(order_id)
                raise SagaFailed(f"Failed at {step.name}")

async def handle_event(event: Event, db: AsyncSession):
    async with db.begin():
        # Check if already processed
        exists = await db.execute(
            select(ProcessedEvent).where(ProcessedEvent.event_id == event.id)
        )
        if exists.scalar():
            return  # Already processed, skip

        # Process the event
        await process_order(event.payload)

        # Mark as processed
        db.add(ProcessedEvent(event_id=event.id, processed_at=datetime.utcnow()))

# WRONG: Direct mutation loses history
class Account:
    def __init__(self, balance: float = 0):
        self.balance = balance

    def deposit(self, amount: float):
        self.balance += amount  # No record of what happened!

    def withdraw(self, amount: float):
        self.balance -= amount  # Can't audit, can't replay

from dataclasses import dataclass, field
from typing import Any

@dataclass
class DomainEvent:
    aggregate_id: str
    version: int
    data: dict[str, Any]

class Account:
    def __init__(self):
        self._changes: list[DomainEvent] = []
        self._version = 0
        self.balance = 0.0

    def deposit(self, amount: float):
        if amount <= 0:
            raise ValueError("Amount must be positive")
        self._raise_event("MoneyDeposited", {"amount": amount})

    def withdraw(self, amount: float):
        if amount > self.balance:
            raise ValueError("Insufficient funds")
        self._raise_event("MoneyWithdrawn", {"amount": amount})

    def _raise_event(self, event_type: str, data: dict):
        event = DomainEvent(
            aggregate_id=self.id,
            version=self._version + 1,
            data={"type": event_type, **data},
        )
        self._apply(event)
        self._changes.append(event)

    def _apply(self, event: DomainEvent):
        match event.data["type"]:
            case "MoneyDeposited":
                self.balance += event.data["amount"]
            case "MoneyWithdrawn":
                self.balance -= event.data["amount"]
        self._version = event.version

# Event store with optimistic concurrency
class EventStore:
    async def save(self, aggregate_id: str, events: list[DomainEvent], expected_version: int):
        async with self.db.transaction():
            current = await self.db.fetchval(
                "SELECT MAX(version) FROM events WHERE aggregate_id = $1",
                aggregate_id,
            )
            if current != expected_version:
                raise ConcurrencyError(f"Expected {expected_version}, got {current}")
            for event in events:
                await self.db.execute(
                    "INSERT INTO events (aggregate_id, version, data) VALUES ($1, $2, $3)",
                    aggregate_id, event.version, event.data,
                )

# CQRS: Separate read model projection
class BalanceProjection:
    async def handle(self, event: DomainEvent):
        match event.data["type"]:
            case "MoneyDeposited":
                await self.db.execute(
                    "UPDATE account_balances SET balance = balance + $1 WHERE id = $2",
                    event.data["amount"], event.aggregate_id,
                )
            case "MoneyWithdrawn":
                await self.db.execute(
                    "UPDATE account_balances SET balance = balance - $1 WHERE id = $2",
                    event.data["amount"], event.aggregate_id,
                )

CREATE TABLE idempotency_records (
    idempotency_key VARCHAR(64) PRIMARY KEY,
    endpoint VARCHAR(256) NOT NULL,
    request_hash VARCHAR(64) NOT NULL,
    response_body TEXT,
    response_status INTEGER DEFAULT 200,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    expires_at TIMESTAMPTZ NOT NULL
);

CREATE INDEX ix_idempotency_expires ON idempotency_records (expires_at);
CREATE INDEX ix_idempotency_endpoint_key ON idempotency_records (endpoint, idempotency_key);

from sqlalchemy import text
from sqlalchemy.dialects.postgresql import insert
from datetime import UTC, datetime, timedelta
import json

async def idempotent_execute(
    db: AsyncSession,
    idempotency_key: str,
    endpoint: str,
    operation,
    ttl_hours: int = 24,
) -> tuple[Any, int, bool]:
    """Execute operation idempotently. Returns (response, status, was_replayed)."""
    # Check for existing
    existing = await db.get(ProcessedRequest, idempotency_key)
    if existing and existing.expires_at > datetime.now(UTC):
        return json.loads(existing.response_body), existing.status_code, True

    # Execute operation
    result, status_code = await operation()

    # Store result (upsert to handle races)
    stmt = insert(ProcessedRequest).values(
        idempotency_key=idempotency_key,
        endpoint=endpoint,
        status_code=status_code,
        response_body=json.dumps(result),
        expires_at=datetime.now(UTC) + timedelta(hours=ttl_hours),
    ).on_conflict_do_nothing()

    await db.execute(stmt)
    return result, status_code, False

class IdempotencyService:
    def _hash_request(self, body: dict) -> str:
        return hashlib.sha256(
            json.dumps(body, sort_keys=True, default=str).encode()
        ).hexdigest()

    async def check_idempotency(self, key: str, endpoint: str, body: dict):
        row = await self.db.execute(
            text("SELECT request_hash, response_body, response_status "
                 "FROM idempotency_records "
                 "WHERE idempotency_key = :key AND endpoint = :endpoint"),
            {"key": key, "endpoint": endpoint},
        )
        existing = row.fetchone()
        if not existing:
            return None

        if existing.request_hash != self._hash_request(body):
            raise HTTPException(
                status_code=422,
                detail="Idempotency key reused with different request body",
            )
        return {"body": json.loads(existing.response_body),
                "status": existing.response_status, "replayed": True}

async def cleanup_expired_records(db: AsyncSession, batch_size: int = 1000) -> int:
    """Delete expired idempotency records. Run daily via scheduler."""
    total_deleted = 0
    while True:
        result = await db.execute(
            text("""
                DELETE FROM idempotency_records
                WHERE idempotency_key IN (
                    SELECT idempotency_key FROM idempotency_records
                    WHERE expires_at < NOW()
                    LIMIT :batch_size
                )
            """),
            {"batch_size": batch_size},
        )
        await db.commit()
        deleted = result.rowcount
        total_deleted += deleted
        if deleted < batch_size:
            break
        await asyncio.sleep(0.1)  # Reduce DB load
    return total_deleted

@app.post("/api/orders", status_code=201)
async def create_order(
    order: OrderCreate,
    idempotency_key: str = Header(..., alias="Idempotency-Key"),
    db: AsyncSession = Depends(get_db),
):
    async def process():
        new_order = Order(**order.model_dump())
        db.add(new_order)
        await db.commit()
        return {"order_id": str(new_order.id)}, 201

    response, status, replayed = await idempotent_execute(
        db=db,
        idempotency_key=idempotency_key,
        endpoint="/api/orders",
        operation=process,
    )
    return JSONResponse(
        content=response, status_code=status,
        headers={"Idempotent-Replayed": "true"} if replayed else {},
    )

const existing = await db.query("SELECT * FROM idempotency_records WHERE key = $1", [key]);
if (!existing) {
  await processPayment(data);  // Race! Two processes both see "not existing"
  await db.query("INSERT INTO idempotency_records (key, result) VALUES ($1, $2)", [key, result]);
}

try {
  await db.query("INSERT INTO idempotency_records (key, status) VALUES ($1, 'processing')", [key]);
  const result = await processPayment(data);
  await db.query("UPDATE idempotency_records SET result = $1 WHERE key = $2", [result, key]);
} catch (UniqueViolationError) {
  return await db.query("SELECT result FROM idempotency_records WHERE key = $1", [key]);
}

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

class IdempotentConsumer:
    """Process events exactly once using idempotency keys."""

    def __init__(self, db: AsyncSession, redis: redis.Redis):
        self.db = db
        self.redis = redis

    async def process(self, event: dict, handler) -> tuple[Any, bool]:
        """Process event idempotently. Returns (result, was_duplicate)."""
        idempotency_key = event.get("idempotency_key")
        if not idempotency_key:
            return await handler(event), False

        # Fast path: check Redis cache
        cache_key = f"processed:{idempotency_key}"
        if await self.redis.exists(cache_key):
            return None, True

        # Slow path: check database
        existing = await self.db.execute(
            select(ProcessedEvent)
            .where(ProcessedEvent.idempotency_key == idempotency_key)
        )
        if existing.scalar_one_or_none():
            await self.redis.setex(cache_key, 86400, "1")  # Backfill cache
            return None, True

        # Process with database lock to prevent races
        try:
            async with self.db.begin_nested():
                self.db.add(ProcessedEvent(idempotency_key=idempotency_key))
                await self.db.flush()
                result = await handler(event)

            await self.redis.setex(cache_key, 86400, "1")
            return result, False

        except IntegrityError:
            return None, True  # Another process claimed it

async def consume_orders(processor: IdempotentEventProcessor):
    consumer = AIOKafkaConsumer("orders", bootstrap_servers="localhost:9092")
    await consumer.start()

    try:
        async for msg in consumer:
            event = json.loads(msg.value)
            result, was_duplicate = await processor.process_event(
                event_id=event["event_id"],
                event_type="order.created",
                handler=handle_order_created,
                order_data=event["data"],
            )
            if was_duplicate:
                logger.info(f"Skipped duplicate: {event['event_id']}")
    finally:
        await consumer.stop()

class IdempotentEventProcessor:
    """Track processed events in database for exactly-once semantics."""

    async def process_event(self, event_id: str, event_type: str,
                           handler, *args, **kwargs) -> tuple:
        async with self.pool.acquire() as conn:
            async with conn.transaction():
                try:
                    await conn.execute(
                        "INSERT INTO processed_events (event_id, event_type) VALUES ($1, $2)",
                        event_id, event_type,
                    )
                except UniqueViolationError:
                    existing = await conn.fetchrow(
                        "SELECT result FROM processed_events WHERE event_id = $1",
                        event_id,
                    )
                    return existing["result"], True

                result = await handler(*args, **kwargs)
                await conn.execute(
                    "UPDATE processed_events SET result = $1 WHERE event_id = $2",
                    json.dumps(result) if result else None, event_id,
                )
                return result, False

# NEVER check-then-act without locking
async def bad_process(key):
    if not await exists(key):  # Race condition!
        await process()
        await mark_processed(key)

# CORRECT: Insert first (claims the key atomically)
async def good_process(key):
    try:
        await insert_processed(key)  # Atomic claim
        await process()
    except IntegrityError:
        pass  # Already processed

async def process_event(event_id: str):
    if await redis.exists(f"processed:{event_id}"):
        return  # Already processed
    await handle_event(event_id)
    await redis.set(f"processed:{event_id}", "1", ex=86400)
# Redis restart = all events reprocessed!

async def process_event(event_id: str):
    # Fast path: Redis check
    if await redis.exists(f"processed:{event_id}"):
        return
    # Durable check: Database with unique constraint
    try:
        await db.execute("INSERT INTO processed_events (event_id) VALUES ($1)", event_id)
        await handle_event(event_id)
        await redis.set(f"processed:{event_id}", "1", ex=86400)
    except UniqueViolationError:
        pass  # Already processed

import hashlib
import json
from typing import Any

def generate_idempotency_key(
    *,
    entity_id: str,
    action: str,
    params: dict[str, Any] | None = None,
) -> str:
    """Generate deterministic idempotency key.

    Same input always produces the same key.
    """
    content = f"{entity_id}:{action}"
    if params:
        content += f":{json.dumps(params, sort_keys=True)}"
    return hashlib.sha256(content.encode()).hexdigest()[:32]

# Examples
key = generate_idempotency_key(
    entity_id="order-123",
    action="create",
    params={"amount": 100, "currency": "USD"},
)

Client Request (Idempotency-Key: abc-123)
     |
     v
Check cache (Redis) --> Exists? --> Return cached response
     |                                (Idempotent-Replayed: true)
     NO
     |
     v
Acquire lock --> Process request --> Cache response --> Return

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
import redis.asyncio as redis
import json

class IdempotencyMiddleware(BaseHTTPMiddleware):
    """Handle Idempotency-Key header for POST/PUT/PATCH."""

    def __init__(self, app, redis_client: redis.Redis, ttl: int = 86400):
        super().__init__(app)
        self.redis = redis_client
        self.ttl = ttl

    async def dispatch(self, request: Request, call_next):
        if request.method not in ("POST", "PUT", "PATCH"):
            return await call_next(request)

        idempotency_key = request.headers.get("Idempotency-Key")
        if not idempotency_key:
            return await call_next(request)

        cache_key = f"idem:{request.url.path}:{idempotency_key}"

        # Check for cached response
        cached = await self.redis.get(cache_key)
        if cached:
            data = json.loads(cached)
            return Response(
                content=data["body"],
                status_code=data["status"],
                media_type="application/json",
                headers={"X-Idempotent-Replayed": "true"},
            )

        # Process request
        response = await call_next(request)

        # Cache successful responses only
        if 200 <= response.status_code < 300:
            body = b"".join([chunk async for chunk in response.body_iterator])
            await self.redis.setex(
                cache_key, self.ttl,
                json.dumps({"body": body.decode(), "status": response.status_code}),
            )
            return Response(content=body, status_code=response.status_code,
                          media_type=response.media_type)

        return response

# NEVER use non-deterministic keys
key = str(uuid.uuid4())  # Different every time!

# NEVER include timestamps in keys
key = f"{event.id}:{datetime.now()}"  # Timestamp varies!

# NEVER skip idempotency for financial operations
@router.post("/payments")
async def create_payment(data):
    return await process_payment(data)  # No idempotency!

# Client generates new key on each retry
key = str(uuid.uuid4())  # New key every time!
await post("/api/orders", headers={"Idempotency-Key": key}, data=order)
# Retry creates duplicate order

# Client generates same key for same operation
key = hashlib.sha256(f"{order_id}:create:{json.dumps(order_data, sort_keys=True)}".encode()).hexdigest()
await post("/api/orders", headers={"Idempotency-Key": key}, data=order)
# Retry returns cached response

# WRONG: No owner check
await redis.delete(f"lock:{name}")  # Might release someone else's lock!

# CORRECT: Atomic owner check via Lua
RELEASE = """
if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('del', KEYS[1])
end
return 0
"""

import asyncio
from datetime import timedelta

async def long_running_task(task_id: str, redis_client):
    lock = RedisLock(redis_client, f"task:{task_id}", ttl=timedelta(seconds=30))

    async with lock:
        # Background heartbeat extends lock every 10s
        async def heartbeat():
            while lock.is_acquired:
                await lock.extend(timedelta(seconds=30))
                await asyncio.sleep(10)

        heartbeat_task = asyncio.create_task(heartbeat())
        try:
            await do_long_work()
        finally:
            heartbeat_task.cancel()

from functools import wraps

def with_lock(lock_name: str, ttl_s: int = 30, retries: int = 3):
    """Decorator to acquire lock before function execution."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, redis_client=None, **kwargs):
            for attempt in range(retries):
                lock = RedisLock(redis_client, lock_name, ttl_ms=ttl_s * 1000)
                if await lock.acquire():
                    try:
                        return await func(*args, **kwargs)
                    finally:
                        await lock.release()
                await asyncio.sleep(0.1 * (2 ** attempt))  # Backoff
            raise LockAcquisitionError(f"Failed after {retries} attempts")
        return wrapper
    return decorator

async def transfer_funds(session, from_account: int, to_account: int, amount):
    # Always lock in sorted order to prevent deadlocks
    accounts = sorted([from_account, to_account])
    for account_id in accounts:
        await session.execute(
            text("SELECT pg_advisory_xact_lock(:ns, :id)"),
            {"ns": NAMESPACE_ACCOUNT, "id": account_id},
        )
    await debit_account(session, from_account, amount)
    await credit_account(session, to_account, amount)
    await session.commit()

await redis.delete(f"lock:{resource_id}")
# If our lock expired and another process acquired it,
# we just deleted their lock!

RELEASE_SCRIPT = """
if redis.call('get', KEYS[1]) == ARGV[1] then
  return redis.call('del', KEYS[1])
end
return 0
"""
result = await redis.eval(RELEASE_SCRIPT, keys=[f"lock:{resource_id}"], args=[owner_id])
# Only deletes if owner_id matches

from contextlib import asynccontextmanager
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

class PostgresAdvisoryLock:
    """PostgreSQL advisory lock (session-level)."""

    def __init__(self, session: AsyncSession, lock_id: int):
        self._session = session
        self._lock_id = lock_id
        self._acquired = False

    async def acquire(self, blocking: bool = True) -> bool:
        if blocking:
            await self._session.execute(
                text("SELECT pg_advisory_lock(:lock_id)"),
                {"lock_id": self._lock_id},
            )
            self._acquired = True
            return True
        else:
            result = await self._session.execute(
                text("SELECT pg_try_advisory_lock(:lock_id)"),
                {"lock_id": self._lock_id},
            )
            self._acquired = result.scalar()
            return self._acquired

    async def release(self) -> bool:
        if not self._acquired:
            return False
        result = await self._session.execute(
            text("SELECT pg_advisory_unlock(:lock_id)"),
            {"lock_id": self._lock_id},
        )
        released = result.scalar()
        if released:
            self._acquired = False
        return released

    async def __aenter__(self):
        await self.acquire()
        return self

    async def __aexit__(self, *_):
        await self.release()

class PostgresTransactionLock:
    """Auto-released on commit/rollback."""

    async def acquire(self, session: AsyncSession, lock_id: int, blocking: bool = True) -> bool:
        if blocking:
            await session.execute(
                text("SELECT pg_advisory_xact_lock(:lock_id)"),
                {"lock_id": lock_id},
            )
            return True
        else:
            result = await session.execute(
                text("SELECT pg_try_advisory_xact_lock(:lock_id)"),
                {"lock_id": lock_id},
            )
            return result.scalar()

import hashlib

def string_to_lock_id(name: str) -> int:
    """Convert any string to a PostgreSQL bigint lock ID."""
    hash_bytes = hashlib.md5(name.encode()).digest()[:8]
    return int.from_bytes(hash_bytes, byteorder="big", signed=True)

# Usage
lock_id = string_to_lock_id("payment:order-123")

# Two-key lock for namespacing
NAMESPACE_PAYMENT = 1
await session.execute(
    text("SELECT pg_advisory_lock(:ns, :id)"),
    {"ns": NAMESPACE_PAYMENT, "id": 12345},
)

async def run_scheduled_job(session: AsyncSession):
    lock_id = string_to_lock_id("daily-report-job")
    lock = PostgresAdvisoryLock(session, lock_id)

    if not await lock.acquire(blocking=False):
        print("Job already running on another instance")
        return

    try:
        await generate_daily_report()
    finally:
        await lock.release()

SELECT l.pid, l.objid as lock_id, l.granted,
       a.application_name, a.client_addr,
       now() - a.state_change as duration
FROM pg_locks l
JOIN pg_stat_activity a ON l.pid = a.pid
WHERE l.locktype = 'advisory';

await session.execute(text("SELECT pg_advisory_lock(:id)"), {"id": lock_id})
await process_critical_section()
# Process crashes before unlock - lock held forever!

async with session.begin():
    await session.execute(text("SELECT pg_advisory_xact_lock(:id)"), {"id": lock_id})
    await process_critical_section()
    # Auto-released on commit/rollback/disconnect

from uuid_utils import uuid7
import redis.asyncio as redis

class RedisLock:
    """Redis lock with Lua scripts for atomicity."""

    ACQUIRE = """
    if redis.call('set', KEYS[1], ARGV[1], 'NX', 'PX', ARGV[2]) then
        return 1
    end
    return 0
    """

    RELEASE = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    end
    return 0
    """

    EXTEND = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('pexpire', KEYS[1], ARGV[2])
    end
    return 0
    """

    def __init__(self, client: redis.Redis, name: str, ttl_ms: int = 30000):
        self._client = client
        self._name = f"lock:{name}"
        self._owner = str(uuid7())
        self._ttl = ttl_ms

    async def acquire(self) -> bool:
        return await self._client.eval(
            self.ACQUIRE, 1, self._name, self._owner, self._ttl
        ) == 1

    async def release(self) -> bool:
        return await self._client.eval(
            self.RELEASE, 1, self._name, self._owner
        ) == 1

    async def extend(self, ttl_ms: int | None = None) -> bool:
        return await self._client.eval(
            self.EXTEND, 1, self._name, self._owner, ttl_ms or self._ttl
        ) == 1

    async def __aenter__(self):
        if not await self.acquire():
            raise LockError(f"Failed to acquire {self._name}")
        return self

    async def __aexit__(self, *_):
        await self.release()

1. Get current time (T1)
2. Try to acquire lock on N nodes sequentially
3. Get current time (T2)
4. Lock valid if:
   - Acquired on majority (N/2 + 1) nodes
   - Elapsed (T2-T1) < TTL - clock_drift
5. If valid: use lock with remaining TTL
6. If invalid: release on all nodes

from dataclasses import dataclass
from datetime import timedelta
import time

@dataclass
class RedlockResult:
    acquired: bool
    validity_time_ms: int = 0
    resource: str = ""
    owner_id: str = ""

class Redlock:
    """Distributed lock across multiple Redis instances."""

    def __init__(self, clients: list[redis.Redis], ttl: timedelta = timedelta(seconds=30)):
        if len(clients) < 3:
            raise ValueError("Redlock requires at least 3 Redis instances")
        self._clients = clients
        self._ttl_ms = int(ttl.total_seconds() * 1000)
        self._quorum = len(clients) // 2 + 1

    async def acquire(self, resource: str, retry_count: int = 3) -> RedlockResult:
        owner_id = str(uuid7())
        for attempt in range(retry_count):
            start = time.monotonic()
            acquired_count = sum(
                1 for c in self._clients
                if await self._try_lock(c, resource, owner_id)
            )
            elapsed_ms = int((time.monotonic() - start) * 1000)
            drift_ms = int(self._ttl_ms * 0.01) + 2
            validity = self._ttl_ms - elapsed_ms - drift_ms

            if acquired_count >= self._quorum and validity > 0:
                return RedlockResult(True, validity, resource, owner_id)

            await self._release_all(resource, owner_id)
        return RedlockResult(acquired=False, resource=resource)

await redis.set(f"lock:{name}", owner_id, ex=30)
# Two processes can both succeed!

acquired = await redis.set(f"lock:{name}", owner_id, nx=True, ex=30)
# Only one process succeeds, others get False

from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.middleware import SlowAPIMiddleware

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis://localhost:6379",
    strategy="moving-window",
)

app = FastAPI()
app.state.limiter = limiter
app.add_middleware(SlowAPIMiddleware)

# Endpoint limits
@router.post("/api/v1/auth/login")
@limiter.limit("10/minute")       # Strict for auth
async def login(request: Request): ...

@router.get("/api/v1/analyses")
@limiter.limit("100/minute")      # Normal for reads
async def list_analyses(request: Request): ...

@router.post("/api/v1/analyses")
@limiter.limit("20/minute")       # Moderate for writes
async def create_analysis(request: Request): ...

def get_user_identifier(request: Request) -> str:
    """Rate limit by user ID if authenticated, else IP."""
    if hasattr(request.state, "user"):
        return f"user:{request.state.user.id}"
    return f"ip:{get_remote_address(request)}"

limiter = Limiter(key_func=get_user_identifier)

from enum import Enum

class UserTier(Enum):
    FREE = "free"
    PRO = "pro"
    ENTERPRISE = "enterprise"

TIER_LIMITS = {
    UserTier.FREE: {"requests": 100, "window": 3600},
    UserTier.PRO: {"requests": 1000, "window": 3600},
    UserTier.ENTERPRISE: {"requests": 10000, "window": 3600},
}

async def get_rate_limit(user: User) -> str:
    limits = TIER_LIMITS[user.tier]
    return f"{limits['requests']}/{limits['window']}seconds"

@router.get("/api/v1/data")
@limiter.limit(get_rate_limit)
async def get_data(request: Request, user: User = Depends(get_current_user)):
    ...

async def add_rate_limit_headers(response: Response, limit: int,
                                  remaining: int, reset_at: datetime):
    response.headers["X-RateLimit-Limit"] = str(limit)
    response.headers["X-RateLimit-Remaining"] = str(remaining)
    response.headers["X-RateLimit-Reset"] = str(int(reset_at.timestamp()))
    response.headers["Retry-After"] = str(
        int((reset_at - datetime.now(timezone.utc)).seconds)
    )

def rate_limit_exceeded_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=429,
        content={
            "type": "https://api.example.com/errors/rate-limit-exceeded",
            "title": "Too Many Requests",
            "status": 429,
            "detail": "Rate limit exceeded. Please retry after the reset time.",
            "instance": str(request.url),
        },
        headers={"Retry-After": "60", "X-RateLimit-Remaining": "0"},
    )

# NEVER use in-memory counters in distributed systems
request_counts = {}  # Lost on restart, not shared!

# NEVER skip rate limiting on internal APIs
@router.get("/internal/admin")
async def admin_endpoint():  # No rate limit = vulnerable
    ...

# NEVER use fixed window without considering edge spikes

counters = {}  # Per-instance state
@app.post("/api/orders")
async def create_order(user_id: str):
    counters[user_id] = counters.get(user_id, 0) + 1
    if counters[user_id] > 100:
        raise RateLimitExceeded()
# Multiple instances = no shared state

@app.post("/api/orders")
@limiter.limit("100/hour")
async def create_order(request: Request, user_id: str):
    # Shared Redis state across all instances
    ...

import redis.asyncio as redis
from datetime import datetime, timezone

class SlidingWindowLimiter:
    def __init__(self, redis_client: redis.Redis, window_seconds: int = 60):
        self.redis = redis_client
        self.window = window_seconds

    async def is_allowed(self, key: str, limit: int) -> tuple[bool, int]:
        """Returns (allowed, remaining)."""
        now = datetime.now(timezone.utc).timestamp()
        window_start = now - self.window
        bucket_key = f"ratelimit:sliding:{key}"

        pipe = self.redis.pipeline()
        pipe.zremrangebyscore(bucket_key, 0, window_start)  # Remove old
        pipe.zcard(bucket_key)                               # Count current
        pipe.zadd(bucket_key, {str(now): now})               # Add this request
        pipe.expire(bucket_key, self.window * 2)             # Set expiry

        results = await pipe.execute()
        current_count = results[1]

        if current_count < limit:
            return True, limit - current_count - 1
        return False, 0

local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

-- Remove old entries
redis.call('ZREMRANGEBYSCORE', key, 0, now - window * 1000)

-- Count current entries
local count = redis.call('ZCARD', key)

if count < limit then
    redis.call('ZADD', key, now, now .. ':' .. math.random())
    redis.call('EXPIRE', key, window)
    return {1, limit - count - 1, 0}
else
    local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
    local retry_after = 0
    if oldest[2] then
        retry_after = math.ceil((tonumber(oldest[2]) + window * 1000 - now) / 1000)
    end
    return {0, 0, retry_after}
end

# Fixed window resets at minute boundaries
window_start = int(time.time() / 60) * 60
if get_count(f"{user}:{window_start}") < 100:
    allow()
# User can send 100 at 0:59 and 100 at 1:01 = 200 in 2 seconds!

now = time.time()
window_start = now - 60  # Last 60 seconds
await redis.zremrangebyscore(key, 0, window_start)  # Remove old
count = await redis.zcard(key)  # Count current
if count < 100:
    await redis.zadd(key, {str(now): now})  # Add request
# Exactly 100 requests per rolling 60-second window

Bucket: capacity=10, refill_rate=5/sec

t=0s: 10 tokens | 10 requests -> 0 tokens (burst allowed)
t=1s: +5 tokens | 5 tokens available
t=2s: +5 tokens | 10 tokens (capped at capacity)

import redis.asyncio as redis
from datetime import datetime, timezone

class TokenBucketLimiter:
    SCRIPT = """
    local key = KEYS[1]
    local capacity = tonumber(ARGV[1])
    local refill_rate = tonumber(ARGV[2])
    local tokens_requested = tonumber(ARGV[3])
    local now = tonumber(ARGV[4])

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

    -- Calculate refill
    local elapsed = (now - last_update) / 1000
    current_tokens = math.min(capacity, current_tokens + elapsed * refill_rate)

    -- Check and consume
    local allowed = 0
    local remaining = math.floor(current_tokens)
    local retry_after = 0

    if current_tokens >= tokens_requested then
        allowed = 1
        remaining = math.floor(current_tokens - tokens_requested)
        current_tokens = current_tokens - tokens_requested
    else
        local needed = tokens_requested - current_tokens
        retry_after = math.ceil(needed / refill_rate)
    end

    redis.call('HMSET', key, 'tokens', current_tokens, 'last_update', now)
    redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) * 2)
    return {allowed, remaining, retry_after}
    """

    def __init__(self, redis_client: redis.Redis,
                 capacity: int = 100, refill_rate: float = 10.0):
        self.redis = redis_client
        self.capacity = capacity
        self.refill_rate = refill_rate

    async def is_allowed(self, key: str, tokens: int = 1) -> bool:
        now = datetime.now(timezone.utc).timestamp() * 1000
        result = await self.redis.eval(
            self.SCRIPT, 1, f"ratelimit:token:{key}",
            self.capacity, self.refill_rate, tokens, now,
        )
        return result[0] == 1

const bucket = await redis.get(`bucket:${user}`);
const tokens = JSON.parse(bucket).tokens + elapsed * refillRate;
if (tokens >= 1) {
  // Race! Another request can pass this check too
  await redis.set(`bucket:${user}`, JSON.stringify({tokens: tokens - 1}));
}

const result = await redis.eval(TOKEN_BUCKET_SCRIPT,
  keys: [`bucket:${user}`],
  args: [capacity, refillRate, tokensRequested, Date.now()]
);
// Single atomic operation, no race conditions

from asyncio import Semaphore, wait_for, TimeoutError
from enum import Enum

class Tier(Enum):
    CRITICAL = 1
    STANDARD = 2
    OPTIONAL = 3

class Bulkhead:
    def __init__(self, tier: Tier, max_concurrent: int, queue_size: int, timeout: float):
        self.tier = tier
        self.semaphore = Semaphore(max_concurrent)
        self.queue_size = queue_size
        self.timeout = timeout
        self.waiting = 0
        self.active = 0

    async def execute(self, fn):
        if self.waiting >= self.queue_size:
            raise BulkheadFullError(f"Tier {self.tier.name} queue full")

        self.waiting += 1
        try:
            await wait_for(self.semaphore.acquire(), timeout=self.timeout)
            self.waiting -= 1
            self.active += 1
            try:
                return await wait_for(fn(), timeout=self.timeout)
            finally:
                self.active -= 1
                self.semaphore.release()
        except TimeoutError:
            self.waiting -= 1
            raise BulkheadTimeoutError(f"Tier {self.tier.name} timeout")

class RejectionPolicy(Enum):
    ABORT = "abort"        # Return error immediately
    CALLER_RUNS = "caller" # Execute in caller's context (blocking)
    DISCARD = "discard"    # Silently drop (for optional ops)
    QUEUE = "queue"        # Wait in bounded queue

TIER_POLICIES = {
    Tier.CRITICAL: RejectionPolicy.QUEUE,      # Wait for slot
    Tier.STANDARD: RejectionPolicy.CALLER_RUNS, # Degrade caller
    Tier.OPTIONAL: RejectionPolicy.DISCARD,     # Skip if busy
}

async def run_analysis(content: str):
    results = {}

    # Tier 1: Must succeed
    results["core"] = await tier1_bulkhead.execute(
        lambda: analyze_core(content)
    )

    # Tier 2: Best effort
    try:
        results["enriched"] = await tier2_bulkhead.execute(
            lambda: enrich_analysis(content)
        )
    except BulkheadFullError:
        results["enriched"] = None  # Skip enrichment

    # Tier 3: Optional (silent failure)
    try:
        await tier3_bulkhead.execute(lambda: warm_cache(results))
    except (BulkheadFullError, BulkheadTimeoutError):
        pass

    return results

async def fetch_data():
    await slow_external_api()  # Takes 30s when degraded
# All requests wait, entire system becomes slow

@bulkhead(tier=Tier.OPTIONAL, max_concurrent=3, timeout=5)
async def fetch_data():
    await slow_external_api()
# Only 3 concurrent + waiting queue, rest get fast rejection

              failures >= threshold
    CLOSED --------------------------------> OPEN
       ^                                      |
       |                                      |
       | probe succeeds              timeout  |
       |                              expires |
       |         +-------------+              |
       +---------+  HALF_OPEN  |<-------------+
                 +-------------+
                       |
                       | probe fails
                       v
                     OPEN

from collections import deque
from enum import Enum
from time import time

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(self, name: str, failure_threshold: int = 5,
                 recovery_timeout: float = 30.0):
        self.name = name
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._success_count = 0
        self._last_failure_time = None
        self._threshold = failure_threshold
        self._recovery_timeout = recovery_timeout

    async def call(self, fn, *args, **kwargs):
        if self._state == CircuitState.OPEN:
            if self._should_attempt_recovery():
                self._state = CircuitState.HALF_OPEN
            else:
                raise CircuitOpenError(self.name)

        try:
            result = await fn(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        if self._state == CircuitState.HALF_OPEN:
            self._success_count += 1
            if self._success_count >= 2:
                self._state = CircuitState.CLOSED
                self._failure_count = 0

    def _on_failure(self):
        self._last_failure_time = time()
        if self._state == CircuitState.HALF_OPEN:
            self._state = CircuitState.OPEN
        else:
            self._failure_count += 1
            if self._failure_count >= self._threshold:
                self._state = CircuitState.OPEN

# Retry INSIDE circuit breaker
@circuit_breaker(failure_threshold=5)
@retry(max_attempts=3, backoff=exponential)
async def call_service():
    ...

# Bulkhead + Circuit breaker
@circuit_breaker(service="analysis")
@bulkhead(tier=Tier.STANDARD, max_concurrent=3)
async def analyze():
    ...

async def call_payment_api():
    return await http.post("https://api.payment.com/charge")
    # Keeps trying even when API is down, causing 30s timeouts on every request
# Entire service becomes unresponsive

@circuit_breaker(failure_threshold=5, recovery_timeout=30)
async def call_payment_api():
    return await http.post("https://api.payment.com/charge")
# After 5 failures, circuit opens and returns fallback immediately

delay = min(base * 2^attempt, max_delay)
jitter = random(0, delay)   # Full jitter (recommended)
sleep(jitter)

RETRYABLE_ERRORS = {
    # HTTP Status Codes
    408, 429, 500, 502, 503, 504,
    # Python Exceptions
    ConnectionError, TimeoutError, ConnectionResetError,
    # LLM API Errors
    "rate_limit_exceeded", "model_overloaded", "server_error",
}

NON_RETRYABLE_ERRORS = {
    400, 401, 403, 404, 422,
    "invalid_api_key", "content_policy_violation",
    "invalid_request_error", "model_not_found",
}