TTL Strategies and Cache Invalidation

Overview

You've got the 3 caching patterns down. But implementing the pattern is only half the job. The other half — and the harder one — is deciding how long a value lives in cache and how to handle the moment it's no longer valid. Phil Karlton was right: cache invalidation is one of the two hard problems in Computer Science.

This capsule gives you the tools to solve it. You'll learn the difference between a fixed TTL and a sliding TTL (which renews with every access), when to use each one, and how to choose the optimal TTL per type of data (it isn't magic: there's a method). Then you'll cover the 3 invalidation strategies: manual (DELETE on write), event-driven (Pub/Sub, a module 4 preview), and version-based (a key with a version number). Each one has tradeoffs and cases where it shines.

You close with stale-while-revalidate, an advanced technique that serves expired data immediately while regenerating it in the background — minimal latency at the cost of eventual consistency. It's the technique CDNs like Cloudflare use and one you'll be able to add to your API when performance is critical.

By the end of the capsule, you'll be able to look at any endpoint in your API and say with confidence: "we cache this data with a TTL of X minutes, we invalidate it with strategy Y, and here's why." That's what separates "I use Redis" from "I have a caching strategy."

The problem (a quick recap)

When data changes in PostgreSQL, your cache in Redis doesn't find out automatically. Serving stale data can be:

• A UX annoyance: the user doesn't see their new profile photo until the cache expires
• A minor bug: the cart shows an old price, but at checkout it's validated against the DB
• A critical bug: a bank balance shows the wrong value, authorization denies access to a user who does have permission

The strategy you choose to solve this depends on how tolerant your system can be of stale data. There's no universal solution — there's an appropriate solution for each situation.

TTL strategies: how long a value lives

Fixed TTL

The simplest one: when you store something in the cache, you define how long it's going to live. Then it deletes itself.

# A fixed 5-minute TTL
r.set("cache:product:42", json.dumps(product), ex=300)

Behavior: the key lives exactly 300 seconds from the SET. If you read it 100 times during those 300 seconds, it still expires at 300.

t=0:    SET cache:product:42 ... EX 300       ← TTL = 300
t=10:   GET cache:product:42                   ← TTL = 290 (it doesn't renew)
t=100:  GET cache:product:42                   ← TTL = 200
t=290:  GET cache:product:42                   ← TTL = 10
t=300:  GET cache:product:42                   ← the key expired, it returns nil

Pros:

• Predictable: you know exactly how long each value lives
• Simple: a single operation (SET ... EX seconds)
• It guarantees the cache doesn't survive more than N seconds without being refreshed

Cons:

• Popular data "dies" every N seconds even though it's still being used heavily
• It creates bursts of cache misses when many keys expire at the same time

Sliding TTL

The TTL renews with every access. The key stays alive as long as it keeps being used.

def get_with_sliding_ttl(key, ttl=300):
    cached = r.get(key)
    if cached:
        # Renew the TTL on every read
        r.expire(key, ttl)
        return cached
    return None

Behavior: after every access, the TTL "resets" to 300 seconds.

t=0:    SET cache:product:42 ... EX 300       ← TTL = 300
t=10:   GET + EXPIRE 300                       ← TTL = 300 (renewed)
t=100:  GET + EXPIRE 300                       ← TTL = 300 (renewed)
t=410:  (no access since t=100, the TTL expires at 100+300=400)
        GET → miss

With a sliding TTL, a popular key can live indefinitely as long as it keeps being used.

Pros:

• Popular data stays "hot" in the cache
• Excellent hit rate for frequently accessed keys

Cons:

• Data can stay stale forever if it keeps being read (what changes is the source, not the cache)
• More operations against Redis (SET + EXPIRE on every read)

When to use fixed vs sliding

The general rule:

• Sessions and active user data → sliding TTL (the classic "user stays logged in while using the app" pattern)
• Cached API responses and catalog data → fixed TTL (renew predictably)

Implementation with the set KEEPTTL flag

There's an interesting case: you want to update the value WITHOUT renewing the existing TTL.

# Without KEEPTTL: the SET resets the TTL
r.set("cache:product:42", new_value)
# Now the key has NO TTL — it lives forever (until a manual DEL)

# With a new TTL
r.set("cache:product:42", new_value, ex=300)
# TTL = 300 (it resets the old one)

# With KEEPTTL: it keeps the existing TTL
r.set("cache:product:42", new_value, keepttl=True)
# TTL = whatever was left before the SET

The use case: refreshing the cached value (because it changed in the DB) but respecting the original TTL.

Designing a TTL per type of data

Choosing a TTL isn't magic. There's a process. Ask yourself:

1. How often does this data change? (every minute, every hour, every day, rarely)
2. What's the cost of serving stale data? (a minor UX issue, a functional bug, a legal problem)
3. What's the cost of a cache miss? (latency, load on the DB)

The optimal TTL is the balance between those three factors.

The decision matrix

A real case: a product e-commerce site

# The public catalog (it barely changes)
r.set("cache:products:list", data, ex=600)  # 10 min

# An individual product with stock
r.set("cache:product:42", data, ex=300)  # 5 min (stock changes more often)

# Categories
r.set("cache:categories", data, ex=1800)  # 30 min

# Common searches
r.set(f"cache:search:{query_hash}", data, ex=120)  # 2 min

# A product's reviews
r.set("cache:reviews:42", data, ex=600)  # 10 min

# Product prices for pro users (with a discount)
r.set("cache:price:42:user:123", data, ex=900)  # 15 min

# Critical inventory (the admin dashboard)
r.set("cache:inventory:summary", data, ex=60)  # 1 min (it needs to be fresh)

Cache invalidation: the 3 strategies

A TTL alone isn't enough for cases where consistency matters more. You need active invalidation when the data changes.

Strategy 1: Manual (DELETE on write)

The simplest one. On every write operation, you also delete the cache for the affected keys.

def update_product(product_id, data):
    # 1. Update the DB
    db_update_product(product_id, data)

    # 2. Invalidate the affected caches
    r.delete(f"cache:product:{product_id}")
    r.delete("cache:products:list")  # the list changed too

    return data


def create_product(data):
    new_id = db_create_product(data)
    r.delete("cache:products:list")  # only the list (there was no detail cache)
    return new_id


def delete_product(product_id):
    db_delete_product(product_id)
    r.delete(f"cache:product:{product_id}", "cache:products:list")
    # Tip: r.delete accepts multiple keys

Pros:

• Immediate: the next read is a miss → it brings back the updated data
• Simple: a single operation

Cons:

• It takes discipline: if you forget the r.delete() in some endpoint, you get a stale data bug
• It doesn't work across microservices (each service only invalidates its own local Redis)
• When many keys have to be invalidated at the same time, there's a risk of cache stampede (covered in capsule 05)

When to use it: most cases in a monolith or a small app. If you're talking about "writes on this endpoint," this is your strategy.

Strategy 2: Event-driven (Pub/Sub)

When data changes, you publish an event. Subscribers (even in other services) listen and invalidate their related caches.

A module 4 preview — Pub/Sub is covered there. But the pattern looks like this:

# The publisher (in the endpoint doing the update)
def update_product(product_id, data):
    db_update_product(product_id, data)

    # Publish an event instead of a local DELETE
    r.publish("invalidate:product", json.dumps({
        "product_id": product_id,
        "type": "update"
    }))


# The subscriber (an async worker)
def listen_invalidations():
    pubsub = r.pubsub()
    pubsub.subscribe("invalidate:product")

    for message in pubsub.listen():
        if message["type"] == "message":
            event = json.loads(message["data"])
            r.delete(f"cache:product:{event['product_id']}")
            r.delete("cache:products:list")

Pros:

• It works across microservices: one service updates, and all the others invalidate their caches
• Decoupled: the publisher doesn't know who subscribes

Cons:

• More complex (a publisher + subscribers + Redis Pub/Sub)
• Redis Pub/Sub is fire-and-forget: if the subscriber is down, the event is lost
• Debugging is harder (asynchronous events)

When to use it: distributed systems with multiple services sharing a cache.

A critical limitation: Pub/Sub isn't durable. For invalidation in critical systems (where you can't lose an event), consider Redis Streams or a real message broker (RabbitMQ, Kafka).

Strategy 3: Version-based

The cache key includes a version number. When you bump the version, every cache key with the old version becomes irrelevant (they eventually expire without ever being used again).

# Define a global version, or one per type
def cache_key_for_product(product_id):
    version = r.get("cache:version:product") or "1"
    return f"cache:product:v{version}:{product_id}"


def get_product(product_id):
    key = cache_key_for_product(product_id)
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    product = db_get_product(product_id)
    r.set(key, json.dumps(product), ex=600)
    return product


def invalidate_all_products():
    """Bump the version: every key with the old version becomes orphaned."""
    r.incr("cache:version:product")
    # The old keys still exist with their TTL,
    # but nobody queries them anymore (every request uses the new version)
    # They eventually expire on their own with the TTL.

Pros:

• Instant mass invalidation: 1 operation invalidates millions of keys
• Zero risk of race conditions (each version is independent)
• It works perfectly across microservices (they all read the same global version)

Cons:

• The old keys take up memory until they expire with their TTL
• More complex: you need to handle the version in every cache operation

When to use it:

• Mass invalidation (e.g., "invalidate every product cache after a bulk import")
• Distributed systems where DELETE-by-pattern doesn't scale

The typical use case:

# Bulk importing a new product catalog (10,000 products)
def import_products_bulk(file_path):
    with open(file_path) as f:
        for product in csv.DictReader(f):
            db.products.insert(product)

    # Invalidate the ENTIRE products cache in one operation
    r.incr("cache:version:product")
    # Without this, you'd have to iterate over 10,000 keys with DELETE

Stale-While-Revalidate

An advanced technique that combines the best of caching and eventual consistency. The idea: when a key expires, instead of blocking the user while you regenerate the data, you serve the expired data immediately and regenerate in the background.

The flow

The normal state (a fresh cache):
Request → Hit (1ms) → Return

When the main TTL expires:
Request → Cache "expired but stale-OK" (1ms) → Return TO THE CLIENT
                                              ↓
                                        [Background: regenerate the cache]

When the cache is completely expired (past the threshold):
Request → Miss → Query DB → Cache → Return (50ms)

Implementation

import time
import json
import threading
import redis

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


def get_product_swr(product_id, fresh_ttl=300, stale_ttl=600):
    """
    Stale-while-revalidate:
    - fresh_ttl: the data is considered "fresh" (3 min)
    - stale_ttl: the data is acceptable as "stale" (6 min)
    After stale_ttl, it's a hard miss.
    """
    cache_key = f"cache:product:{product_id}"
    metadata_key = f"cache:product:{product_id}:meta"

    cached = r.get(cache_key)
    metadata = r.hgetall(metadata_key)

    if cached and metadata:
        cached_at = float(metadata.get("cached_at", 0))
        age = time.time() - cached_at

        if age < fresh_ttl:
            # Fresh: return it
            return json.loads(cached)

        elif age < stale_ttl:
            # Stale but acceptable: return it AND regenerate in the background
            threading.Thread(
                target=regenerate_in_background,
                args=(product_id, cache_key, metadata_key),
                daemon=True
            ).start()
            return json.loads(cached)

    # Hard miss: the cache is empty or too far past expiry
    return regenerate_now(product_id, cache_key, metadata_key)


def regenerate_now(product_id, cache_key, metadata_key):
    product = db_get_product(product_id)
    if product:
        pipe = r.pipeline()
        pipe.set(cache_key, json.dumps(product), ex=600)
        pipe.hset(metadata_key, mapping={"cached_at": time.time()})
        pipe.expire(metadata_key, 600)
        pipe.execute()
    return product


def regenerate_in_background(product_id, cache_key, metadata_key):
    """A worker thread: regenerate the cache silently."""
    try:
        # A lock to avoid multiple concurrent regenerations
        lock_acquired = r.set(
            f"{cache_key}:regen_lock",
            "1",
            nx=True,
            ex=10
        )
        if not lock_acquired:
            return  # another thread is already regenerating

        product = db_get_product(product_id)
        if product:
            pipe = r.pipeline()
            pipe.set(cache_key, json.dumps(product), ex=600)
            pipe.hset(metadata_key, mapping={"cached_at": time.time()})
            pipe.expire(metadata_key, 600)
            pipe.execute()
    finally:
        r.delete(f"{cache_key}:regen_lock")

When to use SWR

✅ Endpoints where latency is critical:

• An e-commerce homepage (every user loads it)
• Common search results
• A personalized feed
• Anything where 1 ms vs 100 ms is the difference between "fast" and "slow"

✅ Data that tolerates being stale (5-10 min):

• Trending posts
• Recommendation lists
• Displayed stock (the real transaction validates at checkout)

❌ Do NOT use SWR for:

• Data where stale is a bug (a balance, authorization)
• Data where the "stale window" is unacceptable (UX or legal)

Pattern-based invalidation

Sometimes you need to invalidate every key matching a pattern:

# Invalidate all of a user's caches
# user:42:profile, user:42:settings, user:42:orders, etc.

# ❌ Wrong: KEYS blocks Redis when there are many keys
keys = r.keys("user:42:*")
r.delete(*keys)

# ✅ Right: SCAN iterates without blocking
def delete_pattern(pattern):
    cursor = 0
    while True:
        cursor, keys = r.scan(cursor=cursor, match=pattern, count=100)
        if keys:
            r.delete(*keys)
        if cursor == 0:
            break

delete_pattern("user:42:*")

SCAN iterates the keyspace in batches without blocking the server. The important difference: KEYS * can freeze Redis in production with millions of keys; SCAN doesn't.

Better still: use version-based invalidation instead of pattern-based. A single INCR logically invalidates millions of keys.

When NOT to invalidate

Sometimes it's better to let the TTL do its job. The cases:

1. Data that doesn't hurt the user if it's stale

# Trending posts: if they're 5 minutes stale, who cares
r.set("cache:trending", data, ex=300)
# You don't need to invalidate when a new post is added, the TTL regenerates it

2. A cache where the invalidation frequency would be higher than the read frequency

# A like counter that changes 1000x per second
# Invalidating on every like = more expensive than serving stale for 5 seconds
r.set("cache:post:42:likes", count, ex=5)
# Just let it expire; don't invalidate

3. When invalidation carries risks

# A pattern-based delete with thousands of keys can:
# - Block Redis if you use KEYS
# - Cause a burst of cache misses → a cache stampede (capsule 05)

# Better: bump the version or wait for the natural TTL

A complete pattern: the cache hierarchy

In complex applications, you organize the cache in layers with different TTLs:

# Layer 1: Hot cache (short, frequent data)
r.set("cache:hot:product:42", data, ex=60)

# Layer 2: Warm cache (medium, common data)
r.set("cache:warm:product:42", data, ex=600)

# Layer 3: Cold cache (long, a fallback)
r.set("cache:cold:product:42", data, ex=3600)


def get_with_hierarchy(product_id):
    # Try hot
    cached = r.get(f"cache:hot:product:{product_id}")
    if cached:
        return json.loads(cached)

    # Try warm
    cached = r.get(f"cache:warm:product:{product_id}")
    if cached:
        # Populate hot
        r.set(f"cache:hot:product:{product_id}", cached, ex=60)
        return json.loads(cached)

    # Try cold
    cached = r.get(f"cache:cold:product:{product_id}")
    if cached:
        # Populate warm + hot
        r.set(f"cache:warm:product:{product_id}", cached, ex=600)
        r.set(f"cache:hot:product:{product_id}", cached, ex=60)
        return json.loads(cached)

    # Hard miss: query the DB and populate every layer
    product = db_get_product(product_id)
    pipe = r.pipeline()
    pipe.set(f"cache:hot:product:{product_id}", json.dumps(product), ex=60)
    pipe.set(f"cache:warm:product:{product_id}", json.dumps(product), ex=600)
    pipe.set(f"cache:cold:product:{product_id}", json.dumps(product), ex=3600)
    pipe.execute()
    return product

When to use it: very large APIs with clear "hot vs warm vs cold data" patterns. For 99% of cases, a single layer with a reasonable TTL is enough.

Troubleshooting

Problem 1: The hit rate drops sharply after a deploy

Cause: The deploy restarted the uvicorn workers, but the cache in Redis is still there. However, if you changed the cache format (e.g., you added a field to the JSON), the new code can't deserialize the old cache.

Solution: Version the cache keys:

# Before
r.set(f"product:{id}", json.dumps(data))

# After (with a version in the key)
CACHE_VERSION = "v2"
r.set(f"product:{CACHE_VERSION}:{id}", json.dumps(data))

When you change the format, bump CACHE_VERSION. The old keys (v1) expire on their own. This avoids deserialization errors after a deploy.

Problem 2: Redis's memory grows out of control

Cause: Very long TTLs on many keys, or invalidation failing silently.

Solution:

1. Configure maxmemory with an eviction policy:

   docker run -d --name redis-dev -p 6379:6379 \
     redis:7 redis-server \
     --maxmemory 1gb \
     --maxmemory-policy allkeys-lru

2. Audit with redis-cli --bigkeys:

   redis-cli --bigkeys
   # Output: "Biggest hash found 'user:cache:1234'"

3. Monitor memory usage per category with SCAN:

   def memory_by_pattern(pattern):
       total = 0
       cursor = 0
       while True:
           cursor, keys = r.scan(cursor=cursor, match=pattern, count=100)
           for key in keys:
               total += r.memory_usage(key) or 0
           if cursor == 0:
               break
       return total
   
   product_mem = memory_by_pattern("cache:product:*")
   user_mem = memory_by_pattern("cache:user:*")

Problem 3: Stale data in the cache because an endpoint forgot to invalidate

Cause: Inevitable in large codebases with multiple endpoints touching the same data.

Solution: Centralize the invalidation in helpers:

class ProductCache:
    PRODUCT_TTL = 600
    LIST_TTL = 300

    @classmethod
    def get_product(cls, product_id):
        ...

    @classmethod
    def invalidate_product(cls, product_id):
        r.delete(f"cache:product:{product_id}")
        r.delete("cache:products:list")  # always

    @classmethod
    def invalidate_all_products(cls):
        r.incr("cache:version:product")  # version-based


# The endpoints use it like this:
@app.put("/products/{id}")
def update_product(id: int, data: ProductUpdate):
    db_update_product(id, data)
    ProductCache.invalidate_product(id)  # ← ONE line, impossible to forget
    return data


@app.delete("/products/{id}")
def delete_product(id: int):
    db_delete_product(id)
    ProductCache.invalidate_product(id)

If the invalidation lives in a class/module, a single fix updates the behavior across every endpoint.

Problem 4: The TTL isn't renewed with SET in redis-py

Cause: You expected KEEPTTL but you didn't pass it explicitly.

# The initial set with a TTL
r.set("key", "value1", ex=300)

# An update — it RESETS the TTL
r.set("key", "value2")  # ← the TTL is now indefinite!

# An update that preserves the TTL
r.set("key", "value2", keepttl=True)  # TTL = whatever was left

keepttl=True is the flag that says "don't touch the existing TTL."

Problem 5: SWR regenerates several times concurrently

Cause: N requests arrive at the same time while the cache is stale, they all detect it as stale, and they all kick off a regeneration.

Solution: Use a lock so only one of them regenerates:

def regenerate_in_background(key):
    lock = r.set(f"{key}:regen_lock", "1", nx=True, ex=10)
    if not lock:
        return  # another thread is already regenerating

    try:
        # Regenerate
        ...
    finally:
        r.delete(f"{key}:regen_lock")

This is a preview of cache stampede protection (capsule 05).

Problem 6: Versioning breaks the TTLs

Cause: The old keys (from the previous version) keep taking up memory with their original TTL.

Solution: This is expected behavior. The old keys clean themselves up when the TTL expires. If you need to clean up immediately (because of memory pressure):

def cleanup_old_versions(prefix, current_version):
    """Clean up keys from previous versions."""
    cursor = 0
    while True:
        cursor, keys = r.scan(cursor=cursor, match=f"{prefix}:v*:*", count=100)
        for key in keys:
            # Extract the version from the key
            parts = key.split(":")
            for part in parts:
                if part.startswith("v") and part[1:].isdigit():
                    version = int(part[1:])
                    if version < current_version:
                        r.delete(key)
                    break
        if cursor == 0:
            break

But in general, let the TTL do the work — it's simpler and it doesn't block.

Exercises

Exercise 1: Comparing fixed vs sliding TTL (Easy)

Implement get_session(token) with a sliding TTL of 30 minutes. Read the session 5 times at 5-second intervals and verify the TTL stays at 30 minutes. Compare it with a get_session_fixed that uses a fixed TTL (it doesn't renew).

import time
import json
import redis

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


def create_session(token, user_id):
    r.set(f"session:{token}", json.dumps({"user_id": user_id}), ex=1800)


def get_session_sliding(token):
    cached = r.get(f"session:{token}")
    if cached:
        r.expire(f"session:{token}", 1800)  # renew the TTL
        return json.loads(cached)
    return None


def get_session_fixed(token):
    cached = r.get(f"session:{token}")
    if cached:
        return json.loads(cached)
    return None


# Test
create_session("abc123", user_id=42)

# Sliding TTL: every read renews it
for i in range(5):
    get_session_sliding("abc123")
    print(f"Sliding read {i+1}: TTL = {r.ttl('session:abc123')}s")
    time.sleep(2)

# Reset with a fixed TTL
create_session("xyz789", user_id=99)

# Fixed TTL: the TTL only counts down
for i in range(5):
    get_session_fixed("xyz789")
    print(f"Fixed read {i+1}: TTL = {r.ttl('session:xyz789')}s")
    time.sleep(2)

Sliding read 1: TTL = 1800s
Sliding read 2: TTL = 1800s
Sliding read 3: TTL = 1800s
Sliding read 4: TTL = 1800s
Sliding read 5: TTL = 1800s

Fixed read 1: TTL = 1800s
Fixed read 2: TTL = 1798s
Fixed read 3: TTL = 1796s
Fixed read 4: TTL = 1794s
Fixed read 5: TTL = 1792s

Exercise 2: Designing a TTL per type of data (Easy-Medium)

For an e-commerce app, propose a TTL for each type of data and justify it:

1. The category list
2. A product's detail
3. The stock count
4. A user's cart contents
5. A user profile
6. Search results for "laptop"
7. The homepage banner
8. A product's review count

Exercise 3: Manual invalidation with a helper (Medium)

Implement a ProductCache class with the methods: get(id), invalidate(id), invalidate_all(). Use the class from a simulated update endpoint and verify the cache is invalidated correctly.

import json
import redis

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


fake_db = {
    1: {"id": 1, "name": "Original", "price": 100},
}


class ProductCache:
    PRODUCT_TTL = 600
    LIST_TTL = 300

    @classmethod
    def get(cls, product_id):
        cached = r.get(f"cache:product:{product_id}")
        if cached:
            return ("hit", json.loads(cached))

        product = fake_db.get(product_id)
        if product:
            r.set(f"cache:product:{product_id}", json.dumps(product), ex=cls.PRODUCT_TTL)
        return ("miss", product)

    @classmethod
    def invalidate(cls, product_id):
        r.delete(f"cache:product:{product_id}")
        r.delete("cache:products:list")

    @classmethod
    def invalidate_all(cls):
        r.incr("cache:version:product")  # a version bump = mass invalidation


# A simulated endpoint
def update_product(product_id, data):
    fake_db[product_id].update(data)
    ProductCache.invalidate(product_id)
    return fake_db[product_id]


# Test
status, p = ProductCache.get(1)
print(f"Read 1: {status} - {p}")

status, p = ProductCache.get(1)
print(f"Read 2: {status} - {p}")

print("\nUpdate product:1 to price=150")
update_product(1, {"price": 150})

status, p = ProductCache.get(1)
print(f"Read 3 (after update): {status} - {p}")

status, p = ProductCache.get(1)
print(f"Read 4: {status} - {p}")

Read 1: miss - {'id': 1, 'name': 'Original', 'price': 100}
Read 2: hit - {'id': 1, 'name': 'Original', 'price': 100}

Update product:1 to price=150
Read 3 (after update): miss - {'id': 1, 'name': 'Original', 'price': 150}
Read 4: hit - {'id': 1, 'name': 'Original', 'price': 150}

Exercise 4: Version-based invalidation (Medium)

Implement a versioned cache for products. Read 3 products. Bump the version. Verify they're now cache misses. Verify with KEYS cache:product:* that the old keys are still there but nobody queries them anymore.

import json
import redis

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

fake_db = {1: {"id": 1, "name": "P1"}, 2: {"id": 2, "name": "P2"}, 3: {"id": 3, "name": "P3"}}


def get_version():
    return r.get("cache:version:product") or "1"


def cache_key(product_id):
    return f"cache:product:v{get_version()}:{product_id}"


def get_product_versioned(product_id):
    key = cache_key(product_id)
    cached = r.get(key)
    if cached:
        return ("hit", json.loads(cached))

    product = fake_db.get(product_id)
    if product:
        r.set(key, json.dumps(product), ex=600)
    return ("miss", product)


def bump_version():
    new_version = r.incr("cache:version:product")
    print(f"Version bumped to v{new_version}")
    return new_version


# Clean up so we start fresh
for k in r.keys("cache:product:*"):
    r.delete(k)
r.delete("cache:version:product")

# Fill the cache (version 1)
print("=== Version 1 ===")
for pid in [1, 2, 3]:
    status, p = get_product_versioned(pid)
    print(f"Get {pid}: {status}")

# Check the keys
print("\nCurrent keys:")
for k in sorted(r.keys("cache:product:*")):
    print(f"  {k}")

# Hits on the second read
print("\nSubsequent reads:")
for pid in [1, 2, 3]:
    status, p = get_product_versioned(pid)
    print(f"Get {pid}: {status}")

# Bump the version = mass invalidation
print("\n=== Bump version ===")
bump_version()

# The following reads are all misses
print("\nReads on version 2:")
for pid in [1, 2, 3]:
    status, p = get_product_versioned(pid)
    print(f"Get {pid}: {status}")

# Check the keys: the old and new ones coexist
print("\nCurrent keys (old and new):")
for k in sorted(r.keys("cache:product:*")):
    print(f"  {k}")

=== Version 1 ===
Get 1: miss
Get 2: miss
Get 3: miss

Current keys:
  cache:product:v1:1
  cache:product:v1:2
  cache:product:v1:3

Subsequent reads:
Get 1: hit
Get 2: hit
Get 3: hit

=== Bump version ===
Version bumped to v2

Reads on version 2:
Get 1: miss
Get 2: miss
Get 3: miss

Current keys (old and new):
  cache:product:v1:1
  cache:product:v1:2
  cache:product:v1:3
  cache:product:v2:1
  cache:product:v2:2
  cache:product:v2:3

Exercise 5: Basic stale-while-revalidate (Hard)

Implement SWR with: fresh_ttl=10s, stale_ttl=20s. Read a product. Wait 12 seconds. Read it again (it should return instantly with stale data, and regenerate in the background). Verify with r.ttl() that the cache was renewed.

import time
import json
import threading
import redis

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


db_query_count = 0


def db_get_product(pid):
    global db_query_count
    db_query_count += 1
    time.sleep(0.05)  # simulated query
    return {"id": pid, "name": "Product", "fetched_at": time.time()}


def get_swr(product_id, fresh_ttl=10, stale_ttl=20):
    cache_key = f"cache:product:{product_id}"
    cached_at_key = f"cache:product:{product_id}:cached_at"

    cached = r.get(cache_key)
    cached_at = r.get(cached_at_key)

    if cached and cached_at:
        age = time.time() - float(cached_at)
        if age < fresh_ttl:
            return ("fresh", json.loads(cached))
        elif age < stale_ttl:
            # Stale: return it AND regenerate in the background
            threading.Thread(target=regenerate, args=(product_id,), daemon=True).start()
            return ("stale", json.loads(cached))

    # Hard miss
    return ("miss", regenerate(product_id))


def regenerate(product_id):
    cache_key = f"cache:product:{product_id}"
    cached_at_key = f"cache:product:{product_id}:cached_at"

    product = db_get_product(product_id)
    pipe = r.pipeline()
    pipe.set(cache_key, json.dumps(product), ex=30)
    pipe.set(cached_at_key, str(time.time()), ex=30)
    pipe.execute()
    return product


# Test
r.delete("cache:product:1", "cache:product:1:cached_at")
db_query_count = 0

print(f"t=0:  {get_swr(1)} (DB queries: {db_query_count})")

time.sleep(5)
print(f"t=5:  {get_swr(1)} (DB queries: {db_query_count})")

time.sleep(7)  # now age = 12s, > fresh_ttl
print(f"t=12: {get_swr(1)} (DB queries: {db_query_count})")

time.sleep(0.5)  # we wait for the background regen
print(f"t=12.5: cumulative DB queries: {db_query_count}")

print(f"t=13: {get_swr(1)} (DB queries: {db_query_count})")

t=0:  ('miss', {'id': 1, 'name': 'Product', 'fetched_at': 1714069200.5}) (DB queries: 1)
t=5:  ('fresh', {'id': 1, 'name': 'Product', 'fetched_at': 1714069200.5}) (DB queries: 1)
t=12: ('stale', {'id': 1, 'name': 'Product', 'fetched_at': 1714069200.5}) (DB queries: 1)
t=12.5: cumulative DB queries: 2
t=13: ('fresh', {'id': 1, 'name': 'Product', 'fetched_at': 1714069212.6}) (DB queries: 2)

Exercise 6: Automatic cleanup (Hard)

Implement an admin endpoint /cache/cleanup that cleans out old keys from previous versions. Use SCAN so you don't block Redis. Report how many keys it deleted.

import redis

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


def cleanup_old_versions(prefix, current_version):
    """Deletes keys with a version < current_version."""
    deleted = 0
    cursor = 0

    while True:
        cursor, keys = r.scan(
            cursor=cursor,
            match=f"{prefix}:v*:*",
            count=100  # batch size
        )

        keys_to_delete = []
        for key in keys:
            # Extract version
            parts = key.split(":")
            for part in parts:
                if part.startswith("v") and part[1:].isdigit():
                    version = int(part[1:])
                    if version < current_version:
                        keys_to_delete.append(key)
                    break

        if keys_to_delete:
            deleted += r.delete(*keys_to_delete)

        if cursor == 0:
            break

    return deleted


# Setup: create keys for v1, v2, v3
for v in [1, 2, 3]:
    for pid in range(1, 11):
        r.set(f"cache:product:v{v}:{pid}", f"data-v{v}-{pid}", ex=3600)

# Verify the setup
total_keys = len(r.keys("cache:product:*"))
print(f"Total keys: {total_keys}")  # 30

# The current version is 3, we want to clean out v1 and v2
deleted = cleanup_old_versions("cache:product", current_version=3)
print(f"Keys deleted: {deleted}")  # 20 (v1 + v2)

remaining = len(r.keys("cache:product:*"))
print(f"Keys remaining: {remaining}")  # 10 (v3 only)

Total keys: 30
Keys deleted: 20
Keys remaining: 10

Summary

In this capsule you learned:

TTL strategies:

• Fixed TTL: it lives exactly N seconds from the SET. Predictable, simple
• Sliding TTL: it renews with every access. For sessions and active user data
• keepttl=True: updates the value without resetting the existing TTL
• Designing a TTL per data type: a matrix based on change frequency, stale tolerance, and the cost of a miss

Cache invalidation (3 strategies):

• Manual (DELETE on write): simple, immediate. It takes discipline. The default for monoliths
• Event-driven (Pub/Sub): decoupled, works across microservices. Pub/Sub is fire-and-forget (a module 4 preview)
• Version-based (a key with a version): instant mass invalidation with a single INCR. Ideal for distributed systems

Stale-while-revalidate:

• Serves the expired value immediately
• Regenerates in the background (with a lock to avoid concurrency)
• Minimal latency, eventual consistency
• Ideal for homepages, common searches, recommendations

When NOT to invalidate:

• Data where stale is fine (a natural TTL is enough)
• Invalidation frequency >> read frequency (counters, likes)
• Pattern-based deletes with thousands of keys (use version-based instead)

Advanced patterns:

• Pattern-based invalidation with SCAN (not KEYS)
• A cache hierarchy with multiple layers (hot/warm/cold)
• Centralizing invalidation in a helper class

Production tips:

• Version the cache keys for safe deploys
• Configure maxmemory with the allkeys-lru policy
• A nightly cleanup with SCAN for orphaned keys
• Audit with redis-cli --bigkeys to spot memory hogs

Additional resources

1. Redis: TTL and EXPIRE — The complete TTL syntax
2. Cache Invalidation Strategies — A formal comparison of the strategies
3. Stale-While-Revalidate spec (HTTP) — The concept's original RFC
4. Vercel: Stale While Revalidate — A modern implementation at the CDN edge
5. Redis EXPIRE deep dive — How expiration works internally in Redis
6. Cache Invalidation in Microservices — Patterns for distributed systems

What's next?

You close this module in Capsule 05: the cache stampede problem (what happens when a popular value expires and a thousand simultaneous requests hit PostgreSQL), and the Caching Pattern Lab mini-project where you implement the 3 patterns over the same dataset and compare hit rates with real numbers. It's module 2's final capsule.

Your workspace is intact. Let's go.