Eric Campbell 9 min read Architecture

System Design: Eventual Consistency Patterns

The CAP theorem forces a choice: during a network partition, you either sacrifice consistency or availability. Strong consistency means every read returns the most recent write, but achieving this...

Key Insights

  • Eventual consistency isn’t a compromise—it’s a deliberate architectural choice that enables horizontal scaling, partition tolerance, and high availability when strong consistency would create unacceptable bottlenecks.
  • CRDTs and event sourcing shift conflict resolution from runtime decisions to data structure design, eliminating entire categories of distributed systems bugs.
  • The hardest part of eventual consistency isn’t the implementation—it’s designing user experiences that remain coherent when different users see different versions of the truth.

Introduction to Eventual Consistency

The CAP theorem forces a choice: during a network partition, you either sacrifice consistency or availability. Strong consistency means every read returns the most recent write, but achieving this requires coordination that blocks operations during partitions. Eventual consistency accepts temporary divergence in exchange for always-on writes and reads.

This isn’t about being lazy with data integrity. It’s about recognizing that many systems can tolerate brief inconsistency windows. Your Twitter feed doesn’t need linearizable reads. Your shopping cart can handle a few seconds of stale data. Your analytics dashboard definitely doesn’t need real-time consistency.

Choose strong consistency when correctness trumps availability: financial transactions, inventory that can’t oversell, anything where inconsistent reads cause real damage. Choose eventual consistency when you need geographic distribution, high write throughput, or fault tolerance that keeps working through network issues.

The patterns in this article help you implement eventual consistency correctly—ensuring your system actually converges to a consistent state rather than drifting into permanent divergence.

Core Patterns Overview

Eventual consistency patterns fall into four main categories:

Read Repair detects inconsistencies when clients read data. If replicas disagree, the system fixes them on the spot. This is reactive—problems get fixed when they’re encountered.

Write-Behind (Async Replication) accepts writes locally and propagates them asynchronously. Fast writes, eventual propagation. The challenge is handling conflicts when concurrent writes hit different replicas.

Anti-Entropy runs background processes that continuously compare replicas and reconcile differences. Proactive consistency maintenance that catches drift before reads expose it.

CRDTs (Conflict-Free Replicated Data Types) design data structures that mathematically guarantee convergence. No conflict resolution logic needed—the merge operation is built into the data type itself.

Most production systems combine these patterns. Cassandra uses read repair plus anti-entropy. DynamoDB uses vector clocks with write-behind replication. Riak pioneered CRDTs for shopping cart scenarios.

Read Repair Pattern

Read repair works by querying multiple replicas and comparing responses. When replicas disagree, the system updates stale replicas with the current value before returning to the client.

The key mechanism is version comparison. Each write increments a version vector—a map of node IDs to logical timestamps. When comparing two values, the one with a dominating version vector (all components greater than or equal) wins. If neither dominates, you have a conflict requiring application-level resolution.

from dataclasses import dataclass, field
from typing import Dict, Optional, List
import random

@dataclass
class VersionedValue:
    value: any
    version_vector: Dict[str, int] = field(default_factory=dict)
    
    def dominates(self, other: 'VersionedValue') -> bool:
        """Returns True if self's version vector dominates other's."""
        if not other.version_vector:
            return bool(self.version_vector)
        
        for node, timestamp in other.version_vector.items():
            if self.version_vector.get(node, 0) < timestamp:
                return False
        return self.version_vector != other.version_vector
    
    def concurrent_with(self, other: 'VersionedValue') -> bool:
        """Returns True if neither version dominates."""
        return not self.dominates(other) and not other.dominates(self)

class ReplicaNode:
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.data: Dict[str, VersionedValue] = {}
    
    def write(self, key: str, value: any, incoming_version: Dict[str, int] = None):
        current = self.data.get(key)
        new_version = dict(incoming_version) if incoming_version else {}
        new_version[self.node_id] = new_version.get(self.node_id, 0) + 1
        
        self.data[key] = VersionedValue(value=value, version_vector=new_version)
    
    def read(self, key: str) -> Optional[VersionedValue]:
        return self.data.get(key)

class ReadRepairCoordinator:
    def __init__(self, replicas: List[ReplicaNode], read_quorum: int):
        self.replicas = replicas
        self.read_quorum = read_quorum
    
    def read_with_repair(self, key: str) -> Optional[any]:
        # Query quorum of replicas
        queried = random.sample(self.replicas, self.read_quorum)
        responses = [(node, node.read(key)) for node in queried]
        
        # Filter out None responses
        valid_responses = [(n, v) for n, v in responses if v is not None]
        if not valid_responses:
            return None
        
        # Find the winning value
        winner = max(valid_responses, key=lambda x: sum(x[1].version_vector.values()))
        winning_value = winner[1]
        
        # Repair stale replicas
        for node, value in valid_responses:
            if value and winning_value.dominates(value):
                node.write(key, winning_value.value, winning_value.version_vector)
                print(f"Repaired stale replica on {node.node_id}")
        
        return winning_value.value

# Usage demonstration
nodes = [ReplicaNode(f"node-{i}") for i in range(3)]
coordinator = ReadRepairCoordinator(nodes, read_quorum=2)

# Simulate inconsistent state (node-0 has newer data)
nodes[0].write("user:123", {"name": "Alice", "email": "alice@new.com"})
nodes[1].write("user:123", {"name": "Alice", "email": "alice@old.com"})

# Read repair will detect and fix the inconsistency
result = coordinator.read_with_repair("user:123")

The quorum size matters. With N replicas, reading from R replicas and writing to W replicas guarantees consistency if R + W > N. Common configurations: (3, 2, 2) gives you consistency with one node failure tolerance.

Conflict-Free Replicated Data Types (CRDTs)

CRDTs eliminate conflict resolution by designing data structures where all merge operations commute, associate, and are idempotent. No matter what order operations arrive, all replicas converge to the same state.

The simplest CRDT is a G-Counter (grow-only counter). Each node maintains its own counter. The value is the sum of all node counters. Merging takes the maximum of each node’s counter.

from typing import Dict
from dataclasses import dataclass, field

@dataclass
class GCounter:
    """Grow-only counter CRDT. Supports increment and merge, never decrements."""
    node_id: str
    counts: Dict[str, int] = field(default_factory=dict)
    
    def increment(self, amount: int = 1):
        """Increment this node's counter."""
        self.counts[self.node_id] = self.counts.get(self.node_id, 0) + amount
    
    def value(self) -> int:
        """Get the total count across all nodes."""
        return sum(self.counts.values())
    
    def merge(self, other: 'GCounter') -> 'GCounter':
        """Merge with another G-Counter. Takes max of each node's count."""
        merged_counts = dict(self.counts)
        for node, count in other.counts.items():
            merged_counts[node] = max(merged_counts.get(node, 0), count)
        
        result = GCounter(node_id=self.node_id)
        result.counts = merged_counts
        return result

# Simulate distributed counting across three data centers
dc_east = GCounter("dc-east")
dc_west = GCounter("dc-west")
dc_eu = GCounter("dc-eu")

# Each DC processes local increments
dc_east.increment(100)  # 100 page views in east
dc_west.increment(150)  # 150 page views in west
dc_eu.increment(75)     # 75 page views in EU

print(f"DC East local view: {dc_east.value()}")   # 100
print(f"DC West local view: {dc_west.value()}")   # 150

# Async replication: east receives west's state
dc_east = dc_east.merge(dc_west)
print(f"DC East after merge with West: {dc_east.value()}")  # 250

# More local increments happen before full sync
dc_east.increment(50)
dc_eu.increment(25)

# Eventually all DCs sync
final_east = dc_east.merge(dc_eu)
final_west = dc_west.merge(dc_east).merge(dc_eu)
final_eu = dc_eu.merge(dc_east).merge(dc_west)

# All converge to same value regardless of merge order
print(f"Final East: {final_east.value()}")   # 400
print(f"Final West: {final_west.value()}")   # 400
print(f"Final EU: {final_eu.value()}")       # 400

For more complex use cases, consider:

  • PN-Counter: Combines two G-Counters (positive and negative) to support decrements
  • LWW-Register: Last-writer-wins using timestamps, good for simple key-value data
  • OR-Set: Observed-remove set that handles concurrent add/remove of the same element

CRDTs trade off expressiveness for guaranteed convergence. You can’t implement arbitrary logic, but what you can implement will always be correct.

Event Sourcing with Async Replication

Event sourcing stores state changes as an immutable log of events. Current state is derived by replaying events. This pairs naturally with eventual consistency—replicas consume the event stream and converge as they catch up.

import time
import json
from dataclasses import dataclass
from typing import List, Dict, Callable
from threading import Thread
from queue import Queue

@dataclass
class Event:
    event_id: str
    event_type: str
    aggregate_id: str
    payload: dict
    timestamp: float
    
    def to_dict(self):
        return {
            "event_id": self.event_id,
            "event_type": self.event_type,
            "aggregate_id": self.aggregate_id,
            "payload": self.payload,
            "timestamp": self.timestamp
        }

class EventStore:
    """Append-only event log with subscriber support."""
    
    def __init__(self):
        self.events: List[Event] = []
        self.subscribers: List[Queue] = []
        self._event_counter = 0
    
    def append(self, event_type: str, aggregate_id: str, payload: dict) -> Event:
        self._event_counter += 1
        event = Event(
            event_id=f"evt-{self._event_counter}",
            event_type=event_type,
            aggregate_id=aggregate_id,
            payload=payload,
            timestamp=time.time()
        )
        self.events.append(event)
        
        # Notify all subscribers (async replication simulation)
        for queue in self.subscribers:
            queue.put(event)
        
        return event
    
    def subscribe(self) -> Queue:
        queue = Queue()
        self.subscribers.append(queue)
        return queue
    
    def replay_from(self, position: int) -> List[Event]:
        return self.events[position:]

class ReplicaProjection:
    """Maintains a materialized view from event stream."""
    
    def __init__(self, replica_id: str):
        self.replica_id = replica_id
        self.state: Dict[str, dict] = {}
        self.processed_events: set = set()
        self.last_position = 0
    
    def apply_event(self, event: Event):
        # Idempotency check - critical for at-least-once delivery
        if event.event_id in self.processed_events:
            print(f"[{self.replica_id}] Skipping duplicate: {event.event_id}")
            return
        
        if event.event_type == "UserCreated":
            self.state[event.aggregate_id] = {
                "id": event.aggregate_id,
                "name": event.payload["name"],
                "email": event.payload["email"],
                "created_at": event.timestamp
            }
        elif event.event_type == "UserEmailChanged":
            if event.aggregate_id in self.state:
                self.state[event.aggregate_id]["email"] = event.payload["new_email"]
        elif event.event_type == "UserDeleted":
            self.state.pop(event.aggregate_id, None)
        
        self.processed_events.add(event.event_id)
        self.last_position += 1
        print(f"[{self.replica_id}] Applied {event.event_type} for {event.aggregate_id}")
    
    def consume_from_queue(self, queue: Queue, delay: float = 0):
        """Simulate async consumption with optional artificial delay."""
        while True:
            event = queue.get()
            if event is None:  # Shutdown signal
                break
            time.sleep(delay)  # Simulate network latency
            self.apply_event(event)

# Demonstration
store = EventStore()

# Create two replicas with different "network latency"
replica_fast = ReplicaProjection("replica-fast")
replica_slow = ReplicaProjection("replica-slow")

queue_fast = store.subscribe()
queue_slow = store.subscribe()

# Start consumers in background threads
Thread(target=replica_fast.consume_from_queue, args=(queue_fast, 0.01), daemon=True).start()
Thread(target=replica_slow.consume_from_queue, args=(queue_slow, 0.1), daemon=True).start()

# Produce events
store.append("UserCreated", "user-1", {"name": "Alice", "email": "alice@example.com"})
store.append("UserCreated", "user-2", {"name": "Bob", "email": "bob@example.com"})
store.append("UserEmailChanged", "user-1", {"new_email": "alice@newdomain.com"})

time.sleep(0.05)
print(f"\nFast replica state: {replica_fast.state}")
print(f"Slow replica state: {replica_slow.state}")

time.sleep(0.2)
print(f"\nAfter convergence:")
print(f"Fast replica state: {replica_fast.state}")
print(f"Slow replica state: {replica_slow.state}")

Key requirements for event-sourced eventual consistency:

  1. Idempotent handlers: Events may be delivered multiple times. Your projection must handle duplicates.
  2. Ordered delivery per aggregate: Events for the same entity must be processed in order. Cross-entity ordering often doesn’t matter.
  3. Checkpointing: Track consumption position to resume after failures without replaying everything.

Anti-Entropy and Merkle Trees

Anti-entropy processes run continuously in the background, comparing data between replicas and reconciling differences. The challenge is efficiency—you can’t compare every record on every sync.

Merkle trees solve this by creating a hierarchical hash structure. Nodes compare root hashes first. If they match, data is identical. If not, they recursively compare children to identify exactly which data ranges differ.

import hashlib
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass

@dataclass
class MerkleNode:
    hash: str
    left: Optional['MerkleNode'] = None
    right: Optional['MerkleNode'] = None
    key_range: Tuple[str, str] = ("", "")
    is_leaf: bool = False
    data_key: Optional[str] = None

class MerkleTree:
    """Simplified Merkle tree for key-value data synchronization."""
    
    def __init__(self, data: Dict[str, str]):
        self.data = data
        self.root = self._build_tree(sorted(data.keys()))
    
    def _hash(self, value: str) -> str:
        return hashlib.sha256(value.encode()).hexdigest()[:16]
    
    def _build_tree(self, keys: List[str]) -> Optional[MerkleNode]:
        if not keys:
            return None
        
        if len(keys) == 1:
            key = keys[0]
            return MerkleNode(
                hash=self._hash(f"{key}:{self.data[key]}"),
                key_range=(key, key),
                is_leaf=True,
                data_key=key
            )
        
        mid = len(keys) // 2
        left = self._build_tree(keys[:mid])
        right = self._build_tree(keys[mid:])
        
        combined_hash = self._hash(
            (left.hash if left else "") + (right.hash if right else "")
        )
        
        return MerkleNode(
            hash=combined_hash,
            left=left,
            right=right,
            key_range=(keys[0], keys[-1])
        )

def find_differences(tree1: MerkleTree, tree2: MerkleTree) -> List[str]:
    """Compare two Merkle trees and return keys that differ."""
    differences = []
    
    def compare_nodes(node1: Optional[MerkleNode], node2: Optional[MerkleNode]):
        # One side missing entirely
        if node1 is None and node2 is None:
            return
        if node1 is None:
            collect_all_keys(node2, differences)
            return
        if node2 is None:
            collect_all_keys(node1, differences)
            return
        
        # Hashes match - subtrees are identical
        if node1.hash == node2.hash:
            return
        
        # Leaf nodes with different hashes
        if node1.is_leaf and node2.is_leaf:
            differences.append(node1.data_key)
            return
        
        # Recurse into children
        compare_nodes(node1.left, node2.left)
        compare_nodes(node1.right, node2.right)
    
    def collect_all_keys(node: MerkleNode, keys: List[str]):
        if node.is_leaf:
            keys.append(node.data_key)
        else:
            if node.left:
                collect_all_keys(node.left, keys)
            if node.right:
                collect_all_keys(node.right, keys)
    
    compare_nodes(tree1.root, tree2.root)
    return differences

# Simulate two replicas with some drift
replica_a_data = {
    "user:001": "alice@example.com",
    "user:002": "bob@example.com",
    "user:003": "charlie@example.com",
    "user:004": "diana@example.com",
    "user:005": "eve@example.com",
}

replica_b_data = {
    "user:001": "alice@example.com",
    "user:002": "bob@CHANGED.com",      # Different!
    "user:003": "charlie@example.com",
    "user:004": "diana@CHANGED.com",    # Different!
    "user:005": "eve@example.com",
}

tree_a = MerkleTree(replica_a_data)
tree_b = MerkleTree(replica_b_data)

print(f"Replica A root hash: {tree_a.root.hash}")
print(f"Replica B root hash: {tree_b.root.hash}")
print(f"Roots match: {tree_a.root.hash == tree_b.root.hash}")

differing_keys = find_differences(tree_a, tree_b)
print(f"\nKeys that differ: {differing_keys}")
print(f"Comparisons needed: ~{len(differing_keys) * 2} (vs {len(replica_a_data)} full scan)")

Cassandra uses Merkle trees during repair operations. Each node builds a tree over its token range, exchanges root hashes with replicas, and only transfers data for differing ranges. This reduces repair bandwidth by orders of magnitude compared to full data comparison.

Practical Considerations

Monitoring consistency lag: Track the delta between write time and when all replicas have the update. Percentile metrics matter more than averages—p99 lag tells you the worst-case user experience.

# Key metrics to track
metrics = {
    "replication_lag_p50_ms": 45,
    "replication_lag_p99_ms": 1200,
    "inconsistent_read_rate": 0.003,  # 0.3% of reads see stale data
    "repair_operations_per_hour": 12,
    "merkle_tree_differences_found": 847,
}

Handling stale reads in UX: Design interfaces that tolerate temporary inconsistency. Show “saving…” states that persist until confirmation. Use optimistic updates with rollback on conflict. Display timestamps so users understand data freshness.

Debugging distributed state: Log version vectors and event IDs with every operation. Build tools to query “what does each replica think this key’s value is?” Implement read-your-writes consistency at the session level even if global consistency is eventual.

Testing: Inject artificial delays and partitions. Use property-based testing to verify convergence: generate random operation sequences, apply them in different orders to different replicas, assert final states match.

Real-world implementations:

  • Cassandra: Tunable consistency with read repair and anti-entropy repair
  • DynamoDB: Vector clocks for conflict detection, application-level resolution
  • CockroachDB: Raft consensus for strong consistency, but understanding eventual consistency helps when you need to relax guarantees for performance
  • Redis Cluster: Async replication with last-writer-wins, simple but lossy

Eventual consistency isn’t a bug to be worked around—it’s a tool that enables distributed systems to scale beyond what’s possible with strong consistency. Master these patterns, and you’ll build systems that stay available and eventually correct, even when the network doesn’t cooperate.

Liked this? There's more.

Every week: one practical technique, explained simply, with code you can use immediately.