Redis

Distributed Cache Consistency Testing: Strategies and Patterns

15 Sep 2026 — 7 min read

Distributed caches introduce consistency problems that single-node caches don't have. When your cache is spread across multiple Redis nodes—or when multiple application instances share a cache—the same key might return different values depending on which node serves the request. Testing these properties requires deliberate setup.

The Consistency Models You're Actually Testing

Before writing tests, understand what consistency guarantees your system claims:

Read-your-writes: After a process writes a value, that same process always reads the new value. Seems obvious, but fails when writes go to primary and reads might hit replicas.

Monotonic reads: If you read value V, you never subsequently read an older value V'. Fails when load balancers route reads across replicas at different replication offsets.

Eventual consistency: All replicas will eventually converge to the same value. The "eventually" has a measurable bound—your tests should verify it.

Linearizability: Operations appear to happen at a single point in time in their real-time order. This is the strongest guarantee and usually comes at a performance cost.

Testing Read-Your-Writes

# replicated_cache.py
import redis
import json
import time

class ReplicatedCache:
    """Cache with a primary for writes and replicas for reads."""
    
    def __init__(self, primary_url: str, replica_urls: list[str]):
        self.primary = redis.Redis.from_url(primary_url, decode_responses=True)
        self.replicas = [redis.Redis.from_url(url, decode_responses=True) 
                        for url in replica_urls]
        self._replica_idx = 0
    
    def _get_replica(self):
        """Round-robin replica selection."""
        r = self.replicas[self._replica_idx % len(self.replicas)]
        self._replica_idx += 1
        return r
    
    def set(self, key: str, value: dict, ttl: int = 300):
        self.primary.setex(key, ttl, json.dumps(value))
    
    def get(self, key: str) -> dict | None:
        # Read from replica
        cached = self._get_replica().get(key)
        if cached:
            return json.loads(cached)
        return None
    
    def get_with_read_your_writes(self, key: str, session_token: str) -> dict | None:
        """
        Implements read-your-writes by reading from primary
        if this session recently wrote.
        """
        write_flag = f'ryw:{session_token}:{key}'
        
        if self.primary.exists(write_flag):
            # This session wrote recently — read from primary
            cached = self.primary.get(key)
        else:
            cached = self._get_replica().get(key)
        
        return json.loads(cached) if cached else None
    
    def set_with_ryw_flag(self, key: str, value: dict, session_token: str, ttl: int = 300):
        """Set value and mark that this session should read-your-writes."""
        write_flag = f'ryw:{session_token}:{key}'
        self.primary.setex(key, ttl, json.dumps(value))
        self.primary.setex(write_flag, 5, '1')  # 5s flag — covers replication lag

# test_read_your_writes.py
import pytest
from unittest.mock import MagicMock, patch
import json
from replicated_cache import ReplicatedCache

@pytest.fixture
def mock_cache():
    """Creates a cache with mock Redis instances for isolation."""
    primary = MagicMock()
    replica1 = MagicMock()
    replica2 = MagicMock()
    
    cache = ReplicatedCache.__new__(ReplicatedCache)
    cache.primary = primary
    cache.replicas = [replica1, replica2]
    cache._replica_idx = 0
    
    return cache, primary, replica1, replica2

def test_write_goes_to_primary(mock_cache):
    cache, primary, replica1, replica2 = mock_cache
    
    cache.set('user:1', {'name': 'Alice'})
    
    primary.setex.assert_called_once()
    replica1.setex.assert_not_called()
    replica2.setex.assert_not_called()

def test_read_uses_replica(mock_cache):
    cache, primary, replica1, replica2 = mock_cache
    replica1.get.return_value = json.dumps({'name': 'Alice'})
    
    result = cache.get('user:1')
    
    assert result == {'name': 'Alice'}
    primary.get.assert_not_called()
    replica1.get.assert_called_once_with('user:1')

def test_read_your_writes_reads_from_primary_when_flag_set(mock_cache):
    cache, primary, replica1, replica2 = mock_cache
    
    # Flag exists (session recently wrote)
    primary.exists.return_value = True
    primary.get.return_value = json.dumps({'name': 'Alice Updated'})
    
    result = cache.get_with_read_your_writes('user:1', 'session-abc')
    
    assert result == {'name': 'Alice Updated'}
    primary.get.assert_called_once_with('user:1')
    replica1.get.assert_not_called()

def test_read_your_writes_reads_from_replica_when_no_flag(mock_cache):
    cache, primary, replica1, replica2 = mock_cache
    
    # No flag — normal read
    primary.exists.return_value = False
    replica1.get.return_value = json.dumps({'name': 'Alice'})
    
    result = cache.get_with_read_your_writes('user:1', 'session-xyz')
    
    replica1.get.assert_called_once_with('user:1')
    primary.get.assert_not_called()

def test_ryw_flag_expires(mock_cache):
    cache, primary, replica1, replica2 = mock_cache
    primary.get.return_value = json.dumps({'name': 'Alice'})
    
    cache.set_with_ryw_flag('user:1', {'name': 'Alice'}, 'session-abc', ttl=300)
    
    # Flag should be set with 5s TTL
    flag_call = primary.setex.call_args_list[1]  # Second setex call
    assert flag_call[0][0] == 'ryw:session-abc:user:1'
    assert flag_call[0][1] == 5

Testing Eventual Consistency

With real Redis replication, you can measure replication lag and verify that your system eventually converges:

# test_eventual_consistency.py
import redis
import time
import pytest
from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import wait_for_logs

@pytest.fixture(scope='module')
def redis_primary_replica():
    """Start a Redis primary with one replica."""
    # Primary
    primary = DockerContainer('redis:7-alpine')
    primary.with_exposed_ports(6379)
    primary.start()
    
    primary_host = primary.get_container_host_ip()
    primary_port = primary.get_exposed_port(6379)
    
    # Replica — connect to primary
    replica = DockerContainer('redis:7-alpine')
    replica.with_exposed_ports(6379)
    
    # Get primary's internal IP for replica config
    primary_ip = primary.get_container_host_ip()
    replica.with_command(
        f'redis-server --replicaof {primary_ip} {primary_port} --replica-read-only yes'
    )
    replica.start()
    
    yield {
        'primary': redis.Redis(host=primary_host, port=int(primary_port), decode_responses=True),
        'replica': redis.Redis(
            host=replica.get_container_host_ip(),
            port=int(replica.get_exposed_port(6379)),
            decode_responses=True
        )
    }
    
    replica.stop()
    primary.stop()

def test_replication_lag_is_bounded(redis_primary_replica):
    """
    Test that writes on primary appear on replica within a bounded time window.
    This is your SLA for replication lag.
    """
    primary = redis_primary_replica['primary']
    replica = redis_primary_replica['replica']
    
    MAX_LAG_MS = 500  # Your acceptable replication lag SLA
    
    primary.set('consistency_test', 'value_123')
    write_time = time.time()
    
    deadline = write_time + (MAX_LAG_MS / 1000)
    
    while time.time() < deadline:
        value = replica.get('consistency_test')
        if value == 'value_123':
            lag_ms = (time.time() - write_time) * 1000
            print(f"Replication lag: {lag_ms:.1f}ms")
            return  # Test passes
        time.sleep(0.01)
    
    pytest.fail(f"Value not replicated within {MAX_LAG_MS}ms")

Testing Cache Stampede Under Concurrent Load

In a distributed system, multiple application instances might try to populate the same cache key simultaneously:

# distributed_lock_cache.py
import redis
import json
import time
import uuid

r = redis.Redis.from_url('redis://localhost', decode_responses=True)

LOCK_TTL = 10  # seconds

def get_with_distributed_lock(key: str, fetch_fn, ttl: int = 300):
    """
    Cache-aside with distributed lock to prevent stampede.
    Only one caller fetches on miss; others wait for the result.
    """
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    
    lock_key = f'lock:{key}'
    lock_value = str(uuid.uuid4())
    
    # Try to acquire distributed lock
    acquired = r.set(lock_key, lock_value, nx=True, ex=LOCK_TTL)
    
    if acquired:
        try:
            # Double-check after acquiring lock
            cached = r.get(key)
            if cached:
                return json.loads(cached)
            
            value = fetch_fn()
            r.setex(key, ttl, json.dumps(value))
            return value
        finally:
            # Release lock only if we still own it (Lua script for atomicity)
            release_script = """
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("DEL", KEYS[1])
            else
                return 0
            end
            """
            r.eval(release_script, 1, lock_key, lock_value)
    else:
        # Wait for the lock holder to populate the cache
        deadline = time.time() + LOCK_TTL
        while time.time() < deadline:
            time.sleep(0.05)
            cached = r.get(key)
            if cached:
                return json.loads(cached)
        
        # Lock holder failed — fetch directly
        return fetch_fn()

# test_distributed_lock.py
import threading
import time
import pytest
import fakeredis
import distributed_lock_cache

def test_distributed_lock_prevents_stampede(monkeypatch):
    fake_r = fakeredis.FakeRedis(decode_responses=True)
    monkeypatch.setattr(distributed_lock_cache, 'r', fake_r)
    
    fetch_count = 0
    fetch_delay = 0.1
    
    def slow_fetch():
        nonlocal fetch_count
        fetch_count += 1
        time.sleep(fetch_delay)
        return {'data': 'expensive'}
    
    results = []
    errors = []
    ready = threading.Barrier(10)
    
    def request():
        ready.wait()  # All threads start simultaneously
        try:
            result = distributed_lock_cache.get_with_distributed_lock(
                'shared_key', slow_fetch
            )
            results.append(result)
        except Exception as e:
            errors.append(e)
    
    threads = [threading.Thread(target=request) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=15)
    
    assert len(errors) == 0
    assert len(results) == 10
    assert all(r == {'data': 'expensive'} for r in results)
    
    # Key assertion: fetch_fn called only once despite 10 concurrent misses
    assert fetch_count == 1

def test_lock_released_after_populate(monkeypatch):
    fake_r = fakeredis.FakeRedis(decode_responses=True)
    monkeypatch.setattr(distributed_lock_cache, 'r', fake_r)
    
    fetch = lambda: {'value': 1}
    distributed_lock_cache.get_with_distributed_lock('key', fetch)
    
    # Lock should be released after successful population
    assert fake_r.get('lock:key') is None

def test_lock_not_released_by_non_owner(monkeypatch):
    """Test that the Lua release script prevents lock theft."""
    fake_r = fakeredis.FakeRedis(decode_responses=True)
    monkeypatch.setattr(distributed_lock_cache, 'r', fake_r)
    
    # Set a lock with a known value
    fake_r.set('lock:test_key', 'owner-uuid', ex=10)
    
    # Attempt to release with wrong value (simulating a different process)
    script = """
    if redis.call("GET", KEYS[1]) == ARGV[1] then
        return redis.call("DEL", KEYS[1])
    else
        return 0
    end
    """
    result = fake_r.eval(script, 1, 'lock:test_key', 'wrong-uuid')
    
    assert result == 0  # Did not release
    assert fake_r.get('lock:test_key') == 'owner-uuid'  # Lock still held

Testing Cache Coherence Across Application Instances

When multiple application instances share a Redis cache, test that writes from one instance are visible to others:

# test_cross_instance_coherence.py
import fakeredis
import json
import pytest

def make_app_instance(fake_server):
    """Create an app instance connected to a shared Redis server."""
    r = fakeredis.FakeRedis(server=fake_server, decode_responses=True)
    
    def write(key, value):
        r.set(key, json.dumps(value))
    
    def read(key):
        val = r.get(key)
        return json.loads(val) if val else None
    
    return {'write': write, 'read': read}

def test_write_from_instance_a_visible_to_instance_b():
    server = fakeredis.FakeServer()
    
    instance_a = make_app_instance(server)
    instance_b = make_app_instance(server)
    
    instance_a['write']('config:feature_flag', {'enabled': True})
    
    result = instance_b['read']('config:feature_flag')
    assert result == {'enabled': True}

def test_delete_from_one_instance_affects_all():
    server = fakeredis.FakeServer()
    
    instance_a = make_app_instance(server)
    instance_b = make_app_instance(server)
    instance_c = make_app_instance(server)
    
    instance_a['write']('shared_state', {'value': 42})
    
    # Verify all can read
    assert instance_b['read']('shared_state') is not None
    assert instance_c['read']('shared_state') is not None
    
    # Delete from instance A
    r_a = fakeredis.FakeRedis(server=fakeredis.FakeServer(), decode_responses=True)
    # Properly: delete via same server
    instance_a_r = fakeredis.FakeRedis(server=server, decode_responses=True)
    instance_a_r.delete('shared_state')
    
    # All instances should see deletion
    assert instance_b['read']('shared_state') is None
    assert instance_c['read']('shared_state') is None

Testing Cache Warm-Up and Cold-Start

Production incidents often happen during cold starts when the cache is empty and all requests hit the DB simultaneously:

# test_cache_warmup.py
import threading
import time
import pytest
import fakeredis
from unittest.mock import MagicMock

def test_cold_start_does_not_overwhelm_database(monkeypatch):
    fake_r = fakeredis.FakeRedis(decode_responses=True)
    
    db_calls = []
    db_lock = threading.Lock()
    
    def db_fetch(user_id):
        with db_lock:
            db_calls.append(user_id)
        time.sleep(0.02)  # Simulate DB latency
        return {'id': user_id, 'name': f'User {user_id}'}
    
    import cache_service
    monkeypatch.setattr(cache_service, 'r', fake_r)
    monkeypatch.setattr(cache_service, 'fetch_from_db', db_fetch)
    
    # Simulate 50 concurrent requests for 5 different users
    threads = []
    results = {i: [] for i in range(1, 6)}
    
    for user_id in range(1, 6):
        for _ in range(10):  # 10 concurrent requests per user
            t = threading.Thread(
                target=lambda uid=user_id: results[uid].append(
                    cache_service.get_user(uid)
                )
            )
            threads.append(t)
    
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=10)
    
    # Each user's data should only be fetched once from DB
    for user_id in range(1, 6):
        user_fetches = [c for c in db_calls if c == user_id]
        assert len(user_fetches) == 1, \
            f"User {user_id} fetched {len(user_fetches)} times (expected 1)"

Verifying Cache Hit Rates

# test_cache_hit_rates.py
import fakeredis
import pytest
from unittest.mock import patch, MagicMock

def test_warm_cache_achieves_target_hit_rate(monkeypatch):
    fake_r = fakeredis.FakeRedis(decode_responses=True)
    fetch_count = 0
    total_requests = 100
    
    def tracked_fetch(key):
        nonlocal fetch_count
        fetch_count += 1
        return {'key': key, 'data': 'value'}
    
    import cache_service
    monkeypatch.setattr(cache_service, 'r', fake_r)
    monkeypatch.setattr(cache_service, 'fetch_from_db', tracked_fetch)
    
    # Warm the cache with 10 unique keys
    unique_keys = [f'item:{i}' for i in range(10)]
    for key in unique_keys:
        cache_service.get(key)
    
    fetch_count = 0  # Reset counter after warmup
    
    # Now simulate 100 requests — 80% to cached keys, 20% to new keys
    import random
    for _ in range(80):
        cache_service.get(random.choice(unique_keys))
    for i in range(20):
        cache_service.get(f'new_item:{i}')
    
    hits = total_requests - fetch_count
    hit_rate = hits / total_requests
    
    TARGET_HIT_RATE = 0.75  # 75% minimum hit rate
    assert hit_rate >= TARGET_HIT_RATE, \
        f"Hit rate {hit_rate:.1%} below target {TARGET_HIT_RATE:.1%}"

Conclusion

Distributed cache consistency testing requires you to reason about what guarantee your system provides and write tests that falsify violations. For read-your-writes: verify that writes to primary are visible to reads from that same session. For eventual consistency: measure the actual replication lag. For stampede protection: use barriers to create genuine concurrent load and assert that fetch functions are called exactly once. The bugs that matter in production—stale data, thundering herds, split-brain reads—are all testable if you design your tests to create the conditions where they occur.