Build a Distributed Key-Value Store
Implement consistent hashing, vector clocks, and gossip in Python.
Introduction
30 Second Summary
Every time you add an item to your cart on Amazon or update a shared document, multiple servers across the world need to agree on what just happened. When those servers lose contact with each other, the system has to make hard choices about consistency and availability.
In this project, you will build a distributed key-value store from scratch in pure Python. You will implement the same core concepts that power real-world systems like Amazon DynamoDB and Apache Cassandra.
What You'll Build
A fully functional distributed key-value store that runs a 3-node cluster locally and demonstrates real distributed systems behavior, including network partitions and automatic convergence.
By the end of this project, you'll have:
- A consistent hash ring with virtual nodes that evenly distributes keys across nodes and handles node removal with minimal data movement.
- A replicated KV store with vector clocks that tracks causality and detects conflicting concurrent writes across nodes.
- A partition simulator with anti-entropy gossip that demonstrates network splits, data divergence, and automatic convergence.
- Secret Mission: Add quorum reads and writes (W + R > N) to guarantee consistency levels.
Are there any prerequisites?
You need basic Python proficiency (classes, dictionaries, lists). No distributed systems experience is required.
This project uses zero external dependencies. Everything is built with the Python standard library, so you implement every concept from scratch.
Before We Start
Distributed key-value stores power services like Amazon DynamoDB, Apache Cassandra, and Riak. They solve one of the hardest problems in computing: keeping data available and consistent across multiple machines, even when some machines lose contact with each other.
In this project, you'll build a distributed KV store from scratch in pure Python. First, let's set up the project directory and files you'll need.
In this step, get ready to:
- Verify Python 3.7+ is installed and working.
- Create the distributed-kv-store project directory.
- Create the five module files for the project.
Verify Python 3 is installed
This project uses Python 3.7+ for f-strings and type hints. It has zero external dependencies, so no package installs are required.
- Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), then type Terminal and press Enter.
- Check your Python version by running this command:
python3 --version✔️ I see a version number
You should see something like Python 3.11.4 (or any version 3.7 or higher). You're good to go.
ⓧ Command not found
You need to install Python 3. Visit the official Python downloads page and install the latest version for your operating system.
- Go to python.org/downloads.
- Download and run the installer for your platform.
- On Windows, make sure to check Add Python to PATH during installation.
- After installation, close and reopen your terminal, then run python3 --version again.
Create the project directory and files
Your distributed KV store is split across five Python modules. Each file handles one responsibility: hashing, vector clocks, individual nodes, cluster coordination, and the demo script.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir distributed-kv-store
cd distributed-kv-store- Create the five module files by running this command:
touch hash_ring.py vector_clock.py node.py cluster.py demo.py- Confirm the files exist by running:
lsYou should see cluster.py, demo.py, hash_ring.py, node.py, and vector_clock.py listed in the output.
Don't see all five files?
Make sure you ran the touch command from inside the distributed-kv-store folder. If you're in the wrong directory, run cd ~/Desktop/distributed-kv-store and try the touch command again.
Help me create files in the right directory.
What is each file for?
- hash_ring.py distributes keys evenly across nodes using consistent hashing.
- vector_clock.py tracks causality so the system detects conflicting writes.
- node.py represents a single machine in the cluster, storing and replicating data.
- cluster.py coordinates all nodes, handles replication, and simulates network partitions.
- demo.py runs the full demonstration showing every concept in action.
Open the project in your editor
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type VS Code and press Enter to open it.
- Click File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop and select the distributed-kv-store folder, then click Open.
You should see all five .py files listed in the Explorer sidebar on the left.
Your project is ready. Next up, you'll implement consistent hashing to distribute keys evenly across nodes.
Implement Consistent Hashing with Virtual Nodes
Your five module files are ready. Now it's time to build the foundation of any distributed key-value store: the mechanism that decides which node owns which data.
Consistent hashing solves a critical problem. When you add or remove a node from a cluster, you want to move as little data as possible. In this step, you'll build a hash ring with virtual nodes that distributes keys evenly and minimizes disruption during topology changes.
In this step, get ready to:
- Build a HashRing class that maps keys to positions using SHA-256.
- Implement a preference list that finds N distinct nodes for replication.
- Test key distribution across 3 nodes and verify minimal disruption when removing a node.
Build the HashRing class with SHA-256 hashing
A consistent hash ring maps both keys and nodes onto a circular number space. Each key belongs to the first node you encounter when walking clockwise from the key's position on the ring.
- Open hash_ring.py in your code editor and add the class foundation:
import hashlib
from typing import Dict, List, Optional
class HashRing:
"""Consistent hash ring with virtual nodes for even key distribution."""
def __init__(self, virtual_nodes: int = 150):
self.virtual_nodes = virtual_nodes
self.ring: Dict[int, str] = {}
self.sorted_keys: List[int] = []
self.nodes: set = set()
def _hash(self, key: str) -> int:
"""Generate a position on the ring using SHA-256."""
digest = hashlib.sha256(key.encode()).hexdigest()
return int(digest[:16], 16)What does this code do?
The HashRing class sets up a circular number space. self.ring maps positions (integers) to node IDs, and self.sorted_keys keeps those positions sorted so you can walk clockwise.
The _hash method uses SHA-256 to convert any string into a position on the ring. Taking the first 16 hex characters gives a 64-bit integer, which is more than enough positions for an even distribution.
Why 150 virtual nodes?
With only one position per physical node, keys cluster unevenly. Virtual nodes solve this by giving each physical node many positions on the ring (150 is standard for even distribution). From the outside, you still see 3 nodes. Internally, the ring has 450 evenly-scattered positions.
Now add the methods for adding and removing nodes from the ring. These are the operations that make consistent hashing powerful.
- Add the following two methods inside the HashRing class, below the _hash method:
def add_node(self, node_id: str) -> None:
"""Add a node with virtual nodes to the ring."""
self.nodes.add(node_id)
for i in range(self.virtual_nodes):
virtual_key = f"{node_id}:vn{i}"
position = self._hash(virtual_key)
self.ring[position] = node_id
self.sorted_keys = sorted(self.ring.keys())
def remove_node(self, node_id: str) -> None:
"""Remove a node and all its virtual nodes from the ring."""
self.nodes.discard(node_id)
self.ring = {pos: nid for pos, nid in self.ring.items() if nid != node_id}
self.sorted_keys = sorted(self.ring.keys())What does this code do?
add_node creates 150 virtual keys (like node-A:vn0, node-A:vn1, etc.), hashes each one to a position, and maps that position back to the physical node. This scatters the node's ownership across the entire ring.
remove_node filters out all positions belonging to that node. Only keys that were assigned to the removed node need to find a new home. Every other key stays put.
With nodes on the ring, you need a way to look up which node owns a given key. This is the clockwise lookup.
- Add the get_node method below remove_node:
def get_node(self, key: str) -> Optional[str]:
"""Find the node responsible for a key (clockwise lookup)."""
if not self.sorted_keys:
return None
position = self._hash(key)
for ring_pos in self.sorted_keys:
if ring_pos >= position:
return self.ring[ring_pos]
# Wrap around to the first node
return self.ring[self.sorted_keys[0]]What does this code do?
get_node hashes the key to find its position on the ring, then walks clockwise through sorted_keys until it finds the first ring position that's equal to or greater than the key's position. That's the owning node.
If the key's position is past the last node on the ring, it wraps around to the first node. This is what makes the data structure a "ring" rather than a line.
- Save hash_ring.py.
Seeing an IndentationError?
Make sure all methods are indented one level inside the HashRing class (4 spaces). The def keyword for each method should be at column 4, and the method body at column 8.
Help me fix an IndentationError in my HashRing class
Implement the preference list for replication
In a distributed store, you don't want data living on just one node. If that node goes down, the data is gone. The preference list solves this by identifying N distinct physical nodes responsible for each key.
- Add the get_preference_list method below get_node in your HashRing class:
def get_preference_list(self, key: str, count: int) -> List[str]:
"""Get N distinct physical nodes for replication (walk clockwise)."""
if not self.sorted_keys:
return []
result: List[str] = []
position = self._hash(key)
# Find starting index
start_idx = 0
for i, ring_pos in enumerate(self.sorted_keys):
if ring_pos >= position:
start_idx = i
break
else:
start_idx = 0
# Walk clockwise collecting distinct physical nodes
visited = 0
while len(result) < count and visited < len(self.sorted_keys):
idx = (start_idx + visited) % len(self.sorted_keys)
node_id = self.ring[self.sorted_keys[idx]]
if node_id not in result:
result.append(node_id)
visited += 1
return resultWhat does this code do?
This method walks clockwise from the key's position, collecting node IDs. The critical detail is the if node_id not in result check. With 150 virtual nodes per physical node, you'll hit many positions belonging to the same node. The method skips duplicates and keeps walking until it finds count distinct physical nodes.
This is exactly how DynamoDB and Cassandra determine which nodes should hold replicas of a given key.
- Save hash_ring.py.
Not sure about the for/else syntax?
Python's for...else runs the else block only if the loop completes without hitting a break. Here, if no ring position is >= the key's position (the key hashed past all nodes), the else sets start_idx = 0 to wrap around.
Explain how the for/else pattern works in my get_preference_list method
Test key distribution across 3 nodes
The last method gives you a way to verify that keys are evenly distributed. You'll use it to confirm your ring is working correctly.
- Add the get_distribution method as the final method in your HashRing class:
def get_distribution(self, sample_keys: List[str]) -> Dict[str, int]:
"""Check how evenly keys are distributed across nodes."""
distribution: Dict[str, int] = {node: 0 for node in self.nodes}
for key in sample_keys:
node = self.get_node(key)
if node:
distribution[node] += 1
return distributionWhat does this code do?
get_distribution takes a list of sample keys, looks up which node owns each one, and counts how many keys land on each node. You'll use this to verify that 150 virtual nodes produce a roughly even split (close to 33% each for 3 nodes).
- Save hash_ring.py.
Now test that everything works. You'll create a ring with 3 nodes, check distribution, and verify the preference list returns distinct nodes.
- Run a quick test in your terminal by running this command:
python3 -c "
from hash_ring import HashRing
ring = HashRing(virtual_nodes=150)
for node in ['node-A', 'node-B', 'node-C']:
ring.add_node(node)
keys = [f'user:{i}' for i in range(1000)]
dist = ring.get_distribution(keys)
print('Distribution:', dist)
print('Preference list for user:42:', ring.get_preference_list('user:42', 3))
"You should see a distribution where each node holds roughly 300-370 keys out of 1000. The preference list for user:42 should return all three node IDs in some order.
Seeing a ModuleNotFoundError?
Make sure your terminal is inside the distributed-kv-store directory where hash_ring.py lives. Run ls to confirm you can see the file listed.
Help me fix a ModuleNotFoundError when importing hash_ring
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
Here's the complete hash_ring.py for comparison:
import hashlib
from typing import Dict, List, Optional
class HashRing:
"""Consistent hash ring with virtual nodes for even key distribution."""
def __init__(self, virtual_nodes: int = 150):
self.virtual_nodes = virtual_nodes
self.ring: Dict[int, str] = {}
self.sorted_keys: List[int] = []
self.nodes: set = set()
def _hash(self, key: str) -> int:
"""Generate a position on the ring using SHA-256."""
digest = hashlib.sha256(key.encode()).hexdigest()
return int(digest[:16], 16)
def add_node(self, node_id: str) -> None:
"""Add a node with virtual nodes to the ring."""
self.nodes.add(node_id)
for i in range(self.virtual_nodes):
virtual_key = f"{node_id}:vn{i}"
position = self._hash(virtual_key)
self.ring[position] = node_id
self.sorted_keys = sorted(self.ring.keys())
def remove_node(self, node_id: str) -> None:
"""Remove a node and all its virtual nodes from the ring."""
self.nodes.discard(node_id)
self.ring = {pos: nid for pos, nid in self.ring.items() if nid != node_id}
self.sorted_keys = sorted(self.ring.keys())
def get_node(self, key: str) -> Optional[str]:
"""Find the node responsible for a key (clockwise lookup)."""
if not self.sorted_keys:
return None
position = self._hash(key)
for ring_pos in self.sorted_keys:
if ring_pos >= position:
return self.ring[ring_pos]
# Wrap around to the first node
return self.ring[self.sorted_keys[0]]
def get_preference_list(self, key: str, count: int) -> List[str]:
"""Get N distinct physical nodes for replication (walk clockwise)."""
if not self.sorted_keys:
return []
result: List[str] = []
position = self._hash(key)
# Find starting index
start_idx = 0
for i, ring_pos in enumerate(self.sorted_keys):
if ring_pos >= position:
start_idx = i
break
else:
start_idx = 0
# Walk clockwise collecting distinct physical nodes
visited = 0
while len(result) < count and visited < len(self.sorted_keys):
idx = (start_idx + visited) % len(self.sorted_keys)
node_id = self.ring[self.sorted_keys[idx]]
if node_id not in result:
result.append(node_id)
visited += 1
return result
def get_distribution(self, sample_keys: List[str]) -> Dict[str, int]:
"""Check how evenly keys are distributed across nodes."""
distribution: Dict[str, int] = {node: 0 for node in self.nodes}
for key in sample_keys:
node = self.get_node(key)
if node:
distribution[node] += 1
return distributionWhy consistent hashing over modulo?
A naive approach (key % num_nodes) moves almost every key when you add or remove a node. With consistent hashing, only K/N keys need to move (where K is total keys and N is total nodes). For 1000 keys across 3 nodes, removing one node moves roughly 333 keys instead of reshuffling all 1000.
This is a key interview talking point. When an interviewer asks "how do you partition data across nodes," this minimal-disruption property is the reason consistent hashing is the answer.
Your hash ring is working. Keys distribute evenly, the preference list finds distinct nodes for replication, and removing a node only moves a fraction of keys. Next up, you'll build the actual KV nodes that store data and replicate writes across the cluster.
Build KV Nodes with Replication
Your hash ring knows which nodes are responsible for which keys. Now you need actual nodes that can store data and talk to each other.
In a real distributed system like DynamoDB, every write goes to a coordinator node that then copies the data to other nodes on the preference list. This is replication. If one node dies, the others still have the data.
In this step, get ready to:
- Create the KVNode class with put, get, and receive_replica methods.
- Build the Cluster class that coordinates writes and replicates to the preference list.
- Test that writing to one node automatically replicates data to all nodes.
Create the KVNode class with put, get, and receive_replica methods
The KVNode class represents a single machine in your cluster. Each node stores key-value pairs and keeps a replication log of every write it handles.
Before building the node, you need a stub for the vector_clock.py module. The node imports from it, so the stub prevents import errors. You will fill in the real vector clock logic in the next step.
- Create the vector clock stub by pasting this into vector_clock.py:
from enum import Enum
from typing import Dict
class ClockOrder(Enum):
BEFORE = "before"
AFTER = "after"
CONCURRENT = "concurrent"
EQUAL = "equal"
class VectorClock:
"""Vector clock for tracking causal ordering of events."""
def __init__(self):
self.clock: Dict[str, int] = {}
def increment(self, node_id: str) -> None:
pass
def merge(self, other: "VectorClock") -> None:
pass
def copy(self) -> "VectorClock":
new_clock = VectorClock()
new_clock.clock = dict(self.clock)
return new_clock
@staticmethod
def compare(a: "VectorClock", b: "VectorClock") -> ClockOrder:
return ClockOrder.EQUAL
def __repr__(self) -> str:
return f"VectorClock({self.clock})"Why a stub instead of the real thing?
Python needs all imported modules to exist before your code can run. The node module imports VectorClock and ClockOrder, so you need these classes to exist now.
The stub has working copy() and __repr__() methods, but increment and compare are placeholders. Replication still works because the stub always returns EQUAL, treating every incoming replica as new data.
- Save vector_clock.py.
Now you can build the node itself. Each node stores versioned values (a value paired with its vector clock and timestamp) and provides three core operations: writing, reading, and receiving replicated writes from other nodes.
- Build the KVNode class with its write path by pasting this into node.py:
import time
from typing import Any, Dict, List, Optional
from vector_clock import VectorClock, ClockOrder
class VersionedValue:
"""A value with its associated vector clock and metadata."""
def __init__(self, value: Any, clock: VectorClock, timestamp: float):
self.value = value
self.clock = clock
self.timestamp = timestamp
def __repr__(self) -> str:
return f"VersionedValue(value={self.value!r}, clock={self.clock})"
class KVNode:
"""A single node in the distributed key-value store."""
def __init__(self, node_id: str):
self.node_id = node_id
self.data: Dict[str, List[VersionedValue]] = {} # key -> list of versions
self.replication_log: List[Dict] = [] # log of all writes for syncing
def put(self, key: str, value: Any, context: Optional[VectorClock] = None) -> VectorClock:
# Start from the context clock (read-modify-write) or a fresh clock
clock = context.copy() if context else VectorClock()
clock.increment(self.node_id)
new_version = VersionedValue(
value=value,
clock=clock,
timestamp=time.time(),
)
if key in self.data:
# Keep only concurrent versions (discard those dominated by new write)
surviving: List[VersionedValue] = []
for existing in self.data[key]:
order = VectorClock.compare(existing.clock, clock)
if order == ClockOrder.CONCURRENT:
surviving.append(existing)
# BEFORE or EQUAL means existing is superseded
surviving.append(new_version)
self.data[key] = surviving
else:
self.data[key] = [new_version]
# Record in replication log for anti-entropy sync
self.replication_log.append({
"type": "put",
"key": key,
"value": value,
"clock": clock.copy(),
"timestamp": new_version.timestamp,
})
return clockWhat does this code do?
VersionedValue bundles a stored value with its vector clock and a timestamp. Every piece of data in the store is wrapped in this container.
The put method handles writes. It increments the vector clock, checks for concurrent versions (conflicts), and logs the write for later syncing. If an existing version is superseded, it gets discarded. If versions are concurrent, they are kept as siblings.
- Add the read and replication methods below the put method in node.py:
def get(self, key: str) -> List[VersionedValue]:
"""Read all versions of a key (may return siblings on conflict)."""
return self.data.get(key, [])
def receive_replica(self, key: str, value: Any, clock: VectorClock, timestamp: float) -> None:
"""Receive a replicated write from another node."""
incoming = VersionedValue(value=value, clock=clock, timestamp=timestamp)
if key not in self.data:
# First time seeing this key, just store it
self.data[key] = [incoming]
return
# Compare incoming with every existing version
surviving: List[VersionedValue] = []
for existing in self.data[key]:
order = VectorClock.compare(existing.clock, clock)
if order == ClockOrder.BEFORE:
continue # Existing is superseded by incoming
elif order == ClockOrder.AFTER:
return # Incoming is outdated, ignore it
elif order == ClockOrder.EQUAL:
return # Duplicate, ignore
else:
surviving.append(existing) # CONCURRENT: keep both
surviving.append(incoming)
self.data[key] = surviving
def get_all_data(self) -> Dict[str, List[VersionedValue]]:
"""Return all stored data (used for anti-entropy sync)."""
return self.dataWhat does receive_replica do?
When a node receives a replicated write, it compares the incoming vector clock against all existing versions of that key. There are four possible outcomes:
- BEFORE means the existing version is older. Discard it.
- AFTER means the incoming write is older. Ignore it.
- EQUAL means it is a duplicate. Ignore it.
- CONCURRENT means both writes happened independently. Keep both as siblings (a conflict).
- Save node.py.
Seeing an import error?
If Python complains about importing from vector_clock, make sure you saved vector_clock.py in the same folder as node.py. Both files must be inside distributed-kv-store/.
help me debug this import error in node.py
✔️ Awesome, I've got everything!
Great. Double check you have saved both vector_clock.py and node.py.
ⓧ I'd like to double check the full code
Here is the complete node.py file:
import time
from typing import Any, Dict, List, Optional
from vector_clock import VectorClock, ClockOrder
class VersionedValue:
"""A value with its associated vector clock and metadata."""
def __init__(self, value: Any, clock: VectorClock, timestamp: float):
self.value = value
self.clock = clock
self.timestamp = timestamp
def __repr__(self) -> str:
return f"VersionedValue(value={self.value!r}, clock={self.clock})"
class KVNode:
"""A single node in the distributed key-value store."""
def __init__(self, node_id: str):
self.node_id = node_id
self.data: Dict[str, List[VersionedValue]] = {}
self.replication_log: List[Dict] = []
def put(self, key: str, value: Any, context: Optional[VectorClock] = None) -> VectorClock:
clock = context.copy() if context else VectorClock()
clock.increment(self.node_id)
new_version = VersionedValue(
value=value,
clock=clock,
timestamp=time.time(),
)
if key in self.data:
surviving: List[VersionedValue] = []
for existing in self.data[key]:
order = VectorClock.compare(existing.clock, clock)
if order == ClockOrder.CONCURRENT:
surviving.append(existing)
surviving.append(new_version)
self.data[key] = surviving
else:
self.data[key] = [new_version]
self.replication_log.append({
"type": "put",
"key": key,
"value": value,
"clock": clock.copy(),
"timestamp": new_version.timestamp,
})
return clock
def get(self, key: str) -> List[VersionedValue]:
"""Read all versions of a key (may return siblings on conflict)."""
return self.data.get(key, [])
def receive_replica(self, key: str, value: Any, clock: VectorClock, timestamp: float) -> None:
"""Receive a replicated write from another node."""
incoming = VersionedValue(value=value, clock=clock, timestamp=timestamp)
if key not in self.data:
self.data[key] = [incoming]
return
surviving: List[VersionedValue] = []
for existing in self.data[key]:
order = VectorClock.compare(existing.clock, clock)
if order == ClockOrder.BEFORE:
continue
elif order == ClockOrder.AFTER:
return
elif order == ClockOrder.EQUAL:
return
else:
surviving.append(existing)
surviving.append(incoming)
self.data[key] = surviving
def get_all_data(self) -> Dict[str, List[VersionedValue]]:
"""Return all stored data (used for anti-entropy sync)."""
return self.dataBuild the Cluster class that coordinates writes and replication
Individual nodes can store data, but they do not know about each other. The Cluster class ties everything together. It owns the hash ring, creates nodes, and handles the coordination pattern: when a client writes to a coordinator node, the cluster replicates that write to every node on the key's preference list.
- Build the Cluster class by pasting this into cluster.py:
from typing import Any, Dict, List, Optional, Set, Tuple
from hash_ring import HashRing
from node import KVNode, VersionedValue
from vector_clock import VectorClock, ClockOrder
class Cluster:
"""Manages a cluster of KV nodes with replication and partition simulation."""
def __init__(self, node_ids: List[str], replication_factor: int = 3, virtual_nodes: int = 150):
self.replication_factor = replication_factor
self.ring = HashRing(virtual_nodes=virtual_nodes)
self.nodes: Dict[str, KVNode] = {}
self.partitions: Set[Tuple[str, str]] = set() # tracks network partitions
# Create a KVNode for each node ID and add it to the hash ring
for node_id in node_ids:
self.nodes[node_id] = KVNode(node_id)
self.ring.add_node(node_id)
def _can_communicate(self, node_a: str, node_b: str) -> bool:
"""Check if two nodes can communicate (no partition between them)."""
pair = tuple(sorted([node_a, node_b]))
return pair not in self.partitions
def put(self, coordinator: str, key: str, value: Any, context: Optional[VectorClock] = None) -> VectorClock:
# Validate the coordinator exists
if coordinator not in self.nodes:
raise ValueError(f"Unknown node: {coordinator}")
# Write to the coordinator node first
clock = self.nodes[coordinator].put(key, value, context)
# Replicate to all nodes on the preference list
targets = self.ring.get_preference_list(key, self.replication_factor)
for target_id in targets:
if target_id == coordinator:
continue # Already written locally
if self._can_communicate(coordinator, target_id):
self.nodes[target_id].receive_replica(
key=key,
value=value,
clock=clock,
timestamp=self.nodes[coordinator].replication_log[-1]["timestamp"],
)
return clock
def get(self, coordinator: str, key: str) -> List[VersionedValue]:
"""Read a key from the coordinator node."""
if coordinator not in self.nodes:
raise ValueError(f"Unknown node: {coordinator}")
return self.nodes[coordinator].get(key)What does the Cluster do?
The constructor creates a KVNode for each node ID and registers them on the hash ring. The partitions set will be used later to simulate network failures.
The put method writes to the coordinator first, then iterates the preference list and sends replicas to every reachable node. The _can_communicate check makes it easy to simulate partitions later without changing the replication logic.
- Save cluster.py.
Seeing a ModuleNotFoundError?
Cluster imports from hash_ring, node, and vector_clock. All four files (hash_ring.py, node.py, vector_clock.py, cluster.py) must be in the same distributed-kv-store/ folder.
help me fix this ModuleNotFoundError in cluster.py
✔️ Awesome, I've got everything!
Great. Double check you have saved cluster.py.
ⓧ I'd like to double check the full code
Here is the complete cluster.py file:
from typing import Any, Dict, List, Optional, Set, Tuple
from hash_ring import HashRing
from node import KVNode, VersionedValue
from vector_clock import VectorClock, ClockOrder
class Cluster:
"""Manages a cluster of KV nodes with replication and partition simulation."""
def __init__(self, node_ids: List[str], replication_factor: int = 3, virtual_nodes: int = 150):
self.replication_factor = replication_factor
self.ring = HashRing(virtual_nodes=virtual_nodes)
self.nodes: Dict[str, KVNode] = {}
self.partitions: Set[Tuple[str, str]] = set()
for node_id in node_ids:
self.nodes[node_id] = KVNode(node_id)
self.ring.add_node(node_id)
def _can_communicate(self, node_a: str, node_b: str) -> bool:
"""Check if two nodes can communicate (no partition between them)."""
pair = tuple(sorted([node_a, node_b]))
return pair not in self.partitions
def put(self, coordinator: str, key: str, value: Any, context: Optional[VectorClock] = None) -> VectorClock:
if coordinator not in self.nodes:
raise ValueError(f"Unknown node: {coordinator}")
clock = self.nodes[coordinator].put(key, value, context)
targets = self.ring.get_preference_list(key, self.replication_factor)
for target_id in targets:
if target_id == coordinator:
continue
if self._can_communicate(coordinator, target_id):
self.nodes[target_id].receive_replica(
key=key,
value=value,
clock=clock,
timestamp=self.nodes[coordinator].replication_log[-1]["timestamp"],
)
return clock
def get(self, coordinator: str, key: str) -> List[VersionedValue]:
"""Read a key from the coordinator node."""
if coordinator not in self.nodes:
raise ValueError(f"Unknown node: {coordinator}")
return self.nodes[coordinator].get(key)Test that writing to one node replicates to all nodes
Time to prove the system works. You will write data through one coordinator node and confirm that all three nodes in the cluster have a copy.
- Run this replication test by running:
python3 -c "
from cluster import Cluster
cluster = Cluster(node_ids=['node-A', 'node-B', 'node-C'], replication_factor=3)
cluster.put('node-A', 'user:1', {'name': 'Alice', 'role': 'engineer'})
print('Replication test:')
for node_id in ['node-A', 'node-B', 'node-C']:
versions = cluster.get(node_id, 'user:1')
values = [v.value for v in versions]
print(f' {node_id} sees: {values}')
"You should see all three nodes reporting the same value:
Replication test:
node-A sees: [{'name': 'Alice', 'role': 'engineer'}]
node-B sees: [{'name': 'Alice', 'role': 'engineer'}]
node-C sees: [{'name': 'Alice', 'role': 'engineer'}]Why does every node have the data?
You wrote to node-A as the coordinator. The cluster looked up the preference list for the key user:1 and found all three nodes. It then called receive_replica on node-B and node-C. This is the same pattern that Cassandra and DynamoDB use to replicate writes across their clusters.
Seeing an error instead of output?
Make sure your terminal is in the distributed-kv-store/ directory. Python needs to find all four module files in the current folder.
If you see a SyntaxError, check that you copied the entire test command including the closing quote. Multi-line commands in the terminal require the quotes to be balanced.
help me debug this replication test
Your cluster is replicating data across all three nodes. Right now the vector clock stub treats every write as equal, so there is no conflict detection. In the next step, you will implement real vector clocks that track causality and catch conflicting writes.
Add Vector Clocks for Conflict Detection
Your cluster can replicate data, but it has a hidden problem. If two nodes write to the same key at the same time, one write silently overwrites the other.
Vector clocks solve this by giving each write a logical timestamp that tracks causality. When two clocks evolved independently, the system knows they're "concurrent" and preserves both versions as siblings instead of losing data.
In this step, get ready to:
- Implement the VectorClock class with increment, merge, and compare operations.
- See how vector clocks integrate with KVNode to version every write.
- Test that concurrent writes produce siblings instead of silent overwrites.
Implement the increment and merge methods
Your vector_clock.py already has the class structure with stub methods. The increment method records that a node performed an event. The merge method combines two clocks by taking the element-wise maximum of every entry.
- In vector_clock.py, find the increment method. It currently contains just pass.
- Replace the entire method body (the line that says pass) with this implementation:
def increment(self, node_id: str) -> None:
"""Record a local event at the given node."""
self.clock[node_id] = self.clock.get(node_id, 0) + 1- Now find the merge method directly below. Replace its pass body with this implementation:
def merge(self, other: "VectorClock") -> None:
"""Merge another vector clock (element-wise max)."""
for node_id, count in other.clock.items():
self.clock[node_id] = max(self.clock.get(node_id, 0), count)- Finally, find the copy method and add the docstring """Create a deep copy of this vector clock.""" as its first line inside the method body.
What does this code do?
increment bumps the counter for a specific node by 1. If the node has never been seen before, it starts at 0 and becomes 1.
merge takes the element-wise maximum of two clocks. This is how a node "absorbs" the knowledge of another node's events. After merging, the clock reflects everything both nodes have seen.
- Save vector_clock.py.
- Verify the increment method works by running this command:
python3 -c "from vector_clock import VectorClock; vc = VectorClock(); vc.increment('node-A'); vc.increment('node-A'); vc.increment('node-B'); print(vc)"You should see VectorClock({'node-A': 2, 'node-B': 1}) printed in your terminal.
Seeing an error or empty clock?
- Make sure you saved vector_clock.py after editing.
- Check that the increment body is indented inside the method (8 spaces or 2 tabs from the left margin).
- If you see VectorClock({}), the pass statement is still there. Remove it and paste the new code.
Help me debug my increment method
Implement the compare method
The compare method is the core of conflict detection. It examines two vector clocks element by element to determine their causal relationship.
- In vector_clock.py, find the compare static method. It currently just returns ClockOrder.EQUAL.
- Replace the entire method body with this implementation:
@staticmethod
def compare(a: "VectorClock", b: "VectorClock") -> ClockOrder:
"""Compare two vector clocks to determine causal ordering."""
all_keys = set(a.clock.keys()) | set(b.clock.keys())
a_less = False
b_less = False
for key in all_keys:
a_val = a.clock.get(key, 0)
b_val = b.clock.get(key, 0)
if a_val < b_val:
a_less = True
elif a_val > b_val:
b_less = True
if a_less and b_less:
return ClockOrder.CONCURRENT
elif a_less:
return ClockOrder.BEFORE
elif b_less:
return ClockOrder.AFTER
else:
return ClockOrder.EQUALWhat does this code do?
The method collects all node IDs from both clocks, then compares each entry. If clock A has any value less than B, it sets a_less = True. If B has any value less than A, it sets b_less = True.
The final decision logic: if both flags are true, neither clock dominates the other. That means CONCURRENT (conflict detected). If only a_less is true, A happened BEFORE B. If only b_less is true, A happened AFTER B.
Why vector clocks over wall-clock timestamps?
Wall clocks on different machines are never perfectly synchronized. A write at 10:00:01 on one machine might actually have happened before a write at 10:00:00 on another. Vector clocks track causality without needing any clock synchronization between nodes.
- Save vector_clock.py.
- Test that compare detects concurrent writes by running this command:
python3 -c "from vector_clock import VectorClock, ClockOrder; a = VectorClock(); a.increment('node-A'); b = VectorClock(); b.increment('node-B'); print(VectorClock.compare(a, b))"You should see ClockOrder.CONCURRENT printed. Two independent writes with no shared history are concurrent.
Still seeing ClockOrder.EQUAL?
- Make sure you replaced the old return ClockOrder.EQUAL line with the full implementation. The old stub always returned EQUAL.
- Check that the @staticmethod decorator is still above the method definition.
- Confirm you saved the file after editing.
Help me debug my compare method
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
from enum import Enum
from typing import Dict
class ClockOrder(Enum):
BEFORE = "before"
AFTER = "after"
CONCURRENT = "concurrent"
EQUAL = "equal"
class VectorClock:
"""Vector clock for tracking causal ordering of events."""
def __init__(self):
self.clock: Dict[str, int] = {}
def increment(self, node_id: str) -> None:
"""Record a local event at the given node."""
self.clock[node_id] = self.clock.get(node_id, 0) + 1
def merge(self, other: "VectorClock") -> None:
"""Merge another vector clock (element-wise max)."""
for node_id, count in other.clock.items():
self.clock[node_id] = max(self.clock.get(node_id, 0), count)
def copy(self) -> "VectorClock":
"""Create a deep copy of this vector clock."""
new_clock = VectorClock()
new_clock.clock = dict(self.clock)
return new_clock
@staticmethod
def compare(a: "VectorClock", b: "VectorClock") -> ClockOrder:
"""Compare two vector clocks to determine causal ordering."""
all_keys = set(a.clock.keys()) | set(b.clock.keys())
a_less = False
b_less = False
for key in all_keys:
a_val = a.clock.get(key, 0)
b_val = b.clock.get(key, 0)
if a_val < b_val:
a_less = True
elif a_val > b_val:
b_less = True
if a_less and b_less:
return ClockOrder.CONCURRENT
elif a_less:
return ClockOrder.BEFORE
elif b_less:
return ClockOrder.AFTER
else:
return ClockOrder.EQUAL
def __repr__(self) -> str:
return f"VectorClock({self.clock})"See how vector clocks integrate with KVNode
Your node.py already uses vector clocks everywhere. Now that the stub methods are real implementations, the integration comes alive automatically.
How does KVNode use vector clocks?
- put() calls clock.increment(self.node_id) on every write. This stamps the write with the node's identity.
- put() also accepts an optional context clock from a previous read. This enables read-modify-write patterns where the new clock builds on the old one.
- receive_replica() calls VectorClock.compare() to decide what to do with incoming data. If the incoming clock is CONCURRENT with an existing value, both are kept as siblings.
You do not need to change node.py. The integration was built in the previous step. The stubs just meant that every comparison returned EQUAL (treating all writes as duplicates).
Test conflict detection across nodes
The real test of your vector clocks is whether the cluster correctly preserves conflicting writes as siblings. Two nodes writing to the same key with the same context clock should produce a conflict, not a silent overwrite.
- Run this end-to-end conflict test:
python3 -c "
from cluster import Cluster
# Create a 3-node cluster
cluster = Cluster(['node-A', 'node-B', 'node-C'])
# Initial write via node-A
clock = cluster.put('node-A', 'user:1', 'original')
print(f'Initial write clock: {clock}')
# Two concurrent updates using the SAME context clock
# This simulates two clients who both read 'original' and then write independently
cluster.put('node-A', 'user:1', 'update-from-A', clock)
cluster.put('node-B', 'user:1', 'update-from-B', clock)
# Check what node-A sees
versions = cluster.get('node-A', 'user:1')
print(f'\nNode-A sees {len(versions)} version(s):')
for v in versions:
print(f' value={v.value!r}, clock={v.clock}')
print(f'\nConflict detected: {len(versions) > 1}')
"What should I see?
You should see 2 versions on node-A. One has the value 'update-from-A' with clock {'node-A': 2}. The other has 'update-from-B' with clock {'node-A': 1, 'node-B': 1}.
These two clocks are concurrent. Node-A has a higher value in the 'node-A' entry, but a lower value in the 'node-B' entry. Neither dominates. The system preserves both writes and lets the application layer decide how to resolve.
Only seeing 1 version?
- If you see only update-from-B, your compare method might still be returning EQUAL. Double check that you replaced the entire old method body.
- If you see only update-from-A, the replication from node-B to node-A might not be reaching. Confirm cluster.py has no partitions set up (it shouldn't by default).
- Run the single-clock test from the previous substep again (ClockOrder.CONCURRENT check) to confirm compare works in isolation.
Help me debug why my cluster isn't detecting conflicts
Your KV store now detects conflicts instead of silently losing data. In an interview, this is the key insight behind eventual consistency systems like DynamoDB and Cassandra. Next up, you'll simulate network partitions and build a gossip protocol that automatically converges diverged state.
Simulate Partitions and Converge with Gossip
Your cluster can replicate data across nodes, and your vector clocks detect conflicts. But what happens when nodes lose the ability to talk to each other?
In this step, you will simulate network partitions that split your cluster, observe how data diverges on each side, and then implement an anti-entropy gossip protocol that converges all nodes back to a consistent state after the partition heals.
In this step, get ready to:
- Add partition and heal methods that block communication between node pairs.
- Implement anti-entropy gossip that synchronizes diverged state across reachable nodes.
- Build and run a full demo that shows divergence during a partition and convergence after healing.
Add partition and heal methods
A network partition occurs when some nodes can no longer reach each other. Your Cluster class already has a partitions set and a _can_communicate method that checks it. Now you need methods to actually create and remove partitions.
- In cluster.py, find the get method at the bottom of your Cluster class.
- Add the following two methods below it:
def partition(self, node_a: str, node_b: str) -> None:
"""Simulate a network partition between two nodes."""
pair = tuple(sorted([node_a, node_b]))
self.partitions.add(pair)
print(f" [PARTITION] {node_a} <-/-> {node_b}")
def heal_partition(self, node_a: str, node_b: str) -> None:
"""Heal a network partition between two nodes."""
pair = tuple(sorted([node_a, node_b]))
self.partitions.discard(pair)
print(f" [HEALED] {node_a} <--> {node_b}")What does this code do?
- partition adds a sorted tuple of two node IDs to the partitions set. The existing _can_communicate method already checks this set before allowing replication.
- heal_partition removes the pair from the set, restoring communication. Sorting ensures that partition("A", "B") and partition("B", "A") produce the same result.
You won't see a change yet. The next chunk uses these methods to build the gossip protocol.
Implement anti-entropy gossip
After a partition heals, nodes still hold diverged data. Anti-entropy gossip is the process of two nodes comparing their state and exchanging any versions the other is missing.
- Add the anti_entropy method below heal_partition in cluster.py:
def anti_entropy(self, node_a: str, node_b: str) -> int:
"""Run anti-entropy sync between two nodes (gossip protocol)."""
if not self._can_communicate(node_a, node_b):
return 0
synced = 0
# Sync from A to B
for key, versions in list(self.nodes[node_a].get_all_data().items()):
for version in versions:
b_versions = self.nodes[node_b].get(key)
should_sync = True
for bv in b_versions:
order = VectorClock.compare(version.clock, bv.clock)
if order in (ClockOrder.BEFORE, ClockOrder.EQUAL):
should_sync = False
break
if should_sync:
self.nodes[node_b].receive_replica(
key=key,
value=version.value,
clock=version.clock,
timestamp=version.timestamp,
)
synced += 1
# Sync from B to A
for key, versions in list(self.nodes[node_b].get_all_data().items()):
for version in versions:
a_versions = self.nodes[node_a].get(key)
should_sync = True
for av in a_versions:
order = VectorClock.compare(version.clock, av.clock)
if order in (ClockOrder.BEFORE, ClockOrder.EQUAL):
should_sync = False
break
if should_sync:
self.nodes[node_a].receive_replica(
key=key,
value=version.value,
clock=version.clock,
timestamp=version.timestamp,
)
synced += 1
return syncedWhat does this code do?
This is a bidirectional sync. For each key-version pair on node A, it checks whether node B already has that version (or a newer one). If not, it sends the version via receive_replica. Then it repeats in the opposite direction.
The vector clock comparison determines whether to sync. If A's version is BEFORE or EQUAL to something B already has, there is nothing new to send. Otherwise, receive_replica handles the merge logic (keeping concurrent versions as siblings).
- Add the full_sync convenience method below anti_entropy:
def full_sync(self) -> int:
"""Run anti-entropy between all reachable node pairs."""
total_synced = 0
node_ids = list(self.nodes.keys())
for i in range(len(node_ids)):
for j in range(i + 1, len(node_ids)):
total_synced += self.anti_entropy(node_ids[i], node_ids[j])
return total_syncedWhat does this code do?
full_sync runs anti-entropy between every unique pair of nodes in the cluster. If a pair is still partitioned, anti_entropy returns 0 for that pair and moves on.
- Save cluster.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved the file before moving on.
ⓧ I'd like to double check the full code
from typing import Any, Dict, List, Optional, Set, Tuple
from hash_ring import HashRing
from node import KVNode, VersionedValue
from vector_clock import VectorClock, ClockOrder
class Cluster:
"""Manages a cluster of KV nodes with replication and partition simulation."""
def __init__(self, node_ids: List[str], replication_factor: int = 3, virtual_nodes: int = 150):
self.replication_factor = replication_factor
self.ring = HashRing(virtual_nodes=virtual_nodes)
self.nodes: Dict[str, KVNode] = {}
self.partitions: Set[Tuple[str, str]] = set()
for node_id in node_ids:
self.nodes[node_id] = KVNode(node_id)
self.ring.add_node(node_id)
def _can_communicate(self, node_a: str, node_b: str) -> bool:
"""Check if two nodes can communicate (no partition between them)."""
pair = tuple(sorted([node_a, node_b]))
return pair not in self.partitions
def put(self, coordinator: str, key: str, value: Any, context: Optional[VectorClock] = None) -> VectorClock:
if coordinator not in self.nodes:
raise ValueError(f"Unknown node: {coordinator}")
clock = self.nodes[coordinator].put(key, value, context)
targets = self.ring.get_preference_list(key, self.replication_factor)
for target_id in targets:
if target_id == coordinator:
continue
if self._can_communicate(coordinator, target_id):
self.nodes[target_id].receive_replica(
key=key,
value=value,
clock=clock,
timestamp=self.nodes[coordinator].replication_log[-1]["timestamp"],
)
return clock
def get(self, coordinator: str, key: str) -> List[VersionedValue]:
"""Read a key from the coordinator node."""
if coordinator not in self.nodes:
raise ValueError(f"Unknown node: {coordinator}")
return self.nodes[coordinator].get(key)
def partition(self, node_a: str, node_b: str) -> None:
"""Simulate a network partition between two nodes."""
pair = tuple(sorted([node_a, node_b]))
self.partitions.add(pair)
print(f" [PARTITION] {node_a} <-/-> {node_b}")
def heal_partition(self, node_a: str, node_b: str) -> None:
"""Heal a network partition between two nodes."""
pair = tuple(sorted([node_a, node_b]))
self.partitions.discard(pair)
print(f" [HEALED] {node_a} <--> {node_b}")
def anti_entropy(self, node_a: str, node_b: str) -> int:
"""Run anti-entropy sync between two nodes (gossip protocol)."""
if not self._can_communicate(node_a, node_b):
return 0
synced = 0
# Sync from A to B
for key, versions in list(self.nodes[node_a].get_all_data().items()):
for version in versions:
b_versions = self.nodes[node_b].get(key)
should_sync = True
for bv in b_versions:
order = VectorClock.compare(version.clock, bv.clock)
if order in (ClockOrder.BEFORE, ClockOrder.EQUAL):
should_sync = False
break
if should_sync:
self.nodes[node_b].receive_replica(
key=key,
value=version.value,
clock=version.clock,
timestamp=version.timestamp,
)
synced += 1
# Sync from B to A
for key, versions in list(self.nodes[node_b].get_all_data().items()):
for version in versions:
a_versions = self.nodes[node_a].get(key)
should_sync = True
for av in a_versions:
order = VectorClock.compare(version.clock, av.clock)
if order in (ClockOrder.BEFORE, ClockOrder.EQUAL):
should_sync = False
break
if should_sync:
self.nodes[node_a].receive_replica(
key=key,
value=version.value,
clock=version.clock,
timestamp=version.timestamp,
)
synced += 1
return synced
def full_sync(self) -> int:
"""Run anti-entropy between all reachable node pairs."""
total_synced = 0
node_ids = list(self.nodes.keys())
for i in range(len(node_ids)):
for j in range(i + 1, len(node_ids)):
total_synced += self.anti_entropy(node_ids[i], node_ids[j])
return total_syncedBuild the demo script
Now you have all the building blocks: consistent hashing, vector clocks, replication, partitions, and gossip. The demo script ties them together into a single runnable demonstration.
- Open demo.py in your editor.
- Add the module docstring, imports, and the first two demo functions by pasting the following code:
"""
Distributed Key-Value Store Demo
================================
Demonstrates consistent hashing, replication, vector clocks,
network partitions, and anti-entropy gossip.
"""
from hash_ring import HashRing
from vector_clock import VectorClock, ClockOrder
from cluster import Cluster
def demo_consistent_hashing():
"""Demonstrate consistent hashing with virtual nodes."""
print("\n" + "=" * 60)
print("CONSISTENT HASHING DEMO")
print("=" * 60)
ring = HashRing(virtual_nodes=150)
for node in ["node-A", "node-B", "node-C"]:
ring.add_node(node)
# Show key distribution
sample_keys = [f"user:{i}" for i in range(1000)]
distribution = ring.get_distribution(sample_keys)
print("\nKey distribution across 3 nodes (1000 keys):")
for node, count in sorted(distribution.items()):
print(f" {node}: {count} keys ({count / 10:.1f}%)")
# Show preference lists for replication
print("\nPreference lists (replication targets):")
for key in ["user:42", "session:abc", "order:999"]:
targets = ring.get_preference_list(key, count=3)
print(f" {key} -> {targets}")
# Demonstrate minimal disruption on node removal
print("\nRemoving node-B from the ring...")
keys_before = {key: ring.get_node(key) for key in sample_keys}
ring.remove_node("node-B")
keys_after = {key: ring.get_node(key) for key in sample_keys}
moved = sum(1 for k in sample_keys if keys_before[k] != keys_after[k])
print(f"Keys that moved: {moved}/1000 ({moved / 10:.1f}%)")
print("(Only keys previously on node-B needed to move)")
new_distribution = ring.get_distribution(sample_keys)
print("\nNew distribution after removal:")
for node, count in sorted(new_distribution.items()):
print(f" {node}: {count} keys ({count / 10:.1f}%)")
def demo_vector_clocks():
"""Demonstrate vector clock causality tracking."""
print("\n" + "=" * 60)
print("VECTOR CLOCK DEMO")
print("=" * 60)
# Causal ordering
print("\n--- Causal chain ---")
a = VectorClock()
a.increment("node-A")
print(f"After node-A writes: {a}")
b = a.copy()
b.increment("node-B")
print(f"After node-B writes (saw A): {b}")
order = VectorClock.compare(a, b)
print(f"A vs B: {order.value}")
# Concurrent writes (conflict!)
print("\n--- Concurrent writes (conflict!) ---")
c1 = VectorClock()
c1.increment("node-A")
c2 = VectorClock()
c2.increment("node-B")
print(f"node-A writes independently: {c1}")
print(f"node-B writes independently: {c2}")
order = VectorClock.compare(c1, c2)
print(f"Comparison: {order.value} <-- CONFLICT DETECTED")
# Merge resolves
print("\n--- Merge after conflict resolution ---")
merged = c1.copy()
merged.merge(c2)
merged.increment("node-A")
print(f"After node-A resolves conflict: {merged}")
print(f"Merged vs c1: {VectorClock.compare(c1, merged).value}")
print(f"Merged vs c2: {VectorClock.compare(c2, merged).value}")What do these functions do?
- demo_consistent_hashing() creates a 3-node ring, shows how evenly 1000 keys distribute, then removes a node to prove only ~1/3 of keys move.
- demo_vector_clocks() shows a causal chain (A before B), concurrent writes that produce a conflict, and a merge that resolves both.
- Add the partition demo and the main entry point below the code you just pasted:
def demo_replication_and_partitions():
"""Demonstrate replication, partitions, and convergence."""
print("\n" + "=" * 60)
print("REPLICATION & PARTITION DEMO")
print("=" * 60)
cluster = Cluster(
node_ids=["node-A", "node-B", "node-C"],
replication_factor=3,
)
# Normal write with replication
print("\n--- Normal write (all nodes connected) ---")
clock = cluster.put("node-A", "user:1", {"name": "Alice", "role": "engineer"})
print(f"Wrote user:1 via node-A, clock: {clock}")
# Verify replication
for node_id in ["node-A", "node-B", "node-C"]:
versions = cluster.get(node_id, "user:1")
values = [v.value for v in versions]
print(f" {node_id} sees: {values}")
# Create a network partition
print("\n--- Creating partition: node-C isolated ---")
cluster.partition("node-A", "node-C")
cluster.partition("node-B", "node-C")
# Write during partition (only reaches A and B)
clock_ab = cluster.put(
"node-A", "user:1",
{"name": "Alice", "role": "senior engineer"},
clock,
)
print(f"Updated user:1 via node-A (partitioned from C)")
print(f" New clock: {clock_ab}")
# Concurrent write on isolated node
clock_c = cluster.put(
"node-C", "user:1",
{"name": "Alice", "role": "tech lead"},
clock,
)
print(f"Updated user:1 via node-C (isolated)")
print(f" New clock: {clock_c}")
# Show divergence
print("\n--- Diverged state (partition active) ---")
for node_id in ["node-A", "node-B", "node-C"]:
versions = cluster.get(node_id, "user:1")
values = [v.value for v in versions]
print(f" {node_id} sees: {values}")
# Heal partition and run anti-entropy
print("\n--- Healing partition and running anti-entropy gossip ---")
cluster.heal_partition("node-A", "node-C")
cluster.heal_partition("node-B", "node-C")
synced = cluster.full_sync()
print(f" Synced {synced} version(s) across nodes")
# Show convergence (siblings from concurrent writes)
print("\n--- Converged state (all nodes see siblings) ---")
for node_id in ["node-A", "node-B", "node-C"]:
versions = cluster.get(node_id, "user:1")
print(f" {node_id} sees {len(versions)} version(s):")
for v in versions:
print(f" value={v.value!r}")
print(f" clock={v.clock}")
print("\n Both concurrent writes are preserved as siblings.")
print(" The application layer decides how to resolve (LWW, merge, ask user).")
def main():
print("=" * 60)
print("DISTRIBUTED KEY-VALUE STORE")
print("=" * 60)
print("Concepts: Consistent Hashing | Replication | Vector Clocks")
print(" Network Partitions | Anti-Entropy Gossip")
demo_consistent_hashing()
demo_vector_clocks()
demo_replication_and_partitions()
print("\n" + "=" * 60)
print("SUCCESS! You have a working distributed KV store.")
print("=" * 60)
print("\nInterview talking points:")
print(" - Why consistent hashing over modulo? (minimal data movement)")
print(" - Why vector clocks over timestamps? (no clock sync needed)")
print(" - What happens during a partition? (diverge, then converge)")
print(" - How do you tune consistency? (quorum: W + R > N)")
if __name__ == "__main__":
main()What does this code do?
- demo_replication_and_partitions() runs the full scenario: write with replication, isolate node-C, write conflicting updates on both sides, show diverged state, heal the partition, run gossip, and show converged siblings.
- main() runs all three demos in sequence and prints interview talking points at the end.
- The if __name__ == "__main__" guard ensures main() only runs when you execute the file directly, not when importing.
- Save demo.py.
- Run the full demo by executing this command:
python demo.py- You should see three sections of output: CONSISTENT HASHING DEMO, VECTOR CLOCK DEMO, and REPLICATION & PARTITION DEMO.
- In the partition demo, look for the diverged state showing node-A and node-B with one value while node-C has a different value.
- After healing, all three nodes should show 2 versions (siblings) for the same key.
Why are siblings preserved instead of picking a winner?
When two writes happen concurrently (neither caused the other), the system cannot safely discard either one. Both are valid from the perspective of their respective nodes.
Real systems like DynamoDB and Cassandra resolve this at the application layer. Common strategies include last-write-wins (LWW using timestamps), semantic merge (combine the values), or asking the user to choose.
Seeing a ModuleNotFoundError?
- Make sure you are running the command from inside the distributed-kv-store folder where all five .py files live.
- Check that hash_ring.py, vector_clock.py, node.py, and cluster.py are all saved in the same directory as demo.py.
- Verify you do not have a typo in the import names at the top of demo.py.
Help me debug my ModuleNotFoundError when running demo.py
✔️ Awesome, I've got everything!
Great. Make sure you have saved both files before moving on.
ⓧ I'd like to double check the full code
Here is the complete demo.py file:
"""
Distributed Key-Value Store Demo
================================
Demonstrates consistent hashing, replication, vector clocks,
network partitions, and anti-entropy gossip.
"""
from hash_ring import HashRing
from vector_clock import VectorClock, ClockOrder
from cluster import Cluster
def demo_consistent_hashing():
"""Demonstrate consistent hashing with virtual nodes."""
print("\n" + "=" * 60)
print("CONSISTENT HASHING DEMO")
print("=" * 60)
ring = HashRing(virtual_nodes=150)
for node in ["node-A", "node-B", "node-C"]:
ring.add_node(node)
sample_keys = [f"user:{i}" for i in range(1000)]
distribution = ring.get_distribution(sample_keys)
print("\nKey distribution across 3 nodes (1000 keys):")
for node, count in sorted(distribution.items()):
print(f" {node}: {count} keys ({count / 10:.1f}%)")
print("\nPreference lists (replication targets):")
for key in ["user:42", "session:abc", "order:999"]:
targets = ring.get_preference_list(key, count=3)
print(f" {key} -> {targets}")
print("\nRemoving node-B from the ring...")
keys_before = {key: ring.get_node(key) for key in sample_keys}
ring.remove_node("node-B")
keys_after = {key: ring.get_node(key) for key in sample_keys}
moved = sum(1 for k in sample_keys if keys_before[k] != keys_after[k])
print(f"Keys that moved: {moved}/1000 ({moved / 10:.1f}%)")
print("(Only keys previously on node-B needed to move)")
new_distribution = ring.get_distribution(sample_keys)
print("\nNew distribution after removal:")
for node, count in sorted(new_distribution.items()):
print(f" {node}: {count} keys ({count / 10:.1f}%)")
def demo_vector_clocks():
"""Demonstrate vector clock causality tracking."""
print("\n" + "=" * 60)
print("VECTOR CLOCK DEMO")
print("=" * 60)
print("\n--- Causal chain ---")
a = VectorClock()
a.increment("node-A")
print(f"After node-A writes: {a}")
b = a.copy()
b.increment("node-B")
print(f"After node-B writes (saw A): {b}")
order = VectorClock.compare(a, b)
print(f"A vs B: {order.value}")
print("\n--- Concurrent writes (conflict!) ---")
c1 = VectorClock()
c1.increment("node-A")
c2 = VectorClock()
c2.increment("node-B")
print(f"node-A writes independently: {c1}")
print(f"node-B writes independently: {c2}")
order = VectorClock.compare(c1, c2)
print(f"Comparison: {order.value} <-- CONFLICT DETECTED")
print("\n--- Merge after conflict resolution ---")
merged = c1.copy()
merged.merge(c2)
merged.increment("node-A")
print(f"After node-A resolves conflict: {merged}")
print(f"Merged vs c1: {VectorClock.compare(c1, merged).value}")
print(f"Merged vs c2: {VectorClock.compare(c2, merged).value}")
def demo_replication_and_partitions():
"""Demonstrate replication, partitions, and convergence."""
print("\n" + "=" * 60)
print("REPLICATION & PARTITION DEMO")
print("=" * 60)
cluster = Cluster(
node_ids=["node-A", "node-B", "node-C"],
replication_factor=3,
)
print("\n--- Normal write (all nodes connected) ---")
clock = cluster.put("node-A", "user:1", {"name": "Alice", "role": "engineer"})
print(f"Wrote user:1 via node-A, clock: {clock}")
for node_id in ["node-A", "node-B", "node-C"]:
versions = cluster.get(node_id, "user:1")
values = [v.value for v in versions]
print(f" {node_id} sees: {values}")
print("\n--- Creating partition: node-C isolated ---")
cluster.partition("node-A", "node-C")
cluster.partition("node-B", "node-C")
clock_ab = cluster.put(
"node-A", "user:1",
{"name": "Alice", "role": "senior engineer"},
clock,
)
print(f"Updated user:1 via node-A (partitioned from C)")
print(f" New clock: {clock_ab}")
clock_c = cluster.put(
"node-C", "user:1",
{"name": "Alice", "role": "tech lead"},
clock,
)
print(f"Updated user:1 via node-C (isolated)")
print(f" New clock: {clock_c}")
print("\n--- Diverged state (partition active) ---")
for node_id in ["node-A", "node-B", "node-C"]:
versions = cluster.get(node_id, "user:1")
values = [v.value for v in versions]
print(f" {node_id} sees: {values}")
print("\n--- Healing partition and running anti-entropy gossip ---")
cluster.heal_partition("node-A", "node-C")
cluster.heal_partition("node-B", "node-C")
synced = cluster.full_sync()
print(f" Synced {synced} version(s) across nodes")
print("\n--- Converged state (all nodes see siblings) ---")
for node_id in ["node-A", "node-B", "node-C"]:
versions = cluster.get(node_id, "user:1")
print(f" {node_id} sees {len(versions)} version(s):")
for v in versions:
print(f" value={v.value!r}")
print(f" clock={v.clock}")
print("\n Both concurrent writes are preserved as siblings.")
print(" The application layer decides how to resolve (LWW, merge, ask user).")
def main():
print("=" * 60)
print("DISTRIBUTED KEY-VALUE STORE")
print("=" * 60)
print("Concepts: Consistent Hashing | Replication | Vector Clocks")
print(" Network Partitions | Anti-Entropy Gossip")
demo_consistent_hashing()
demo_vector_clocks()
demo_replication_and_partitions()
print("\n" + "=" * 60)
print("SUCCESS! You have a working distributed KV store.")
print("=" * 60)
print("\nInterview talking points:")
print(" - Why consistent hashing over modulo? (minimal data movement)")
print(" - Why vector clocks over timestamps? (no clock sync needed)")
print(" - What happens during a partition? (diverge, then converge)")
print(" - How do you tune consistency? (quorum: W + R > N)")
if __name__ == "__main__":
main()Your distributed key-value store is fully operational. It hashes keys to nodes, replicates writes, detects conflicts with vector clocks, simulates network failures, and converges via gossip. Next up, the Secret Mission challenges you to add quorum reads and writes for tunable consistency.
Secret mission
Implement Quorum Reads and Writes
Your distributed store currently sends writes to all nodes and reads from one. Real systems like DynamoDB let you tune the consistency vs. availability tradeoff with quorum parameters. Can you add configurable W and R values that guarantee read-your-writes consistency?
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally with zero external dependencies, so there are no ongoing costs or background processes to worry about. Decide whether to keep your files for future reference or delete them entirely.
Resources you used:
- Project directory on your Desktop containing all Python files.
- Python files: hash_ring.py, vector_clock.py, node.py, cluster.py, and demo.py.
Keep everything running
No action needed. Choose this if you want to keep experimenting with your distributed KV store, extend it further, or reference the code for interview prep.
- Your project folder stays on your Desktop.
- You can run python demo.py anytime to revisit the demonstrations.
Pause - I'll come back to this later
Since this is a local Python project with no running processes, there is nothing to pause. Your files sit on disk and use negligible storage.
- Leave the project folder where it is.
- Optionally move it to a different location if you want to keep your Desktop tidy.
Delete - I don't want to use this again
Remove all project files from your machine to start fresh.
- Delete the project folder from your Desktop by dragging it to the Trash (macOS) or Recycle Bin (Windows).
- Empty the Trash or Recycle Bin to permanently remove the files.
Nice Work!
Nice Work!
You just built a distributed key-value store from scratch in pure Python. This is the same architecture that powers Amazon DynamoDB, Apache Cassandra, and Riak.
You've learned how to:
- Build a consistent hash ring with 150 virtual nodes that distributes keys evenly and handles node additions or removals with minimal data movement.
- Implement vector clocks that track causality and detect conflicting concurrent writes without a central coordinator.
- Simulate network partitions and converge diverged state automatically using anti-entropy gossip.
- Add quorum reads and writes that let you tune the consistency vs. availability tradeoff with W + R > N.
Ready to quiz yourself?