Cache Invalidation & Consistency — Vấn Đề Khó Nhất Trong Distributed Systems

t=0s:  Key expires
t=0.0: Request #1 → miss → start DB query (sẽ mất 2s)
t=0.1: Request #2 → miss → start DB query
t=0.3: Request #3 → miss → start DB query
...
t=1.9: Request #200 → miss → start DB query
t=2.0: Request #1 xong → set cache
t=2.1: Request #2 xong → set cache (ghi đè)
...       (200 DB queries chạy song song)
t=4.0: Request #200 xong → set cache (ghi đè lần 200)

import random

BASE_TTL = 3600  # 1 giờ

def set_with_jitter(key: str, value, base_ttl: int = BASE_TTL, jitter_pct: float = 0.1):
    """
    Thêm ±10% random jitter vào TTL.
    1000 items expire trong khoảng 3240-3960s thay vì cùng lúc 3600s.
    """
    jitter = int(base_ttl * jitter_pct)
    ttl = base_ttl + random.randint(-jitter, jitter)
    r.setex(key, ttl, value)

# Naive mutex — có bug
acquired = r.set(lock_key, "1", nx=True, ex=5)
if acquired:
    data = fetch_from_db()  # <- nếu crash ở đây
    r.setex(cache_key, ttl, data)
    r.delete(lock_key)  # <- lock không được release
# Lock tự expire sau 5s (OK), nhưng 5s đó mọi request làm gì?
# Option A: chờ → latency spike
# Option B: đọc thẳng DB → không scale

acquired = r.set(lock_key, "1", nx=True, ex=5)  # 5s lock
if acquired:
    data = fetch_from_db()  # mất 7s vì DB slow
    # Lúc này lock đã expire sau 5s
    # Request khác đã acquire lock và đang fetch
    # Bây giờ có 2 process cùng populate cache
    r.setex(cache_key, ttl, data)  # race condition
    r.delete(lock_key)  # xóa lock của người khác!

import uuid

def get_with_safe_mutex(key: str, fetch_fn, ttl: int, lock_timeout: int = 10) -> any:
    lock_key = f"lock:{key}"
    lock_owner = str(uuid.uuid4())  # Unique ID cho lock này

    cached = r.get(key)
    if cached:
        return json.loads(cached)

    # Lưu owner ID vào lock value
    acquired = r.set(lock_key, lock_owner, nx=True, ex=lock_timeout)

    if acquired:
        try:
            value = fetch_fn()
            r.setex(key, ttl, json.dumps(value))
            return value
        finally:
            # Chỉ xóa lock nếu vẫn là owner
            # Dùng Lua để atomic check-and-delete
            release_lock(lock_key, lock_owner)
    else:
        # Exponential backoff thay vì fixed sleep
        for attempt in range(5):
            time.sleep(0.05 * (2 ** attempt))  # 50ms, 100ms, 200ms...
            cached = r.get(key)
            if cached:
                return json.loads(cached)
        # Fallback sau 5 lần retry
        return fetch_fn()

# Lua script: atomic check owner + delete
RELEASE_LOCK_SCRIPT = """
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end
"""
release_lock_sha = r.script_load(RELEASE_LOCK_SCRIPT)

def release_lock(lock_key: str, owner_id: str):
    r.evalsha(release_lock_sha, 1, lock_key, owner_id)

cache:article:123                    → full article object
cache:feed:user:456                  → feed của user 456 (chứa article 123)
cache:feed:user:789                  → feed của user 789 (cũng chứa article 123)
cache:category:technology:page:1     → listing page (có article 123)
cache:search:results:"python"        → search results (có article 123)
cache:author:johndoe:articles        → bài của author johndoe
cache:homepage:featured              → featured section
cache:sitemap                        → sitemap XML
... (và vài chục key khác)

def update_article(article_id: str, data: dict):
    db.update_article(article_id, data)

    # Manually list tất cả affected keys — nightmare to maintain
    keys_to_delete = [
        f"cache:article:{article_id}",
        f"cache:author:{data['author_id']}:articles",
        f"cache:category:{data['category']}:page:1",
        # Còn user feeds? Còn search? Còn homepage?
        # Bạn chắc chắn sẽ quên một vài cái
    ]
    r.delete(*keys_to_delete)

article:123 → tags: ["article:123", "author:johndoe", "category:tech"]
user:456 feed → tags: ["user:456", "article:123", "article:789"]
homepage → tags: ["featured", "article:123", "article:456"]

class TaggedCache:
    def __init__(self, redis_client):
        self.r = redis_client

    def set(self, key: str, value, ttl: int, tags: list[str]):
        """Cache entry với tag metadata."""
        pipe = self.r.pipeline()

        # Lưu actual value
        pipe.setex(key, ttl, json.dumps(value))

        # Với mỗi tag, thêm key vào set của tag đó
        for tag in tags:
            tag_set_key = f"tag:{tag}"
            pipe.sadd(tag_set_key, key)
            # Tag set cũng cần expire, lâu hơn content một chút
            pipe.expire(tag_set_key, ttl + 300)

        pipe.execute()

    def get(self, key: str):
        data = self.r.get(key)
        return json.loads(data) if data else None

    def invalidate_tag(self, tag: str):
        """Xóa tất cả keys có tag này."""
        tag_set_key = f"tag:{tag}"

        # Lấy tất cả keys thuộc tag này
        keys = self.r.smembers(tag_set_key)

        if keys:
            pipe = self.r.pipeline()
            # Xóa tất cả content keys
            for key in keys:
                pipe.delete(key)
            # Xóa tag set
            pipe.delete(tag_set_key)
            pipe.execute()

        return len(keys)

    def invalidate_tags(self, tags: list[str]):
        """Invalidate nhiều tags cùng lúc."""
        all_keys = set()
        tag_set_keys = []

        pipe = self.r.pipeline()
        for tag in tags:
            pipe.smembers(f"tag:{tag}")
            tag_set_keys.append(f"tag:{tag}")
        results = pipe.execute()

        for keys in results:
            all_keys.update(keys)

        if all_keys or tag_set_keys:
            pipe = self.r.pipeline()
            for key in all_keys:
                pipe.delete(key)
            for tag_key in tag_set_keys:
                pipe.delete(tag_key)
            pipe.execute()

cache = TaggedCache(r)

# Cache article với đầy đủ tags
def get_article(article_id: str) -> dict:
    key = f"article:{article_id}"
    cached = cache.get(key)
    if cached:
        return cached

    article = db.get_article(article_id)
    cache.set(
        key=key,
        value=article,
        ttl=3600,
        tags=[
            f"article:{article_id}",
            f"author:{article['author_id']}",
            f"category:{article['category']}",
        ]
    )
    return article

# Cache user feed
def get_user_feed(user_id: str) -> list:
    key = f"feed:{user_id}"
    cached = cache.get(key)
    if cached:
        return cached

    feed = db.get_feed(user_id)
    # Tag với tất cả article IDs trong feed
    article_tags = [f"article:{a['id']}" for a in feed]

    cache.set(
        key=key,
        value=feed,
        ttl=300,
        tags=[f"user:{user_id}"] + article_tags
    )
    return feed

# Khi article update → invalidate một tag, tất cả bị xóa
def update_article(article_id: str, data: dict):
    db.update_article(article_id, data)
    count = cache.invalidate_tag(f"article:{article_id}")
    print(f"Invalidated {count} cache entries")
    # Không cần biết 12 cái key kia là gì — tag lo hết

# Đảm bảo tag set luôn sống lâu hơn content
tag_ttl = max(ttl * 2, ttl + 600)  # ít nhất 10 phút dư
pipe.expire(tag_set_key, tag_ttl)

# Không tag feed của từng user vào article
# Thay vào đó: invalidate theo category, feed service tự refresh

def update_article(article_id: str, data: dict):
    db.update_article(article_id, data)

    # Chỉ invalidate article-level cache
    cache.invalidate_tag(f"article:{article_id}")

    # Publish event để feed service tự xử lý
    pubsub.publish("article.updated", {
        "article_id": article_id,
        "category": data["category"],
        "author_id": data["author_id"]
    })

def invalidate_large_tag(tag: str, batch_size: int = 500):
    """Invalidate tag lớn theo batch để không block Redis."""
    tag_set_key = f"tag:{tag}"

    while True:
        # SPOP lấy và xóa batch_size members
        keys = r.spop(tag_set_key, batch_size)
        if not keys:
            break

        pipe = r.pipeline()
        for key in keys:
            pipe.delete(key)
        pipe.execute()

        # Nhường CPU giữa các batch
        time.sleep(0.01)

Bao lâu chấp nhận stale data?
├── 0ms  → Strong consistency  (đắt, slow, thường không cần)
├── <1s  → Synchronous replication (latency tăng)
├── <5s  → Async replication với notification
└── Vài phút → TTL-based eventual consistency (đơn giản nhất)

User (Paris) → Write → Redis Paris → Async → Redis Singapore
User (SG)    → Read  → Redis Singapore (có thể stale < Xms)

import time

def write_with_version(key: str, value, region: str):
    """Write với logical timestamp để LWW chính xác hơn."""
    versioned = {
        "value": value,
        "timestamp": time.time_ns(),  # nanoseconds
        "region": region,
        "version": generate_version()  # monotonic per-key counter
    }
    r.set(key, json.dumps(versioned))
    replicate_async(key, versioned)  # push sang regions khác

def merge_on_conflict(existing: dict, incoming: dict) -> dict:
    """Khi receive replication, check conflict."""
    if incoming["timestamp"] > existing["timestamp"]:
        return incoming  # LWW: newer wins
    return existing  # Existing wins, drop incoming

def receive_replication(key: str, incoming: dict):
    """Handler khi nhận replication từ region khác."""
    existing_raw = r.get(key)
    if not existing_raw:
        r.set(key, json.dumps(incoming))
        return

    existing = json.loads(existing_raw)
    winner = merge_on_conflict(existing, incoming)
    r.set(key, json.dumps(winner))

# Counter CRDT: mỗi region track increment của mình
def increment_counter(counter_name: str, region: str, amount: int = 1):
    key = f"counter:{counter_name}:region:{region}"
    r.incrby(key, amount)

def get_global_counter(counter_name: str, regions: list[str]) -> int:
    """Tổng hợp counter từ tất cả regions."""
    total = 0
    for region in regions:
        key = f"counter:{counter_name}:region:{region}"
        val = r.get(key)
        total += int(val or 0)
    return total

# View count không bao giờ conflict — chỉ cần sum
# Like count cũng vậy
# Nhưng "current price" thì không dùng CRDT được

Write path:  User (Paris) → API (Paris) → Primary Redis (US-East) → Async → Redis Paris
Read path:   User (Paris) → API (Paris) → Redis Paris (có thể stale)

import os

REGION = os.getenv("REGION", "us-east")  # current region
PRIMARY_REGION = "us-east"

class MultiRegionCache:
    def __init__(self):
        self.local = redis.Redis(host=f"redis.{REGION}.internal")
        self.primary = redis.Redis(host=f"redis.{PRIMARY_REGION}.internal")

    def get(self, key: str):
        # Luôn đọc local trước
        data = self.local.get(key)
        if data:
            return json.loads(data)

        # Local miss → thử primary (cross-region, có latency)
        data = self.primary.get(key)
        if data:
            # Backfill local cache
            value = json.loads(data)
            self.local.setex(key, 300, data)  # 5 phút local TTL
            return value

        return None

    def set(self, key: str, value, ttl: int):
        serialized = json.dumps(value)

        if REGION == PRIMARY_REGION:
            # Đang ở primary: write local, replication tự lo
            self.primary.setex(key, ttl, serialized)
        else:
            # Đang ở secondary: write về primary
            # Primary tự replicate về secondary sau
            self.primary.setex(key, ttl, serialized)
            # Optionally: update local ngay để read-your-writes
            self.local.setex(key, min(ttl, 60), serialized)

    def invalidate(self, key: str):
        """Invalidate trên tất cả regions."""
        # Publish invalidation event thay vì cross-region DELETE
        self.primary.publish("cache:invalidate", key)
        # Mỗi region subscribe và xóa local copy

# Subscriber chạy ở mỗi region
def start_invalidation_subscriber(local_redis, primary_redis):
    pubsub = primary_redis.pubsub()
    pubsub.subscribe("cache:invalidate")

    for message in pubsub.listen():
        if message["type"] == "message":
            key = message["data"].decode()
            local_redis.delete(key)
            print(f"[{REGION}] Invalidated: {key}")

Cache-Control: max-age=60, stale-while-revalidate=300

import threading
import time
import json
from dataclasses import dataclass, asdict
from typing import Callable, TypeVar, Optional

T = TypeVar('T')

@dataclass
class SWREntry:
    value: any
    fresh_until: float     # Serve fresh
    stale_until: float     # Serve stale + background refresh
    # Sau stale_until: block và fetch sync

    @property
    def is_fresh(self) -> bool:
        return time.time() < self.fresh_until

    @property
    def is_stale(self) -> bool:
        return self.fresh_until <= time.time() < self.stale_until

    @property
    def is_expired(self) -> bool:
        return time.time() >= self.stale_until


class SWRCache:
    def __init__(self, redis_client):
        self.r = redis_client
        self._refresh_in_progress = set()  # Track ongoing refreshes
        self._lock = threading.Lock()

    def get_or_compute(
        self,
        key: str,
        fetch_fn: Callable[[], T],
        fresh_ttl: int,          # Bao lâu thì fresh
        stale_ttl: int,          # Thêm bao lâu nữa vẫn serve stale
        error_ttl: int = 30,     # Cache error bao lâu (tránh error stampede)
    ) -> T:
        raw = self.r.get(key)

        if raw:
            entry = SWREntry(**json.loads(raw))

            if entry.is_fresh:
                return entry.value  # Happy path

            if entry.is_stale:
                # Serve stale ngay, trigger background refresh
                self._trigger_background_refresh(key, fetch_fn, fresh_ttl, stale_ttl)
                return entry.value

            # is_expired: phải fetch sync, không còn gì để serve
            return self._fetch_and_cache(key, fetch_fn, fresh_ttl, stale_ttl)

        # Total miss
        return self._fetch_and_cache(key, fetch_fn, fresh_ttl, stale_ttl)

    def _trigger_background_refresh(self, key, fetch_fn, fresh_ttl, stale_ttl):
        """Chỉ trigger một background refresh mỗi lúc cho mỗi key."""
        with self._lock:
            if key in self._refresh_in_progress:
                return  # Đã có refresh đang chạy, không cần trigger thêm
            self._refresh_in_progress.add(key)

        def do_refresh():
            try:
                self._fetch_and_cache(key, fetch_fn, fresh_ttl, stale_ttl)
            except Exception as e:
                print(f"Background refresh failed for {key}: {e}")
                # Không crash, không notify user — họ đã nhận stale data rồi
            finally:
                with self._lock:
                    self._refresh_in_progress.discard(key)

        threading.Thread(target=do_refresh, daemon=True).start()

    def _fetch_and_cache(self, key, fetch_fn, fresh_ttl, stale_ttl) -> any:
        now = time.time()
        try:
            value = fetch_fn()
            entry = SWREntry(
                value=value,
                fresh_until=now + fresh_ttl,
                stale_until=now + fresh_ttl + stale_ttl,
            )
            total_ttl = fresh_ttl + stale_ttl
            self.r.setex(key, total_ttl, json.dumps(asdict(entry)))
            return value

        except Exception as e:
            # Fetch thất bại: cache error ngắn để tránh hammer DB
            error_entry = SWREntry(
                value=None,
                fresh_until=now + 30,      # Error "fresh" 30s
                stale_until=now + 30,      # Không có stale window cho error
            )
            self.r.setex(key, 30, json.dumps(asdict(error_entry)))
            raise  # Re-raise để caller biết

swr = SWRCache(r)

def get_product_detail(product_id: str) -> dict:
    return swr.get_or_compute(
        key=f"product:{product_id}",
        fetch_fn=lambda: db.get_product(product_id),
        fresh_ttl=60,          # Fresh 1 phút
        stale_ttl=300,         # Serve stale thêm 5 phút nếu cần
    )

# User nhận response ngay, kể cả khi stale
# DB chỉ bị hit bởi background thread, không phải user-facing request

CDN layer:   max-age=60, stale-while-revalidate=300
App layer:   fresh_ttl=30, stale_ttl=120

Vấn đề: CDN có thể serve stale 5 phút.
         App có thể serve stale 2 phút.
         Total staleness: lên đến 7 phút.

Nếu data thực sự thay đổi, user có thể thấy stale đến 7 phút.
Có acceptable không? Depends on use case.

def update_product_and_bust(product_id: str, data: dict):
    db.update_product(product_id, data)

    # 1. Invalidate application cache
    r.delete(f"product:{product_id}")

    # 2. Purge CDN — nếu CDN support API
    cdn_client.purge(f"/products/{product_id}")
    cdn_client.purge(f"/api/v1/products/{product_id}")

    # 3. Hoặc dùng versioned URLs thay vì purge
    new_version = increment_version(product_id)
    # Frontend tự đọc version mới từ /api/product-meta/{id}

Write → [Cache] → [DB] (synchronous)

def write_through(key: str, value, ttl: int):
    # Ghi cache và DB trong cùng một transaction
    db.save(key, value)          # DB first
    r.setex(key, ttl, json.dumps(value))  # Then cache

Write → [Cache] → async → [DB]

def write_behind(key: str, value, ttl: int):
    r.setex(key, ttl, json.dumps(value))  # Cache ngay
    dirty_queue.push({"key": key, "value": value})  # Async DB write

# Background worker
def flush_dirty_queue():
    while True:
        items = dirty_queue.pop_batch(100)
        for item in items:
            db.save(item["key"], item["value"])
        time.sleep(0.1)

Write → [DB] only
Read:  miss → [DB] → populate cache

def cache_aside_write(key: str, value):
    db.save(key, value)
    r.delete(key)  # Invalidate cache, không update

def cache_aside_read(key: str):
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    value = db.get(key)
    r.setex(key, 3600, json.dumps(value))
    return value

Read → [Cache] → [DB] (cache tự động populate, không phải app)

Thread A: read → miss → query DB (chưa xong)
Thread B: write new value vào DB → DELETE cache key
Thread A: xong → SET cache với OLD value
Kết quả: Cache có giá trị cũ, DB có giá trị mới. Inconsistent!

# Safer cache-aside write
def safe_write(key: str, new_value):
    # Mark key as being updated với TTL ngắn
    r.set(f"updating:{key}", "1", ex=5)

    try:
        db.save(key, new_value)
        r.delete(key)  # Invalidate
    finally:
        r.delete(f"updating:{key}")

def safe_read(key: str):
    # Nếu key đang được update, skip cache
    if r.exists(f"updating:{key}"):
        return db.get(key)  # Read trực tiếp DB

    cached = r.get(key)
    if cached:
        return json.loads(cached)

    value = db.get(key)
    # Chỉ cache nếu không đang update
    if not r.exists(f"updating:{key}"):
        r.setex(key, 3600, json.dumps(value))
    return value

def versioned_write(entity_id: str, new_value: dict):
    # Increment version trong DB atomically
    new_version = db.update_with_version(entity_id, new_value)

    # Cache với version
    cache_value = {**new_value, "_version": new_version}
    r.setex(f"entity:{entity_id}", 3600, json.dumps(cache_value))

def versioned_read(entity_id: str) -> dict:
    cached = r.get(f"entity:{entity_id}")
    if cached:
        data = json.loads(cached)
        # Verify version khớp với DB (optional, dùng cho critical paths)
        db_version = db.get_version(entity_id)
        if data["_version"] == db_version:
            return data
        # Version mismatch → invalidate và re-read
        r.delete(f"entity:{entity_id}")

    value = db.get_entity(entity_id)
    r.setex(f"entity:{entity_id}", 3600, json.dumps(value))
    return value

# Bug: đọc rồi mới set trong pipeline tách biệt
def buggy_increment_and_cache(user_id: str):
    current = r.get(f"user:{user_id}:points")
    new_points = (int(current) if current else 0) + 10

    # Giữa đây và setex, thread khác có thể đã set value khác
    time.sleep(0.001)  # Simulate processing delay

    r.set(f"user:{user_id}:points", new_points)
    # Lost update — value của thread khác bị ghi đè

# Fix: INCRBY atomic
def correct_increment(user_id: str, amount: int = 10):
    return r.incrby(f"user:{user_id}:points", amount)

# Bug: classic TOCTOU (Time-of-check to time-of-use)
def buggy_deduct_inventory(product_id: str, quantity: int) -> bool:
    stock = int(r.get(f"stock:{product_id}") or 0)
    if stock >= quantity:
        # Giữa check và update, người khác có thể đã mua hết
        r.decrby(f"stock:{product_id}", quantity)
        return True
    return False

# Fix: Lua script để atomic check-and-decrement
DEDUCT_SCRIPT = """
local stock = tonumber(redis.call('GET', KEYS[1]) or '0')
if stock >= tonumber(ARGV[1]) then
    redis.call('DECRBY', KEYS[1], ARGV[1])
    return 1
else
    return 0
end
"""
deduct_sha = r.script_load(DEDUCT_SCRIPT)

def safe_deduct_inventory(product_id: str, quantity: int) -> bool:
    result = r.evalsha(deduct_sha, 1, f"stock:{product_id}", quantity)
    return bool(result)

# Scenario: Invalidate A triggers invalidate B triggers invalidate C
# Nếu có cycle, infinite loop

# Ví dụ thực: article update → invalidate feed → feed rebuild query article → miss → query DB
# Nếu article đang update, DB query trả về version cũ → feed cache stale ngay sau khi vừa build

# Fix: Event-driven với idempotency key
def handle_article_update(article_id: str, event_id: str):
    # Idempotency check: đã xử lý event này chưa?
    if r.get(f"processed_event:{event_id}"):
        return  # Skip duplicate events

    r.setex(f"processed_event:{event_id}", 3600, "1")

    # Invalidate với timestamp để downstream biết "fresh after X"
    timestamp = time.time()
    r.set(f"article_updated:{article_id}", timestamp)

    # Downstream cache check timestamp trước khi serve
    cache.invalidate_tag(f"article:{article_id}")

# RedLock algorithm (simplified) — dùng khi single Redis không đủ tin cậy
# Yêu cầu: 2N+1 Redis nodes, acquire lock trên majority (N+1)

class RedLock:
    def __init__(self, redis_nodes: list):
        self.nodes = redis_nodes
        self.quorum = len(redis_nodes) // 2 + 1

    def acquire(self, resource: str, ttl_ms: int) -> tuple[bool, str]:
        token = str(uuid.uuid4())
        start = time.time_ns() // 1_000_000  # ms

        acquired_count = 0
        for node in self.nodes:
            try:
                if node.set(resource, token, nx=True, px=ttl_ms):
                    acquired_count += 1
            except Exception:
                pass  # Node down → skip

        elapsed = time.time_ns() // 1_000_000 - start
        validity = ttl_ms - elapsed

        if acquired_count >= self.quorum and validity > 0:
            return True, token

        # Không đạt quorum → release tất cả
        self.release(resource, token)
        return False, ""

    def release(self, resource: str, token: str):
        for node in self.nodes:
            try:
                node.evalsha(release_lock_sha, 1, resource, token)
            except Exception:
                pass

Nguồn thay đổi
       │
       ├─ Tôi cần tất cả reader biết ngay?
       │       Có → Strong consistency → Write-through + sync invalidation
       │       Không → Eventual OK → TTL + async invalidation
       │
       ├─ Có bao nhiêu cache entry bị ảnh hưởng?
       │       Ít (1-10) → Delete từng key
       │       Nhiều (10-1000) → Tag-based invalidation
       │       Rất nhiều (>1000) → Event-driven, downstream tự xử lý
       │
       ├─ Data đắt hay rẻ để recompute?
       │       Rẻ (<50ms) → Lazy invalidation, để miss tự resolve
       │       Đắt (>500ms) → Background refresh, SWR, mutex
       │
       └─ Single region hay multi-region?
               Single → Redis pub/sub invalidation đủ
               Multi → Replication lag là fact of life, design around it