At 2:17 AM, my phone buzzed so violently it rattled the whole desk. Users were cursing in the support tickets — they’d update their nickname, hit refresh a few minutes later, and the old name would pop right back. My first instinct: the cache. I pulled up the monitoring dashboards. Redis memory looked fine, database connections steady as a rock. Yet users kept getting a snapshot from 12 hours ago. After six hours of log diving, I finally found the culprit: a 200ms concurrency window buried inside the “write DB then delete cache” logic of the Cache‑Aside pattern. Just long enough for a read request to sneak in, fetch the stale data from MySQL, and write it back into Redis before the delete could fire.
After the dust settled, I realized that hunting this down manually — combing through logs, eyeballing timestamps — was never going to scale. What I needed was an automated test suite that could simulate concurrency, inspect Redis state directly, and compare that state against the database like a forensic snapshot. So I built a solution using pytest + fakeredis + concurrency injection + snapshot assertions, and locked the whole class of cache‑consistency bugs inside a CI cage.
Why do cache‑consistency bugs only explode in production?
The scenario is classic. You have a User service: reads check Redis first, fall back to MySQL on a miss, and backfill the cache. Writes go to MySQL first, then delete the corresponding Redis key. It’s the textbook Cache‑Aside pattern.
Looks bulletproof on paper. But the devil lives in the concurrency. Imagine request A is an update, and B is a read:
- A writes MySQL successfully — nickname changes from
"Tom"to"Jerry". - Before A gets a chance to delete the cache, B arrives.
- B queries Redis. The key has expired or doesn’t exist yet → cache miss.
- B hits MySQL and reads "Jerry" (the new value). Lucky case? Sure, but what if the old value
"Tom"was still sitting in cache before step 1? Then B returns the stale data straight from Redis. - The truly deadly interleaving, though, is this:
- Cache has no key at the start.
- A writes DB → B reads DB (getting the old value) → A deletes cache (a no‑op) → B writes the old value back into Redis. Result: the cache is now permanently stale until the next write or TTL eviction.
Even sneakier: in many codebases, the “delete cache” step isn’t atomic, or it gets swallowed by an exception handler. You end up with a database that’s completely up‑to‑date while the cache clings to yesterday’s data. You stare at the code for half an hour and see nothing wrong, but once QPS climbs, that probabilistic bug turns into a full‑blown outage.
Conventional testing — stepping through a debugger, sprinkling logs, manually flushing caches — is blind to these race windows. You need a way to reproduce concurrent execution orders in a test and then capture the full Redis state to make assertions.
Design choices: why an in‑memory snapshot instead of a real Redis instance?
In theory, you could spin up a real Redis instance and run integration tests against it. Anyone who’s tried knows it’s a nightmare:
- Environment dependencies: Local dev and CI both need Redis installed, with matching versions — otherwise you’re debugging command differences instead of your app.
-
Data pollution: Multiple test cases share the same Redis. You must
FLUSHALLbefore every test, but clean‑up is never perfect and parallel execution causes cross‑talk. - Slowness: Network I/O drags test speed from microseconds to tens of milliseconds. When you have hundreds of cases, you want to throw your laptop out the window.
-
Time‑travel difficulties: Testing TTL or eviction logic forces you to sprinkle
time.sleepor manipulate the system clock — a time‑bomb for CI pipelines.
That’s why I chose fakeredis — a pure‑Python, in‑memory library that speaks the same redis-py commands. It lives inside the process, gets thrown away after each test, provides perfect isolation, and runs blazingly fast. More importantly, it gives us a natural entry point for an in‑memory snapshot: after the test executes, we can walk all keys, dump their values into a dict, and compare that dict against the expected database state. It’s like taking a crime‑scene photograph that preserves every piece of evidence.
Why not just mock the Redis calls? A mock verifies “did you invoke this command?”, but it can never verify whether the actual cache content is correct. Worse, every time the code adds a new SETEX, your carefully crafted assert_called_once_with lines break, sending maintenance costs through the roof. A memory snapshot doesn’t care what commands were called — it only looks at the final state. That’s the kind of testing that actually pays off.
Core implementation: from concurrency reproduction to snapshot evidence
1. First, build the caching layer with the concurrency bug built in
The code below demonstrates the problematic “bad case”. In a real project, this might sit inside your ORM or service layer; I’ve simplified it into a UserCache class:
import time
from typing import Optional
import redis # used only for type hints; tests will swap in fakeredis
class UserCache:
def __init__(self, db_conn, redis_conn):
self.db = db_conn
self.redis = redis_conn
def get_user(self, user_id: int) -> Optional[dict]:
key = f"user:{user_id}"
data = self.redis.get(key)
if data:
return eval(data) # 简化反序列化
# cache miss -> 查库
user = self.db.query_one("SELECT * FROM users WHERE id=?", (user_id,))
if user:
self.redis.set(key, str(user))
return user
def update_user(self, user_id: int, new_name: str):
# 1. 更新数据库
self.db.execute("UPDATE users SET name=? WHERE id=?", (new_name, user_id))
# 时间窗口就在这里：下面的删缓存在“稍后”执行
# 并发读可能在这中间把旧值写回 Redis
self.redis.delete(f"user:{user_id}")
That window might only be a few microseconds in reality, but it’s wide enough to get hit under load. We need to precisely recreate it in a test.
2. Set up fakeredis + the snapshot foundation with pytest fixtures
This snippet solves two things: an isolated “fake Redis” for every test, and a utility for taking those forensic snapshots:
# conftest.py
import pytest
import fakeredis
from collections.abc import Mapping
@pytest.fixture
def fake_redis():
"""每个测试独立的 fakeredis 实例，省去 FLUSHALL"""
server = fakeredis.Fa
Top comments (1)
Cache bugs are painful because the system often looks correct at each individual layer. The in-memory snapshot approach is useful because it makes the timeline visible: write, invalidate, read, refresh. Once you can test the sequence, the bug stops being a ghost and becomes a contract failure.