Advanced Patterns

These patterns apply primarily to new implementations and significant refactors. For existing systems, propose changes through proper change management processes.

Cache-Aside Pattern (Lazy Loading)

The most common caching pattern. Application code manages the cache explicitly: read from cache first, fetch from database on miss, write result to cache.

Implementation

# CORRECT: Cache-aside with TTL and error handling
import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_product(product_id: int) -> dict:
    cache_key = f"myapp:cache:product:{product_id}"

    # Step 1: Check cache
    cached = r.get(cache_key)
    if cached is not None:
        return json.loads(cached)

    # Step 2: Cache miss - query database
    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    if product is None:
        # Cache negative result with short TTL to prevent cache penetration
        r.set(cache_key, json.dumps(None), ex=60)
        return None

    # Step 3: Store in cache with TTL
    r.set(cache_key, json.dumps(product), ex=3600)
    return product

def invalidate_product(product_id: int) -> None:
    """Invalidate cache on write (do not update cache directly)."""
    r.unlink(f"myapp:cache:product:{product_id}")

# WRONG: Cache-aside without TTL
def get_product_bad(product_id: int) -> dict:
    cached = r.get(f"myapp:cache:product:{product_id}")
    if cached:
        return json.loads(cached)
    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    r.set(f"myapp:cache:product:{product_id}", json.dumps(product))  # No TTL
    return product
    # Problem: Stale data lives forever, memory grows unbounded

# WRONG: Updating cache instead of invalidating on write
def update_product_bad(product_id: int, data: dict) -> None:
    db.execute("UPDATE products SET ... WHERE id = %s", product_id)
    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    r.set(f"myapp:cache:product:{product_id}", json.dumps(product), ex=3600)
    # Problem: Race condition between UPDATE and SELECT
    # Another process may UPDATE between our two queries, caching stale data

Cache Stampede Prevention

When a popular cache key expires, many concurrent requests may hit the database simultaneously.

# CORRECT: Lock-based stampede prevention
def get_with_lock(key: str, ttl: int, lock_ttl: int = 10) -> dict:
    """Cache-aside with lock-based stampede prevention."""
    cached = r.get(key)
    if cached is not None:
        return json.loads(cached)

    lock_key = f"myapp:lock:cache:{key}"
    acquired = r.set(lock_key, "1", nx=True, ex=lock_ttl)

    if acquired:
        try:
            value = expensive_database_query()
            r.set(key, json.dumps(value), ex=ttl)
            return value
        finally:
            r.unlink(lock_key)
    else:
        # Wait briefly and retry from cache
        time.sleep(0.1)
        cached = r.get(key)
        if cached is not None:
            return json.loads(cached)
        # Fallback: query database (lock holder may have failed)
        return expensive_database_query()

# CORRECT: Probabilistic early recomputation (PER)
def get_with_per(key: str, ttl: int, beta: float = 1.0) -> dict:
    """Cache with early recomputation before expiry."""
    cached = r.get(key)
    remaining = r.ttl(key)

    if cached is not None and remaining > 0:
        # As TTL approaches 0, probability of recomputation increases
        delta = ttl - remaining
        if delta > 0 and delta < beta * ttl * (-math.log(random.random())):
            return json.loads(cached)  # Not yet time to recompute

    value = expensive_database_query()
    r.set(key, json.dumps(value), ex=ttl)
    return value

# WRONG: No stampede protection
def get_stampede_vulnerable(key: str) -> dict:
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    # 1000 concurrent requests all arrive here at once
    result = expensive_database_query()  # DB hit 1000 times
    r.set(key, json.dumps(result), ex=3600)
    return result

Write-Through Pattern

Every write goes to both cache and database synchronously.

# CORRECT: Write-through with database-first ordering
def create_user(user_data: dict) -> dict:
    # Step 1: Write to database first (source of truth)
    user = db.execute(
        "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING *",
        user_data['name'], user_data['email']
    )

    # Step 2: Write to cache after successful DB write
    cache_key = f"myapp:user:{user['id']}:profile"
    r.hset(cache_key, mapping={
        'id': str(user['id']),
        'name': user['name'],
        'email': user['email'],
    })
    r.expire(cache_key, 3600)
    return user

# WRONG: Write to cache before database
def create_user_bad(user_data: dict) -> dict:
    r.hset(f"myapp:user:new:profile", mapping=user_data)
    user = db.execute("INSERT INTO users ...", user_data)  # May fail!
    # Problem: Cache has phantom data if DB write fails
    return user

Write-Behind (Write-Back) Pattern

Writes go to cache immediately and are asynchronously persisted to the database.

# CORRECT: Write-behind with stream-based persistence queue
def update_inventory(product_id: int, quantity: int) -> None:
    # Step 1: Update cache immediately (fast path)
    r.hset(f"myapp:inventory:{product_id}", 'quantity', quantity)
    r.expire(f"myapp:inventory:{product_id}", 86400)

    # Step 2: Queue async write to database
    r.xadd('myapp:stream:db_writes', {
        'table': 'inventory',
        'product_id': str(product_id),
        'quantity': str(quantity),
        'timestamp': str(int(time.time())),
    }, maxlen=100000)

# Background worker persists writes
def process_db_writes():
    while True:
        entries = r.xreadgroup(
            'db_writers', 'worker-1',
            {'myapp:stream:db_writes': '>'},
            count=100, block=5000
        )
        for stream, messages in entries:
            for msg_id, data in messages:
                try:
                    db.execute("UPDATE inventory SET quantity = %s WHERE product_id = %s",
                               data['quantity'], data['product_id'])
                    r.xack('myapp:stream:db_writes', 'db_writers', msg_id)
                except Exception as e:
                    log.error(f"Write-behind persistence failed: {e}")

# WRONG: Write-behind without persistence queue
def update_inventory_bad(product_id: int, quantity: int) -> None:
    r.hset(f"myapp:inventory:{product_id}", 'quantity', quantity)
    # Problem: No persistence mechanism, data lost on Redis restart

Fixed Window Rate Limiter

Counts requests in fixed time windows. Simple but allows burst at window boundaries.

-- CORRECT: Atomic fixed window rate limiter (Lua script)
-- KEYS[1] = rate limit key
-- ARGV[1] = limit, ARGV[2] = window in seconds
local current = redis.call('INCR', KEYS[1])
if current == 1 then
    redis.call('EXPIRE', KEYS[1], ARGV[2])
end
if current > tonumber(ARGV[1]) then
    return 0  -- Rate limited
end
return 1  -- Allowed

# CORRECT: Python wrapper
def check_rate_limit_fixed(client_id: str, limit: int, window: int) -> bool:
    window_num = int(time.time() // window)
    key = f"myapp:rate:{client_id}:{window_num}"
    pipe = r.pipeline()
    pipe.incr(key)
    pipe.expire(key, window + 1)
    results = pipe.execute()
    return results[0] <= limit

# WRONG: Non-atomic rate limiter
def check_rate_bad(client_id: str, limit: int) -> bool:
    key = f"myapp:rate:{client_id}"
    current = r.get(key)           # Read
    if current and int(current) >= limit:
        return False
    r.incr(key)                     # Write
    return True
    # Problem: Race condition between read and write

Sliding Window Rate Limiter

Uses a sorted set to track individual request timestamps. Accurate without boundary burst.

# CORRECT: Sliding window with sorted set
def check_rate_limit_sliding(client_id: str, limit: int, window: int) -> bool:
    key = f"myapp:rate:sliding:{client_id}"
    now = time.time()
    window_start = now - window

    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, window_start)  # Remove expired
    pipe.zcard(key)                               # Count remaining
    pipe.zadd(key, {f"{now}:{uuid.uuid4().hex[:8]}": now})  # Add current
    pipe.expire(key, window + 1)                  # Set TTL
    results = pipe.execute()

    current_count = results[1]
    if current_count >= limit:
        r.zrem(key, f"{now}:{uuid.uuid4().hex[:8]}")  # Remove the one we just added
        return False
    return True

# WRONG: Sliding window without cleanup
def check_rate_bad_sliding(client_id: str, limit: int) -> bool:
    key = f"myapp:rate:sliding:{client_id}"
    r.zadd(key, {str(time.time()): time.time()})
    count = r.zcard(key)  # Counts ALL entries including expired
    return count <= limit
    # Problem: Never removes old entries, sorted set grows forever

Token Bucket Rate Limiter

Allows controlled burst traffic up to bucket capacity while maintaining a steady average rate.

-- CORRECT: Token bucket (Lua script for atomicity)
-- KEYS[1] = bucket key
-- ARGV[1] = capacity, ARGV[2] = refill rate, ARGV[3] = now, ARGV[4] = tokens requested
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])

local 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 >= requested then
    tokens = tokens - requested
    redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
    redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) + 1)
    return 1  -- Allowed
else
    redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
    redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) + 1)
    return 0  -- Rate limited
end

Rate limiter comparison:
  Fixed window:    Simple, low memory, boundary burst possible
  Sliding window:  Accurate, higher memory (sorted set per client), no burst
  Token bucket:    Controlled burst, low memory, most flexible
  Recommendation:  Token bucket for API gateways, fixed window for per-user limits

Distributed Lock with SET NX EX

Single-Instance Lock

# CORRECT: Lock with owner verification and atomic release
class RedisLock:
    def __init__(self, redis_client, name: str, ttl: int = 30):
        self.redis = redis_client
        self.name = f"myapp:lock:{name}"
        self.ttl = ttl
        self.owner = str(uuid.uuid4())

    def acquire(self, timeout: int = 10) -> bool:
        deadline = time.time() + timeout
        while time.time() < deadline:
            if self.redis.set(self.name, self.owner, nx=True, ex=self.ttl):
                return True
            time.sleep(0.1)
        return False

    def release(self) -> bool:
        # Atomic: verify owner then delete
        script = """
        if redis.call('GET', KEYS[1]) == ARGV[1] then
            return redis.call('DEL', KEYS[1])
        end
        return 0
        """
        return self.redis.eval(script, 1, self.name, self.owner) == 1

    def extend(self, additional_ttl: int = None) -> bool:
        ttl = additional_ttl or self.ttl
        script = """
        if redis.call('GET', KEYS[1]) == ARGV[1] then
            return redis.call('EXPIRE', KEYS[1], ARGV[2])
        end
        return 0
        """
        return self.redis.eval(script, 1, self.name, self.owner, ttl) == 1

# Usage
lock = RedisLock(r, "order:5001:process")
if lock.acquire():
    try:
        process_order(5001)
    finally:
        lock.release()

# WRONG: Lock without owner verification on release
def release_bad(name: str):
    r.delete(f"lock:{name}")
    # Problem: Any process can release any lock
    # Process A acquires, times out, Process B acquires, Process A releases B's lock

# WRONG: Lock without TTL
def acquire_bad(name: str) -> bool:
    return r.setnx(f"lock:{name}", "1")
    # Problem: If process crashes, lock held forever (deadlock)

# WRONG: Non-atomic check-and-delete
def release_race(name: str, owner: str):
    if r.get(f"lock:{name}") == owner:
        r.delete(f"lock:{name}")
    # Problem: Race between GET and DELETE

Redlock Algorithm

Distributed lock across multiple independent Redis instances for higher fault tolerance.

Redlock steps:
1. Record current time
2. Try to acquire lock on N instances (minimum 3, recommended 5)
3. Lock is acquired if majority (N/2 + 1) succeed AND elapsed time < TTL
4. Effective TTL = original TTL - elapsed time - clock drift
5. If failed: release lock on all instances

Requirements:
- N independent Redis instances (not replicas)
- Clock drift factor: 1% of TTL
- Retry with random delay to prevent split-brain

Fencing Tokens

Prevent stale lock holders from performing writes after expiry.

# CORRECT: Fencing token with monotonic counter
class FencedLock:
    def __init__(self, redis_client, name: str, ttl: int = 30):
        self.redis = redis_client
        self.name = f"myapp:lock:{name}"
        self.fence_key = f"myapp:fence:{name}"
        self.ttl = ttl
        self.owner = str(uuid.uuid4())

    def acquire(self) -> int | None:
        if self.redis.set(self.name, self.owner, nx=True, ex=self.ttl):
            return self.redis.incr(self.fence_key)
        return None

# Storage validates: reject writes with fence_token <= last_seen_token

Fencing flow:
1. Client A acquires lock, gets fence_token=33
2. Client A starts processing (slow), lock expires
3. Client B acquires lock, gets fence_token=34
4. Client B writes with token=34 -> Accepted
5. Client A writes with token=33 -> Rejected (33 < 34)

Pub/Sub vs Streams

Use Pub/Sub when:
  - Fire-and-forget messaging is acceptable
  - All subscribers must receive all messages (fan-out)
  - No message persistence needed
  - Cache invalidation, config updates, real-time notifications

Use Streams when:
  - Messages must be delivered reliably (at-least-once)
  - Consumer groups for load-balanced processing
  - Message acknowledgment and retry needed
  - Consumers may disconnect and resume later
  - Event sourcing, job queues, audit logs

# CORRECT: Pub/Sub for cache invalidation (fire-and-forget)
def invalidate_cache(entity_type: str, entity_id: int):
    r.publish(f"myapp:invalidate:{entity_type}", json.dumps({'id': entity_id}))

# WRONG: Pub/Sub for critical message delivery
r.publish("myapp:critical:payment", json.dumps({'order_id': 5001}))
# Problem: If no subscriber connected, message is lost forever

# CORRECT: Streams for reliable processing
processor = OrderEventProcessor(r, "processors", "worker-1")
processor.publish({'type': 'order_created', 'order_id': '5001'})

# Consumer loop with acknowledgment
for stream, messages in processor.consume():
    for msg_id, data in messages:
        try:
            handle_event(data)
            processor.ack(msg_id)
        except Exception:
            pass  # Stays in pending list for retry/claim

Job Queue Patterns

Stream-Based Queue

# CORRECT: Job queue with streams, consumer groups, and dead letter
class StreamQueue:
    def __init__(self, redis_client, name: str, group: str, consumer: str):
        self.redis = redis_client
        self.stream = f"myapp:queue:stream:{name}"
        self.group = group
        self.consumer = consumer
        try:
            self.redis.xgroup_create(self.stream, self.group, id='0', mkstream=True)
        except redis.ResponseError as e:
            if 'BUSYGROUP' not in str(e):
                raise

    def enqueue(self, job_type: str, data: dict) -> str:
        return self.redis.xadd(self.stream, {
            'type': job_type,
            'data': json.dumps(data),
        }, maxlen=500000)

    def dequeue(self, count: int = 1, block_ms: int = 5000) -> list:
        return self.redis.xreadgroup(
            self.group, self.consumer,
            {self.stream: '>'}, count=count, block=block_ms
        ) or []

    def ack(self, msg_id: str):
        self.redis.xack(self.stream, self.group, msg_id)

    def claim_stale(self, min_idle_ms: int = 60000):
        return self.redis.xautoclaim(
            self.stream, self.group, self.consumer,
            min_idle_time=min_idle_ms, start_id='0-0', count=10
        )

Priority Queue with Sorted Set

# CORRECT: Priority queue using sorted set
class PriorityQueue:
    def __init__(self, redis_client, name: str):
        self.redis = redis_client
        self.key = f"myapp:queue:priority:{name}"

    def enqueue(self, job_id: str, priority: int = 0):
        self.redis.zadd(self.key, {job_id: priority})

    def dequeue(self) -> str | None:
        result = self.redis.zpopmin(self.key)
        return result[0][0] if result else None

Session Storage Pattern

# CORRECT: Hash-based session with sliding expiry
class SessionStore:
    def __init__(self, redis_client, ttl: int = 1800):
        self.redis = redis_client
        self.prefix = "myapp:session"
        self.ttl = ttl

    def create(self, user_id: int, metadata: dict = None) -> str:
        import secrets
        token = secrets.token_urlsafe(32)
        key = f"{self.prefix}:{token}"
        pipe = self.redis.pipeline()
        pipe.hset(key, mapping={
            'user_id': str(user_id),
            'created_at': str(int(time.time())),
            'ip': metadata.get('ip', '') if metadata else '',
        })
        pipe.expire(key, self.ttl)
        pipe.sadd(f"{self.prefix}:user:{user_id}:active", token)
        pipe.execute()
        return token

    def get(self, token: str) -> dict | None:
        key = f"{self.prefix}:{token}"
        session = self.redis.hgetall(key)
        if not session:
            return None
        # Sliding expiry: renew TTL on access
        self.redis.expire(key, self.ttl)
        return session

    def destroy(self, token: str):
        key = f"{self.prefix}:{token}"
        session = self.redis.hgetall(key)
        if session:
            pipe = self.redis.pipeline()
            pipe.unlink(key)
            if 'user_id' in session:
                pipe.srem(f"{self.prefix}:user:{session['user_id']}:active", token)
            pipe.execute()

# WRONG: Session as JSON string without sliding expiry
def create_session_bad(user_id: int) -> str:
    token = secrets.token_urlsafe(32)
    r.set(f"session:{token}", json.dumps({'user_id': user_id}))
    # Problem: No TTL, no sliding expiry, full deserialize for any read
    return token

Leaderboard Pattern

# CORRECT: Real-time leaderboard with sorted sets
class Leaderboard:
    def __init__(self, redis_client, name: str):
        self.redis = redis_client
        self.key = f"myapp:leaderboard:{name}"

    def add_score(self, player: str, score: float):
        self.redis.zadd(self.key, {player: score})

    def increment(self, player: str, delta: float) -> float:
        return self.redis.zincrby(self.key, delta, player)

    def rank(self, player: str) -> int | None:
        return self.redis.zrevrank(self.key, player)

    def top(self, count: int = 10) -> list:
        return self.redis.zrevrange(self.key, 0, count - 1, withscores=True)

    def around(self, player: str, count: int = 5) -> list:
        rank = self.redis.zrevrank(self.key, player)
        if rank is None:
            return []
        return self.redis.zrevrange(
            self.key, max(0, rank - count), rank + count, withscores=True
        )

# WRONG: String keys per player
def add_score_bad(player: str, score: float):
    r.set(f"score:{player}", score)
    # Problem: No sorting, ranking requires scanning all keys

Circuit Breaker Pattern

# CORRECT: Circuit breaker with Redis state
class CircuitBreaker:
    CLOSED, OPEN, HALF_OPEN = "closed", "open", "half_open"

    def __init__(self, redis_client, service: str,
                 failure_threshold: int = 5, recovery_timeout: int = 30):
        self.redis = redis_client
        self.prefix = f"myapp:circuit:{service}"
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout

    def can_execute(self) -> bool:
        state = self.redis.get(f"{self.prefix}:state") or self.CLOSED
        if state == self.CLOSED:
            return True
        if state == self.OPEN:
            opened_at = float(self.redis.get(f"{self.prefix}:opened_at") or 0)
            if time.time() - opened_at >= self.recovery_timeout:
                self.redis.set(f"{self.prefix}:state", self.HALF_OPEN)
                return True
            return False
        return True  # HALF_OPEN allows testing

    def record_success(self):
        state = self.redis.get(f"{self.prefix}:state") or self.CLOSED
        if state == self.HALF_OPEN:
            self.redis.set(f"{self.prefix}:state", self.CLOSED)
            self.redis.delete(f"{self.prefix}:failures")

    def record_failure(self):
        state = self.redis.get(f"{self.prefix}:state") or self.CLOSED
        if state == self.HALF_OPEN:
            self.redis.set(f"{self.prefix}:state", self.OPEN)
            self.redis.set(f"{self.prefix}:opened_at", time.time())
        elif state == self.CLOSED:
            failures = self.redis.incr(f"{self.prefix}:failures")
            self.redis.expire(f"{self.prefix}:failures", 60)
            if failures >= self.failure_threshold:
                self.redis.set(f"{self.prefix}:state", self.OPEN)
                self.redis.set(f"{self.prefix}:opened_at", time.time())

Anti-Pattern Summary

Pattern             | Anti-Pattern                        | Fix
--------------------|-------------------------------------|-----------------------------------
Cache-aside         | No TTL on cache keys                | Always SET ... EX
Cache-aside         | Update cache on write               | Invalidate (UNLINK), let read repop
Cache-aside         | No stampede protection              | Lock or PER-based repopulation
Write-through       | Cache before database               | Database first, then cache
Write-behind        | No persistence queue                | Use Streams for async DB writes
Rate limiting       | Non-atomic read + increment         | Lua script or pipeline
Rate limiting       | No TTL on rate keys                 | EXPIRE on every counter
Locks               | No owner verification on release    | Lua: compare owner then DEL
Locks               | No TTL on lock                      | Always SET NX EX
Locks               | Non-atomic release                  | Lua for atomic check + delete
Sessions            | No sliding expiry                   | EXPIRE on every access
Sessions            | JSON string instead of hash         | HSET for partial field updates
Queues              | No dead letter handling             | DLQ after max retries
Queues              | No acknowledgment                   | XACK with consumer groups
Leaderboards        | String keys per player              | Single sorted set
Pub/Sub             | Reliable delivery expectation       | Use Streams instead

Bloom Filter Pattern

Use bloom filters for probabilistic membership testing where false negatives are impossible and false positives are acceptable at a known rate.

# CORRECT: Bloom filter for duplicate detection (requires Redis Bloom module)
class BloomFilter:
    def __init__(self, redis_client, name: str):
        self.redis = redis_client
        self.key = f"myapp:bloom:{name}"

    def create(self, error_rate: float = 0.01, capacity: int = 1000000):
        try:
            self.redis.execute_command('BF.RESERVE', self.key, error_rate, capacity)
        except redis.ResponseError:
            pass  # Filter already exists

    def add(self, item: str) -> bool:
        return bool(self.redis.execute_command('BF.ADD', self.key, item))

    def exists(self, item: str) -> bool:
        return bool(self.redis.execute_command('BF.EXISTS', self.key, item))

# Usage: Prevent duplicate email sends
bloom = BloomFilter(r, "sent_emails")
bloom.create(error_rate=0.001, capacity=10000000)

def should_send_email(email_id: str) -> bool:
    if bloom.exists(email_id):
        return False  # Probably already sent
    bloom.add(email_id)
    return True

Bloom filter properties:
- False negatives: IMPOSSIBLE (if bloom says "not exists", it definitely does not)
- False positives: Possible at configured rate (e.g., 0.1% with 0.001 error rate)
- Memory: ~1.2 MB per 1M items at 1% error rate, ~1.8 MB at 0.1%
- Use cases: Email dedup, URL filtering, username availability pre-check

# WRONG: Set for massive duplicate detection
def check_sent_bad(email_id: str) -> bool:
    return r.sismember("myapp:sent_emails", email_id)
    # Problem: Set with 10M members uses 400MB+; bloom filter uses ~12MB

HyperLogLog Counting Pattern

Use HyperLogLog for approximate cardinality estimation with constant 12 KB memory per key.

# CORRECT: Unique visitor counting with daily rollup
class UniqueCounter:
    def __init__(self, redis_client, prefix: str = "myapp:visitors"):
        self.redis = redis_client
        self.prefix = prefix

    def track(self, visitor_id: str, date: str = None):
        if date is None:
            date = time.strftime('%Y-%m-%d')
        self.redis.pfadd(f"{self.prefix}:daily:{date}", visitor_id)
        self.redis.expire(f"{self.prefix}:daily:{date}", 86400 * 90)

    def daily_count(self, date: str) -> int:
        return self.redis.pfcount(f"{self.prefix}:daily:{date}")

    def weekly_count(self, start_date: str) -> int:
        from datetime import datetime, timedelta
        start = datetime.strptime(start_date, '%Y-%m-%d')
        keys = [
            f"{self.prefix}:daily:{(start + timedelta(days=i)).strftime('%Y-%m-%d')}"
            for i in range(7)
        ]
        weekly_key = f"{self.prefix}:weekly:{start_date}"
        self.redis.pfmerge(weekly_key, *keys)
        self.redis.expire(weekly_key, 86400 * 7)
        return self.redis.pfcount(weekly_key)

# Usage
counter = UniqueCounter(r)
counter.track("user:1001")
counter.track("user:1002")
counter.track("user:1001")  # Duplicate, not counted
daily = counter.daily_count("2024-01-31")  # ~2 (0.81% standard error)

# WRONG: Set for counting millions of unique items
def track_visitor_bad(visitor_id: str):
    r.sadd("myapp:all_visitors", visitor_id)
    # Problem: 10M members use 400MB+ of memory
    # HyperLogLog uses 12KB regardless of cardinality

Geospatial Query Pattern

# CORRECT: Location-based search
class GeoService:
    def __init__(self, redis_client, category: str):
        self.redis = redis_client
        self.key = f"myapp:geo:{category}"

    def add(self, name: str, lon: float, lat: float):
        self.redis.geoadd(self.key, (lon, lat, name))

    def nearby(self, lon: float, lat: float, radius: float,
               unit: str = 'km', count: int = 20) -> list:
        return self.redis.geosearch(
            self.key, longitude=lon, latitude=lat,
            radius=radius, unit=unit, sort='ASC', count=count,
            withcoord=True, withdist=True)

    def distance(self, a: str, b: str, unit: str = 'km') -> float:
        return self.redis.geodist(self.key, a, b, unit=unit)

# Usage
geo = GeoService(r, "restaurants")
geo.add("joes-pizza", -73.9857, 40.7484)
nearby = geo.nearby(-73.9830, 40.7500, 1, 'km')

-- WRONG: Storing coordinates in separate string keys
SET myapp:store:nyc:lat "40.748817"
SET myapp:store:nyc:lon "-73.985428"
-- Problem: Cannot do radius queries, requires application-level distance math

Cache Invalidation Strategies

Event-Driven Invalidation via Pub/Sub

# CORRECT: Invalidate cache across all app instances
def on_product_update(product_id: int):
    """Called after database update."""
    r.unlink(f"myapp:cache:product:{product_id}")
    r.publish("myapp:invalidate:product", json.dumps({'id': product_id}))

# Listener in each app instance
def invalidation_listener():
    pubsub = r.pubsub()
    pubsub.subscribe("myapp:invalidate:product")
    for msg in pubsub.listen():
        if msg['type'] == 'message':
            data = json.loads(msg['data'])
            local_cache.delete(f"product:{data['id']}")

Tag-Based Invalidation

# CORRECT: Tag-based invalidation for related cache entries
def cache_with_tags(key: str, value: str, tags: list, ttl: int = 3600):
    pipe = r.pipeline()
    pipe.set(key, value, ex=ttl)
    for tag in tags:
        pipe.sadd(f"myapp:tag:{tag}", key)
        pipe.expire(f"myapp:tag:{tag}", ttl + 60)
    pipe.execute()

def invalidate_by_tag(tag: str):
    keys = r.smembers(f"myapp:tag:{tag}")
    if keys:
        r.unlink(*keys)
    r.unlink(f"myapp:tag:{tag}")

# Usage
cache_with_tags("myapp:cache:product:5001", data,
                tags=["category:electronics", "brand:acme"])
invalidate_by_tag("category:electronics")  # Clears all electronics cache

Pattern Selection Guide

Need                          | Pattern                    | Key Redis Feature
------------------------------|----------------------------|----------------------
Fast reads, occasional writes | Cache-aside                | String GET/SET EX
Always-fresh cache            | Write-through              | HSET + EXPIRE
Low write latency             | Write-behind               | XADD (async queue)
API rate limiting             | Token bucket               | HMSET + Lua script
Per-user quotas               | Fixed window               | INCR + EXPIRE
Strict rate compliance        | Sliding window             | Sorted Set + ZCARD
Mutual exclusion              | SET NX EX lock             | SET NX EX + Lua release
HA mutual exclusion           | Redlock                    | Multi-instance SET NX
User sessions                 | Hash + sliding expiry      | HSET + EXPIRE on access
Real-time rankings            | Sorted Set leaderboard     | ZADD + ZREVRANGE
Reliable messaging            | Streams + consumer groups  | XREADGROUP + XACK
Fire-and-forget notifications | Pub/Sub                    | PUBLISH + SUBSCRIBE
Background jobs               | Stream queue               | XADD + XREADGROUP
Priority jobs                 | Sorted Set + List          | ZADD + ZPOPMIN
Deduplication at scale        | Bloom filter               | BF.ADD + BF.EXISTS
Unique counting               | HyperLogLog                | PFADD + PFCOUNT
Location search               | Geospatial                 | GEOADD + GEOSEARCH
Service resilience            | Circuit breaker            | Hash state machine