3 Years with Redis — I Never Once Verified Cache Consistency

#python #programming

At 2 a.m., an alert call yanked me out of a dream — users were complaining that they were seeing other people’s orders on their own homepage. My first instinct was an authentication failure, but the logs showed auth was fine. The real cause gave me chills: the user data in Redis hadn’t been deleted after an update, and the frontend was rendering stale cache left behind by a previous request. After fixing the bug, I stared at the screen for ten minutes, thinking: this logic has been in production for nearly three years. Why was it never caught by automated tests? Because I had never verified whether the cache and database were actually consistent.

If you also think “cache inconsistency” is some dark art that only appears in other teams’ postmortems, I urge you to spend five minutes on this article. What we’ll do is straightforward: build a reusable Redis cache consistency validation test suite with Pytest, turning “consistency as I assume it” into “consistency as proved by the machine.”

What does “consistent” even mean?

Let’s break it down. Most of us follow a similar caching pattern: on reads, hit Redis first; if it’s a miss, query the database and write back to the cache. On writes, go straight to the database and then delete the corresponding key (the Cache-Aside pattern). The pattern itself is sound, but a single oversight in the code can cause disaster:

Failed cache deletion: After updating the DB, Redis gets disconnected due to a network hiccup. The delete throws an exception that gets swallowed, and old data stays in the cache forever.
Concurrent read/write: Thread A experiences a cache miss, fetches the old value from the DB, but before it can write it back to Redis, Thread B has already completed an update + cache deletion. Thread A then leisurely writes the old value with set, and the cache is now dirty.
Cache structure mismatch: A field type changes in the DB, but the cache retains the old format due to serialization quirks. When you read and parse it, things explode.

These scenarios can’t be tested manually — you can’t coordinate two hands to trigger a cache miss exactly before a deletion. Unit tests usually mock Redis; coverage looks great, but those mocked return values bear no resemblance to real Redis network behavior, serialization, or connection pool state. They don’t validate the “cache itself.”

So we need an automated test suite that works with a real Redis (or a high-fidelity simulation), locks down these abnormal timing sequences, and runs on every push. That’s exactly what we’ll do with Pytest.

Tooling: why fakeredis, not a real instance

To test Redis behavior, we have two paths:

Real Redis: spin up a local redis-server, or connect to a testing environment instance.
fakeredis: a pure Python in-memory implementation that adapts the redis-py API, with high command coverage.

I chose fakeredis not to save resources, but for test isolation and determinism. With a real Redis and parallel test execution (e.g., pytest-xdist), you must rigorously ensure unique key names, database index isolation, and flushing after each test. Miss one step and you’ll get flaky false positives. fakeredis gives you function-scoped in-memory instances — one instance per test. It can even run alongside Python 3.8 through 3.12 in CI with zero overhead. The trade-off is that it doesn’t support a few commands (like Redis 6 ACLs or some Stream commands), but the get/set/delete trio for Cache-Aside is more than enough.

If your team insists on the “real thing,” you can create a conditional fixture that switches between the two implementations using pytest.mark.parametrize or custom markers — I’ll provide a template for that later.

Core implementation: caging the concurrent timing

Let’s first design an extremely simplified but realistic service: a UserCacheService that reads user info by user_id and caches it in Redis. We’ll fake the database with an in-memory dict, so we can freely mutate data in tests. To reproduce the failure, I’ll deliberately make the cache deletion silently fail (simulating a network timeout) without raising an exception, so our tests can catch the bug.

1. The code under test: a buggy cache service

This code is the “suspect” from our incident — after updating the DB, it calls _delete_cache, but that call can silently fail, and update_user doesn’t check the return value at all.

# user_service.py
import json
from typing import Optional, Dict
from redis import Redis

class UserCacheService:
    def __init__(self, redis_client: Redis, db: dict):
        self.redis = redis_client
        self.db = db          # 假装是数据库连接
        self.ttl = 300

    def get_user(self, user_id: str) -> Optional[Dict]:
        cache_key = f"user:{user_id}"
        cached = self.redis.get(cache_key)
        if cached:
            return json.loads(cached)

        # 缓存未命中，穿透到 DB
        data = self.db.get(user_id)
        if data:
            # 回写缓存，序列化一下
            self.redis.setex(cache_key, self.ttl, json.dumps(data))
        return data

    def update_user(self, user_id: str, data: Dict) -> None:
        # 先更新 DB
        self.db[user_id] = data
        # 然后删缓存——注意这里没有做异常处理！
        try:
            self._delete_cache(user_id)
        except Exception:
            # 这里是 bug 的根源：吞掉异常，认为删缓存成功
            pass

    def _delete_cache(self, user_id: str) -> None:
        # 模拟可能失败的操作，真实场景可能是网络超时
        self.redis.delete(f"user:{user_id}")

2. Pytest fixture: a clean world for every test

We’ll build fixtures with fakeredis to make Redis and the DB function-scoped, ensuring each test starts from scratch without cross-contamination.

# conftest.py (或直接写在测试文件里)
import pytest
from fakeredis import FakeRedis
from u

DEV Community