Distributed Systems Tutorial: Learn DistSys from Scratch (2026)
After building and operating distributed systems that handle billions of requests daily, I have come to appreciate that distribution introduces a fundamentally new class of challenges: partial failure, clock skew, and network partitions. A distributed system is a collection of independent computers that appears to its users as a single coherent system. This tutorial covers the core concepts — the CAP theorem, consensus protocols, replication strategies, and remote procedure calls.
We will dissect how systems like etcd, Cassandra, and Google Spanner achieve consistency, availability, and fault tolerance. You will implement a Raft-like consensus protocol from scratch, explore quorum-based replication, and understand why building distributed systems is genuinely harder than building single-machine systems.
The CAP Theorem and the FLP Impossibility
Eric Brewer's CAP theorem states that a distributed data store can provide at most two of three guarantees: Consistency (every read receives the most recent write), Availability (every request receives a non-error response), and Partition Tolerance (the system continues despite network failures). Since partitions are inevitable, the practical choice is between CP (sacrifice availability) and AP (sacrifice consistency). FLP proves that in asynchronous systems with crash failures, no consensus protocol can guarantee both safety and liveness.
class DistributedKVStore:
def __init__(self, nodes, mode='CP'):
self.nodes = nodes; self.data = {n:{} for n in nodes}; self.mode = mode
def write(self, key, value):
if self.mode == 'CP':
for node in self.nodes: self.data[node][key] = value
return True
elif self.mode == 'AP': self.data[self.nodes[0]][key] = value; return True
def read(self, key, node_idx=0):
if self.mode == 'CP': return self.data[node_idx].get(key, None)
elif self.mode == 'AP': return self.data[node_idx].get(key, None)RPC Frameworks and Serialization
Remote Procedure Call (RPC) is the backbone of inter-service communication in distributed systems. gRPC uses Protocol Buffers for serialization and HTTP/2 for transport, supporting bidirectional streaming. Thrift and Avro are alternative serialization frameworks. Idempotency and retry logic are essential design considerations — at-most-once, at-least-once, and exactly-once semantics must be explicitly chosen.
syntax = "proto3";
service KeyValueStore {
rpc Get(GetRequest) returns (GetResponse);
rpc Put(PutRequest) returns (PutResponse);
}
message GetRequest { string key = 1; }
message GetResponse { string value = 1; bool found = 2; }
message PutRequest { string key = 1; string value = 2; }
message PutResponse { bool success = 1; }Consensus: Paxos and Raft
Consensus algorithms allow a group of nodes to agree on a single value despite failures. Paxos was the first practical consensus protocol but is notoriously difficult to implement correctly. Raft was designed as a more understandable alternative, decomposing consensus into leader election, log replication, and safety. In Raft, nodes are in one of three states: leader, follower, or candidate. A majority (quorum) of nodes must agree before a log entry is committed.
class RaftNode:
def __init__(self, node_id, all_nodes):
self.id = node_id; self.all_nodes = all_nodes; self.state = 'follower'
self.current_term = 0; self.voted_for = None; self.log = []; self.commit_index = 0
def request_vote(self, candidate_id, term):
if term < self.current_term: return False
if self.voted_for is None or self.voted_for == candidate_id:
self.voted_for = candidate_id; self.current_term = term; return True
return False
def append_entries(self, leader_id, term, entries, prev_log_index, leader_commit):
if term < self.current_term: return False
if prev_log_index >= 0 and prev_log_index < len(self.log):
self.log = self.log[:prev_log_index + 1] + entries; return True
return FalseReplication: Leader-Follower and Quorum
Replication maintains copies of data across multiple nodes for fault tolerance and read scalability. In leader-follower replication, one node handles writes and propagates changes to followers. Quorum replication requires a read quorum (R nodes) and a write quorum (W nodes) such that R + W > N (total nodes) to ensure read-write conflict detection.
class QuorumStore:
def __init__(self, nodes, R, W):
self.nodes = nodes; self.N = len(nodes); self.R = R; self.W = W
assert R + W > self.N
def write(self, key, value, version):
acks = 0
for node in self.nodes:
if node.put(key, value, version): acks += 1
return acks >= self.W
def read(self, key):
results = [node.get(key) for node in self.nodes if node.get(key)[0] is not None]
if len(results) >= self.R:
return max(results, key=lambda x: x[1])[0]
return NoneDistributed Transactions and Two-Phase Commit
Two-phase commit (2PC) ensures atomicity across multiple nodes: the coordinator sends a prepare request to all participants, and if all vote yes, sends a commit. If any votes no, the coordinator sends an abort. The blocking nature of 2PC makes it unsuitable for high-contention workloads. Three-phase commit (3PC) reduces blocking by adding a pre-commit phase.
class TwoPhaseCommit:
def __init__(self, coordinator, participants):
self.coordinator = coordinator; self.participants = participants
def execute(self, transaction):
votes = []
for p in self.participants:
try: p.prepare(transaction); votes.append(True)
except: votes.append(False)
if all(votes):
for p in self.participants: p.commit(transaction)
return True
else:
for p in self.participants: p.abort(transaction)
return FalseFailure Detection and Timeout-Based Liveness
Failure detectors are crucial because network failures make it impossible to distinguish a crashed node from a slow one. The eventually perfect failure detector guarantees that every crashed node is eventually suspected and no correct node is suspected after some time. Phi-Accrual failure detectors maintain a sliding window of inter-arrival times and compute a suspicion level rather than a binary decision.
import time, statistics, math
class PhiAccrualFailureDetector:
def __init__(self, threshold=8.0):
self.threshold = threshold; self.intervals = []; self.last_hb = None
def heartbeat(self):
now = time.time()*1000
if self.last_hb: self.intervals.append(now - self.last_hb)
self.last_hb = now
def phi(self):
if len(self.intervals) < 1: return 0.0
mean = statistics.mean(self.intervals)
std = max(statistics.stdev(self.intervals), 100)
x = (time.time()*1000 - self.last_hb - mean) / std
return -math.log10(1 - 0.5*math.erfc(-x/math.sqrt(2)))
def is_available(self): return self.phi() < self.thresholdFrequently Asked Questions
What is the difference between the CAP theorem and the FLP impossibility?
CAP limits distributed data stores to two of three properties (consistency, availability, partition tolerance). FLP proves that in asynchronous systems with crash failures, no consensus protocol can guarantee both safety and liveness.
Why is Raft considered easier to implement than Paxos?
Raft decomposes consensus into leader election, log replication, and safety with strong leadership. It uses randomized timeouts to reduce split votes and restricts log entries to flow only from leader to followers.
What is the difference between at-least-once and exactly-once semantics?
At-least-once guarantees the operation executes at least once (may retry on failure), possibly causing duplicates. Exactly-once guarantees no duplicates but requires idempotent operations or distributed transaction coordination.
How do distributed systems handle clock skew?
Hybrid logical clocks (HLC) combine physical time with a logical counter to provide causal ordering without requiring synchronized clocks. Google Spanner uses TrueTime (GPS + atomic clocks) with bounded clock uncertainty.
Originally published on Ayodhyyya. Last updated June 1, 2026.